139 lines
5.2 KiB
Python
139 lines
5.2 KiB
Python
"""
|
|
Notifications API - Web push subscription management and VAPID key endpoint
|
|
"""
|
|
|
|
import os
|
|
import uuid
|
|
|
|
import flask
|
|
import jwt
|
|
import core.auth as auth
|
|
import core.postgres as postgres
|
|
import core.notifications as notifications
|
|
|
|
|
|
def _get_user_uuid(token):
|
|
try:
|
|
payload = jwt.decode(token, os.getenv("JWT_SECRET"), algorithms=["HS256"])
|
|
return payload.get("sub")
|
|
except (jwt.ExpiredSignatureError, jwt.InvalidTokenError):
|
|
return None
|
|
|
|
|
|
def _auth(request):
|
|
"""Extract and verify token. Returns user_uuid or None."""
|
|
header = request.headers.get("Authorization", "")
|
|
if not header.startswith("Bearer "):
|
|
return None
|
|
token = header[7:]
|
|
user_uuid = _get_user_uuid(token)
|
|
if not user_uuid or not auth.verifyLoginToken(token, userUUID=user_uuid):
|
|
return None
|
|
return user_uuid
|
|
|
|
|
|
def register(app):
|
|
|
|
@app.route("/api/notifications/vapid-public-key", methods=["GET"])
|
|
def api_vapidPublicKey():
|
|
"""Return the VAPID public key for push subscription."""
|
|
key = os.environ.get("VAPID_PUBLIC_KEY")
|
|
if not key:
|
|
return flask.jsonify({"error": "VAPID not configured"}), 503
|
|
return flask.jsonify({"public_key": key}), 200
|
|
|
|
@app.route("/api/notifications/subscribe", methods=["POST"])
|
|
def api_pushSubscribe():
|
|
"""Register a push subscription. Body: {endpoint, keys: {p256dh, auth}}"""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
data = flask.request.get_json()
|
|
if not data:
|
|
return flask.jsonify({"error": "missing body"}), 400
|
|
|
|
endpoint = data.get("endpoint")
|
|
keys = data.get("keys", {})
|
|
p256dh = keys.get("p256dh")
|
|
auth_key = keys.get("auth")
|
|
|
|
if not endpoint or not p256dh or not auth_key:
|
|
return flask.jsonify({"error": "missing endpoint or keys"}), 400
|
|
|
|
# Upsert: remove existing subscription with same endpoint for this user
|
|
existing = postgres.select("push_subscriptions", where={
|
|
"user_uuid": user_uuid,
|
|
"endpoint": endpoint,
|
|
})
|
|
if existing:
|
|
postgres.delete("push_subscriptions", {"id": existing[0]["id"]})
|
|
|
|
row = {
|
|
"id": str(uuid.uuid4()),
|
|
"user_uuid": user_uuid,
|
|
"endpoint": endpoint,
|
|
"p256dh": p256dh,
|
|
"auth": auth_key,
|
|
}
|
|
postgres.insert("push_subscriptions", row)
|
|
# Ensure web_push_enabled is set in notifications settings
|
|
notifications.setNotificationSettings(user_uuid, {"web_push_enabled": True})
|
|
return flask.jsonify({"subscribed": True}), 201
|
|
|
|
@app.route("/api/notifications/settings", methods=["GET"])
|
|
def api_getNotificationSettings():
|
|
"""Return notification channel settings for the current user."""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
settings = notifications.getNotificationSettings(user_uuid)
|
|
if not settings:
|
|
return flask.jsonify({
|
|
"discord_webhook": "",
|
|
"discord_enabled": False,
|
|
"ntfy_topic": "",
|
|
"ntfy_enabled": False,
|
|
"web_push_enabled": False,
|
|
}), 200
|
|
return flask.jsonify({
|
|
"discord_webhook": settings.get("discord_webhook") or "",
|
|
"discord_enabled": bool(settings.get("discord_enabled")),
|
|
"ntfy_topic": settings.get("ntfy_topic") or "",
|
|
"ntfy_enabled": bool(settings.get("ntfy_enabled")),
|
|
"web_push_enabled": bool(settings.get("web_push_enabled")),
|
|
}), 200
|
|
|
|
@app.route("/api/notifications/settings", methods=["PUT"])
|
|
def api_updateNotificationSettings():
|
|
"""Update notification channel settings. Accepts partial updates."""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
data = flask.request.get_json()
|
|
if not data:
|
|
return flask.jsonify({"error": "missing body"}), 400
|
|
result = notifications.setNotificationSettings(user_uuid, data)
|
|
if not result:
|
|
return flask.jsonify({"error": "no valid fields provided"}), 400
|
|
return flask.jsonify({"updated": True}), 200
|
|
|
|
@app.route("/api/notifications/subscribe", methods=["DELETE"])
|
|
def api_pushUnsubscribe():
|
|
"""Remove a push subscription. Body: {endpoint}"""
|
|
user_uuid = _auth(flask.request)
|
|
if not user_uuid:
|
|
return flask.jsonify({"error": "unauthorized"}), 401
|
|
data = flask.request.get_json()
|
|
if not data or not data.get("endpoint"):
|
|
return flask.jsonify({"error": "missing endpoint"}), 400
|
|
|
|
postgres.delete("push_subscriptions", {
|
|
"user_uuid": user_uuid,
|
|
"endpoint": data["endpoint"],
|
|
})
|
|
# If no subscriptions remain, disable web push
|
|
remaining = postgres.select("push_subscriptions", where={"user_uuid": user_uuid})
|
|
if not remaining:
|
|
notifications.setNotificationSettings(user_uuid, {"web_push_enabled": False})
|
|
return flask.jsonify({"unsubscribed": True}), 200
|