diff --git a/app.py b/app.py index f5868da..5359acf 100644 --- a/app.py +++ b/app.py @@ -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(): diff --git a/webpush.py b/webpush.py index fbfa106..c46bf93 100644 --- a/webpush.py +++ b/webpush.py @@ -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.