add a password for the regenerate_vapid endpoint

This commit is contained in:
2023-12-15 16:55:12 -07:00
parent 381a7930d8
commit 35747dff26
2 changed files with 25 additions and 2 deletions

23
app.py
View File

@@ -1,3 +1,8 @@
"""
Environment variables:
- SQLALCHEMY_DATABASE_URI: path to sqlite file, starting with "sqlite:////"
- ADMIN_PASSWORD: password for admin user for sensitive endpoints
"""
from typing import Dict, Tuple, Union, Optional
from flask import Flask, request, jsonify, Response
@@ -14,10 +19,10 @@ import os
import threading
import time
app = Flask(__name__)
CONTACT_EMAIL = "mailto:info@timesafari.app"
app = Flask(__name__)
class WebPushService():
"""
This class provides services for sending web push notifications.
@@ -210,6 +215,7 @@ class WebPushService():
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.
@@ -219,6 +225,19 @@ class WebPushService():
- If there's an error during the operation, a JSON response with the error message is returned with a 500 status code.
"""
envPassword = os.getenv('ADMIN_PASSWORD')
auth = request.authorization
if (auth is None
or auth.username is None
or auth.username != 'admin'
or auth.password is None
or auth.password != envPassword):
return (
jsonify(error='Wrong password'),
401,
{'WWW-Authenticate': 'Basic realm="Login Required"'}
)
# Creating a context for the application to enable database operations
try:
with self.app.app_context():

View File

@@ -63,11 +63,15 @@ if len(sys.argv) > 2:
try:
#print(str(subscription_info))
#sys.exit(0)
result = webpush(subscription_info,
data,
vapid_private_key=private_key_base64,
vapid_claims={"sub": "mailto:matthew.raymer@gmail.com"})
print(f"Result from remote service: {result}")
# log the .reason from Apple
# https://developer.apple.com/documentation/usernotifications/sending_web_push_notifications_in_web_apps_and_browsers#3994594
except WebPushException as ex:
print(f"An error occurred: {ex}")
# Check if there is a response from the remote service.