trent_larson/py-push-server
4
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merge pull request 'Create a send-test endpoint for a client to send themselves a test' (#4) from test-message into master

Browse Source 
Reviewed-on: #4
This commit was merged in pull request #4.
This commit is contained in:
trentlarson
2023-12-24 10:49:23 -05:00
parent f15ea55ad7 7f6a696aae
commit e5d5bafeb3
3 changed files with 86 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -1 +1,2 @@
*~
.aider*

7
README.md
View File

@@ -32,7 +32,7 @@ sudo docker run -d -p 8900:3000 -v ~/py-push-server-db:/app/instance/data --name
On a production server for security (eg /web-push/generate_vapid): set an environment variable `ADMIN_PASSWORD` for permissions; one way is to add this to the `docker run` command: `-e ADMIN_PASSWORD=<anything secure>`
Finally, after it's started, generate a new VAPID by hitting the `regenerate_vapid` endpoint with a POST, eg. `curl -X POST localhost:8080/web-push/regenerate_vapid`
Finally, after it's started, generate a new VAPID by hitting the `regenerate-vapid` endpoint with a POST, eg. `curl -X POST localhost:8080/web-push/regenerate-vapid`
@@ -244,7 +244,10 @@ pip install gunicorn
source venv/bin/activate
gunicorn -b 0.0.0.0:3000 --log-level=debug --workers=1 app:app # 3 workers trigger 3 daily subscription runs
cp data/webpush.db.empty data/webpush.db
# 3 workers would trigger 3 daily subscription runs
gunicorn -b 0.0.0.0:3000 --log-level=debug --workers=1 app:app
```

95
app.py
View File

@@ -43,7 +43,7 @@ class WebPushService():
# Setting the application instance
self.app = app
self.app.add_url_rule('/web-push/regenerate_vapid', view_func=self.regenerate_vapid, methods=['POST'])
self.app.add_url_rule('/web-push/regenerate-vapid', view_func=self.regenerate_vapid, methods=['POST'])
# Setting the database URI for the application
db_uri = os.getenv('SQLALCHEMY_DATABASE_URI', 'sqlite:////app/instance/data/webpush.db')
@@ -102,7 +102,8 @@ class WebPushService():
db.session.commit()
except Exception as e:
print(f"Error generating VAPID keys: {e}")
print(f"Error generating VAPID keys: {str(e)}")
raise e
def _initialize(self) -> None:
@@ -135,7 +136,8 @@ class WebPushService():
- vapid_key (VAPIDKey): The VAPID key model instance containing the private key used for sending the notification.
Returns:
- bool: True if the push notification was sent successfully, False otherwise.
- request.Response https://requests.readthedocs.io/en/latest/api.html#requests.Response if the push notification was sent successfully
or False if there was an exception
Notes:
- The `webpush` function is used to send the notification.
@@ -144,14 +146,15 @@ class WebPushService():
# Sending the push notification using the webpush function
try:
webpush(
result = webpush(
subscription_info=subscription_info,
data=json.dumps(message),
vapid_private_key=vapid_key.private_key,
vapid_claims={"sub": CONTACT_EMAIL}
)
# "because sometimes that's what I had to do to make it work!" - Matthew
time.sleep(1)
return True
return {"success": True, "message": result.text, "result": result}
except WebPushException as ex:
now = datetime.datetime.now().isoformat()
@@ -172,7 +175,7 @@ class WebPushService():
else:
print("Error other than unsubscribed/expired.", ex.args[0], flush=True)
return False
return {"success": False, "message": ex.args[0]}
def _send_daily_notifications(self) -> None:
@@ -234,19 +237,19 @@ class WebPushService():
1. Deletes the current VAPID keys from the database.
2. Generates and stores new VAPID keys using the `_generate_and_save_vapid_keys` method.
URL: /web-push/regenerate_vapid
URL: /web-push/regenerate-vapid
Method: POST
Header: Authentication: Basic ...
Returns:
- Tuple[str, int]: A JSON response indicating the success or failure of the operation, along with the appropriate HTTP status code.
- tuple with "success" as True or False, and "message" message string
Notes:
- If the operation is successful, a JSON response with a success message is returned with a 200 status code.
- If there's an error during the operation, a JSON response with the error message is returned with a 500 status code.
"""
# This default can be invoked thus: curl -X POST -H "Authorization: Basic YWRtaW46YWRtaW4=" localhost:3000/web-push/regenerate_vapid
# This default can be invoked thus: curl -X POST -H "Authorization: Basic YWRtaW46YWRtaW4=" localhost:3000/web-push/regenerate-vapid
envPassword = os.getenv('ADMIN_PASSWORD', 'admin')
auth = request.authorization
if (auth is None
@@ -273,7 +276,7 @@ class WebPushService():
return jsonify(success=True, message="VAPID keys regenerated successfully"), 200
except Exception as e:
return jsonify(error=f'Error regenerating VAPID keys: {str(e)}'), 500
return jsonify(success=False, message=f'Error regenerating VAPID keys: {str(e)}'), 500
@staticmethod
@@ -317,7 +320,7 @@ class WebPushService():
Method: POST
Returns:
- Tuple[str, int]: A JSON response indicating the success or failure of the operation, along with the appropriate HTTP status code.
- A JSON response with "success" as True or False, "message" as a response message string, and potentially a "result" as a request.Response object from sending a test notification
Notes:
- If the operation is successful, a confirmation push notification is sent to the subscriber with a success message.
@@ -333,7 +336,7 @@ class WebPushService():
# Checking if the VAPID key is available
if not vapid_key:
return jsonify(success=False, error="No VAPID keys available"), 500
return jsonify(success=False, message="No VAPID keys available"), 500
# Creating a new Subscription instance with the provided data
subscription = Subscription(endpoint=content['endpoint'],
@@ -361,10 +364,10 @@ class WebPushService():
message = {"title": "Subscription Successful", "message": "Thank you for subscribing!"}
# Sending the confirmation push notification
success = WebPushService._send_push_notification(subscription_info, message, vapid_key)
result = WebPushService._send_push_notification(subscription_info, message, vapid_key)
# Returning the operation status
return jsonify(success=success)
return jsonify(success=result["status_code"]==201, message=result["text"])
@staticmethod
@@ -404,7 +407,69 @@ class WebPushService():
# If the subscription is not found, return an error message
else:
return jsonify(success=False, error="Subscription not found"), 404
return jsonify(success=False, message="Subscription not found"), 404
@staticmethod
@app.route('/web-push/send-test', methods=['POST'])
def send_test() -> Tuple[str, int]:
"""
Endpoint to send a test push notification to a specific client.
This method:
1. Retrieves the subscription information from the incoming request.
2. Looks up the subscription in the database.
3. Calls the _send_push_notification method to send a test push notification.
4. Returns the result of the _send_push_notification call.
URL: /web-push/send-test
Method: POST
Returns:
- A JSON response with the result of the _send_push_notification call.
Notes:
- The incoming request should contain the parameters "endpoint", "p256dh", and "auth".
"""
# Retrieving the subscription information from the incoming request
content = request.json
endpoint = content['endpoint']
p256dh = content['keys']['p256dh']
auth = content['keys']['auth']
# Looking up the subscription in the database
subscription = Subscription.query.filter_by(endpoint=endpoint, p256dh=p256dh, auth=auth).first()
# If the subscription is found, call the _send_push_notification method
if subscription:
subscription_info = {
"endpoint": subscription.endpoint,
"keys": {
"p256dh": subscription.p256dh,
"auth": subscription.auth
}
}
title = "Test Notification"
if "title" in content:
title = content['title']
message = "This is a test notification."
if "message" in content:
message = content['message']
vapid_key = VAPIDKey.query.filter_by(id=subscription.vapid_key_id).first()
result = WebPushService._send_push_notification(
subscription_info,
{"title": title, "message": message},
vapid_key
)
return jsonify(
success=result["result"].status_code==201,
message=result["result"].text,
)
else:
return jsonify({"success": False, "message": "Subscription not found"}), 404
web_push_service = WebPushService(app, "app")