diff --git a/devplacepy/database/schema.py b/devplacepy/database/schema.py index 0df180ed..13bfef34 100644 --- a/devplacepy/database/schema.py +++ b/devplacepy/database/schema.py @@ -134,7 +134,28 @@ def init_db(): ) _index(db, "notifications", "idx_notifications_user", ["user_uid"]) _index(db, "notifications", "idx_notifications_user_read", ["user_uid", "read"]) + push_registration = get_table("push_registration") + for column, example in ( + ("uid", ""), + ("user_uid", ""), + ("provider", "webpush"), + ("endpoint", ""), + ("key_auth", ""), + ("key_p256dh", ""), + ("token", ""), + ("created_at", ""), + ("deleted_at", ""), + ): + if not push_registration.has_column(column): + push_registration.create_column_by_example(column, example) _index(db, "push_registration", "idx_push_registration_user", ["user_uid"]) + _index(db, "push_registration", "idx_push_registration_provider", ["provider"]) + if "push_registration" in db.tables: + with db: + db.query( + "UPDATE push_registration SET provider = 'webpush' " + "WHERE provider IS NULL OR provider = ''" + ) _index(db, "sessions", "idx_sessions_token", ["session_token"]) projects = get_table("projects") for column, example in ( diff --git a/devplacepy/docs_api/groups/push.py b/devplacepy/docs_api/groups/push.py index 2dad1962..882a6340 100644 --- a/devplacepy/docs_api/groups/push.py +++ b/devplacepy/docs_api/groups/push.py @@ -8,8 +8,15 @@ GROUP = { "intro": """ # Web Push -Browser push notifications via the Web Push protocol. Fetch the public VAPID key, then -register a `PushSubscription` obtained from the browser's `PushManager`. +Push notifications are delivered by one or more providers. `webpush` is the default and +implements the Web Push protocol: fetch the public VAPID key, then register a +`PushSubscription` obtained from the browser's `PushManager`. `apns` delivers to an Apple +Push Notification service device token and is only offered when an administrator has +configured it. + +`GET /push.json` lists the providers that currently accept registrations. A registration +body without a `provider` field is a `webpush` registration, so existing clients need no +change. There is no server-side unsubscribe endpoint: unsubscription is handled entirely in the browser by calling `PushManager.unsubscribe()` on the subscription. The server stops delivering @@ -26,39 +33,60 @@ four ways to sign requests. method="GET", path="/push.json", title="Get the public key", - summary="Return the VAPID public key for subscribing.", + summary="Return the VAPID public key and the providers that accept registrations.", auth="public", - sample_response={"publicKey": "BASE64_VAPID_KEY"}, + sample_response={ + "publicKey": "BASE64_VAPID_KEY", + "providers": {"webpush": {"publicKey": "BASE64_VAPID_KEY"}}, + }, ), endpoint( id="push-register", method="POST", path="/push.json", title="Register a subscription", - summary="Register a browser push subscription. Sends a welcome notification.", + summary="Register a push subscription. Sends a welcome notification.", auth="user", encoding="json", interactive=False, params=[ + field( + "provider", + "json", + "string", + False, + "webpush", + "Provider to register with. Omit for webpush.", + ), field( "endpoint", "json", "string", - True, + False, "https://fcm.googleapis.com/...", - "Subscription endpoint URL.", + "Subscription endpoint URL. Required for webpush.", ), field( "keys", "json", "string", - True, + False, '{"p256dh":"...","auth":"..."}', - "Subscription keys object.", + "Subscription keys object. Required for webpush.", + ), + field( + "token", + "json", + "string", + False, + "a1b2c3...", + "Hexadecimal device token. Required for apns.", ), ], notes=[ - 'The body must be JSON: `{"endpoint": "...", "keys": {"p256dh": "...", "auth": "..."}}`.' + 'A webpush body is JSON: `{"endpoint": "...", "keys": {"p256dh": "...", "auth": "..."}}`.', + 'An APNs body is JSON: `{"provider": "apns", "token": "..."}`.', + "A provider that is unknown, disabled or unconfigured returns 400.", ], sample_response={"registered": True}, ), diff --git a/devplacepy/main.py b/devplacepy/main.py index 8cf846dd..f44256ad 100644 --- a/devplacepy/main.py +++ b/devplacepy/main.py @@ -115,6 +115,7 @@ from devplacepy.services.containers.service import ContainerService from devplacepy.services.xmlrpc import XmlrpcService from devplacepy.services.audit import AuditService from devplacepy.services.audit import record as audit +from devplacepy.services.push import PushService from devplacepy.services.telegram import TelegramService from devplacepy.services.telegram.outbox_service import TelegramOutboxService @@ -275,6 +276,7 @@ async def lifespan(app: FastAPI): service_manager.register(ContainerService()) service_manager.register(XmlrpcService()) service_manager.register(AuditService()) + service_manager.register(PushService()) service_manager.register(TelegramService()) service_manager.register(TelegramOutboxService()) if not os.environ.get("DEVPLACE_DISABLE_SERVICES"): diff --git a/devplacepy/push/__init__.py b/devplacepy/push/__init__.py new file mode 100644 index 00000000..81e12273 --- /dev/null +++ b/devplacepy/push/__init__.py @@ -0,0 +1,29 @@ +# retoor + +from devplacepy.push.delivery import notify_user +from devplacepy.push.providers.webpush import ( + browser_base64, + create_notification_authorization, + create_notification_info_with_payload, + ensure_certificates, + generate_pkcs8_private_key, + generate_private_key, + generate_public_key, + hkdf, + public_key_standard_b64, +) +from devplacepy.push.store import register + +__all__ = [ + "browser_base64", + "create_notification_authorization", + "create_notification_info_with_payload", + "ensure_certificates", + "generate_pkcs8_private_key", + "generate_private_key", + "generate_public_key", + "hkdf", + "notify_user", + "public_key_standard_b64", + "register", +] diff --git a/devplacepy/push/delivery.py b/devplacepy/push/delivery.py new file mode 100644 index 00000000..994ea67c --- /dev/null +++ b/devplacepy/push/delivery.py @@ -0,0 +1,83 @@ +# retoor + +import logging +from typing import Any + +from devplacepy import stealth +from devplacepy.database import get_int_setting +from devplacepy.push import providers, store + +logger = logging.getLogger(__name__) + +TIMEOUT_KEY = "push_delivery_timeout_seconds" +DEFAULT_TIMEOUT_SECONDS = 10 +MIN_TIMEOUT_SECONDS = 1 +MAX_TIMEOUT_SECONDS = 120 + + +def timeout_seconds() -> float: + seconds = get_int_setting(TIMEOUT_KEY, DEFAULT_TIMEOUT_SECONDS) + return float(min(max(seconds, MIN_TIMEOUT_SECONDS), MAX_TIMEOUT_SECONDS)) + + +def group_by_provider( + registrations: list[dict[str, Any]], +) -> dict[str, list[dict[str, Any]]]: + grouped: dict[str, list[dict[str, Any]]] = {} + for registration in registrations: + grouped.setdefault(store.provider_of(registration), []).append(registration) + return grouped + + +async def notify_user(user_uid: str, payload: dict[str, Any]) -> None: + registrations = store.active_for_user(user_uid) + if not registrations: + logger.debug("No active push subscriptions for user %s", user_uid) + return + + grouped = group_by_provider(registrations) + async with stealth.stealth_async_client(timeout=timeout_seconds()) as client: + for name, rows in grouped.items(): + provider = providers.PROVIDERS.get(name) + if provider is None: + logger.warning( + "Unknown push provider %s on %s subscriptions of user %s", + name, + len(rows), + user_uid, + ) + continue + if not providers.is_active(provider): + logger.debug( + "Push provider %s is not active; skipping %s subscriptions", + name, + len(rows), + ) + continue + try: + prepared = provider.prepare(payload) + except Exception as exc: + logger.error("Push provider %s could not build a payload: %s", name, exc) + continue + for registration in rows: + await _deliver_one(provider, client, registration, prepared, user_uid) + + +async def _deliver_one(provider, client, registration, prepared, user_uid) -> None: + try: + outcome = await provider.deliver(client, registration, prepared) + except Exception as exc: + logger.error("Push provider %s raised for %s: %s", provider.name, user_uid, exc) + return + if outcome.status == providers.ACCEPTED: + logger.debug("Push delivered to %s via %s", user_uid, provider.name) + return + if outcome.status == providers.DEAD: + try: + store.mark_dead(registration["id"]) + except Exception as exc: + logger.error("Could not soft-delete push subscription: %s", exc) + return + logger.warning( + "Push rejected by %s for %s: %s", provider.name, user_uid, outcome.detail + ) diff --git a/devplacepy/push/providers/__init__.py b/devplacepy/push/providers/__init__.py new file mode 100644 index 00000000..e8619e45 --- /dev/null +++ b/devplacepy/push/providers/__init__.py @@ -0,0 +1,76 @@ +# retoor + +import logging +from typing import Any + +from devplacepy.push.providers.apns import ApnsProvider +from devplacepy.push.providers.base import ( + ACCEPTED, + DEAD, + REJECTED, + Delivery, + PushProvider, +) +from devplacepy.push.providers.webpush import WebPushProvider + +logger = logging.getLogger(__name__) + +DEFAULT_PROVIDER = WebPushProvider.name + +PROVIDERS: dict[str, PushProvider] = { + provider.name: provider for provider in (WebPushProvider(), ApnsProvider()) +} + +__all__ = [ + "ACCEPTED", + "DEAD", + "DEFAULT_PROVIDER", + "Delivery", + "PROVIDERS", + "PushProvider", + "REJECTED", + "active", + "admin_fields", + "client_config", + "get", + "is_active", + "names", +] + + +def get(name: str | None) -> PushProvider | None: + if not isinstance(name, str): + name = "" + return PROVIDERS.get(name.strip().lower() or DEFAULT_PROVIDER) + + +def names() -> list[str]: + return list(PROVIDERS) + + +def active() -> list[PushProvider]: + return [provider for provider in PROVIDERS.values() if is_active(provider)] + + +def admin_fields() -> list: + return [field for provider in PROVIDERS.values() for field in provider.all_fields()] + + +def client_config() -> dict[str, Any]: + return {provider.name: _client_config(provider) for provider in active()} + + +def is_active(provider: PushProvider) -> bool: + try: + return provider.is_active() + except Exception as exc: + logger.error("Push provider %s failed its readiness check: %s", provider.name, exc) + return False + + +def _client_config(provider: PushProvider) -> dict[str, Any]: + try: + return provider.client_config() + except Exception as exc: + logger.error("Push provider %s failed to describe itself: %s", provider.name, exc) + return {} diff --git a/devplacepy/push/providers/apns.py b/devplacepy/push/providers/apns.py new file mode 100644 index 00000000..1445ad88 --- /dev/null +++ b/devplacepy/push/providers/apns.py @@ -0,0 +1,243 @@ +# retoor + +import hashlib +import json +import logging +import string +import time +from typing import Any + +import httpx +import jwt + +from devplacepy.config import SECONDS_PER_DAY +from devplacepy.database import get_setting +from devplacepy.push.providers.base import ( + ACCEPTED, + DEAD, + REJECTED, + Delivery, + PushProvider, +) +from devplacepy.services.base import ConfigField +from devplacepy.utils import DEFAULT_PUSH_URL, PUSH_ICON, generate_uid + +logger = logging.getLogger(__name__) + +TEAM_ID_KEY = "push_apns_team_id" +KEY_ID_KEY = "push_apns_key_id" +AUTH_KEY_KEY = "push_apns_auth_key" +TOPIC_KEY = "push_apns_topic" +ENVIRONMENT_KEY = "push_apns_environment" + +PROVIDER_LABEL = "Apple Push (APNs)" +DEFAULT_ENVIRONMENT = "production" +HOSTS = { + "production": "api.push.apple.com", + "sandbox": "api.sandbox.push.apple.com", +} +ENVIRONMENT_OPTIONS = [ + {"value": "production", "label": "Production"}, + {"value": "sandbox", "label": "Sandbox"}, +] + +TOKEN_REFRESH_SECONDS = 45 * 60 +TOKEN_MIN_LENGTH = 64 +TOKEN_MAX_LENGTH = 200 +THREAD_ID = "devplace-notification" +PUSH_TYPE = "alert" +PRIORITY = "10" +DEAD_REASONS = frozenset( + { + "BadDeviceToken", + "DeviceTokenNotForTopic", + "ExpiredToken", + "Unregistered", + "TopicDisallowed", + } +) + +_token_state: dict[str, Any] = {} + + +def _setting(key: str) -> str: + return get_setting(key, "").strip() + + +def _environment() -> str: + value = _setting(ENVIRONMENT_KEY) or DEFAULT_ENVIRONMENT + return value if value in HOSTS else DEFAULT_ENVIRONMENT + + +def host() -> str: + return HOSTS[_environment()] + + +def _fingerprint(team_id: str, key_id: str, auth_key: str) -> str: + return hashlib.sha256(f"{team_id}:{key_id}:{auth_key}".encode("utf-8")).hexdigest() + + +def provider_token(team_id: str, key_id: str, auth_key: str) -> str: + fingerprint = _fingerprint(team_id, key_id, auth_key) + issued_at = int(time.time()) + state = _token_state.get("current") + if ( + state + and state["fingerprint"] == fingerprint + and issued_at - state["issued_at"] < TOKEN_REFRESH_SECONDS + ): + if state["token"] is None: + raise ValueError(state["error"]) + return state["token"] + try: + token = jwt.encode( + {"iss": team_id, "iat": issued_at}, + auth_key, + algorithm="ES256", + headers={"kid": key_id}, + ) + except Exception as exc: + message = f"APNs auth key is not usable: {exc}" + _token_state["current"] = { + "token": None, + "error": message, + "issued_at": issued_at, + "fingerprint": fingerprint, + } + logger.error(message) + raise ValueError(message) from exc + _token_state["current"] = { + "token": token, + "error": "", + "issued_at": issued_at, + "fingerprint": fingerprint, + } + return token + + +def _reason(response: httpx.Response) -> str: + try: + body = response.json() + except ValueError: + return "" + if isinstance(body, dict): + return str(body.get("reason", "")) + return "" + + +class ApnsProvider(PushProvider): + name = "apns" + label = PROVIDER_LABEL + config_fields = [ + ConfigField( + TEAM_ID_KEY, + "Team ID", + type="str", + default="", + help="Ten character Apple Developer team identifier, used as the token iss claim.", + group=PROVIDER_LABEL, + ), + ConfigField( + KEY_ID_KEY, + "Key ID", + type="str", + default="", + help="Ten character identifier of the APNs auth key, sent as the token kid header.", + group=PROVIDER_LABEL, + ), + ConfigField( + AUTH_KEY_KEY, + "Auth key (.p8)", + type="text", + default="", + secret=True, + help="Contents of the APNs .p8 signing key, including the BEGIN and END lines. Leave blank to keep the stored key.", + group=PROVIDER_LABEL, + ), + ConfigField( + TOPIC_KEY, + "Topic", + type="str", + default="", + help="Bundle identifier of the receiving app, sent as the apns-topic header.", + group=PROVIDER_LABEL, + ), + ConfigField( + ENVIRONMENT_KEY, + "Environment", + type="select", + default=DEFAULT_ENVIRONMENT, + options=ENVIRONMENT_OPTIONS, + help="Production delivers to App Store builds, sandbox to development builds.", + group=PROVIDER_LABEL, + ), + ] + + def is_configured(self) -> bool: + return bool( + _setting(TEAM_ID_KEY) + and _setting(KEY_ID_KEY) + and _setting(AUTH_KEY_KEY) + and _setting(TOPIC_KEY) + ) + + def parse_registration(self, body: dict[str, Any]) -> dict[str, Any] | None: + token = body.get("token") + if not isinstance(token, str): + return None + token = token.strip() + if not TOKEN_MIN_LENGTH <= len(token) <= TOKEN_MAX_LENGTH: + return None + if any(character not in string.hexdigits for character in token): + return None + return {"token": token} + + def prepare(self, payload: dict[str, Any]) -> str: + return json.dumps( + { + "aps": { + "alert": { + "title": payload.get("title") or "DevPlace", + "body": payload.get("message") or "", + }, + "sound": "default", + "thread-id": THREAD_ID, + }, + "url": payload.get("url") or DEFAULT_PUSH_URL, + "icon": payload.get("icon") or PUSH_ICON, + } + ) + + def headers(self) -> dict[str, str]: + return { + "authorization": f"bearer {provider_token(_setting(TEAM_ID_KEY), _setting(KEY_ID_KEY), _setting(AUTH_KEY_KEY))}", + "apns-topic": _setting(TOPIC_KEY), + "apns-push-type": PUSH_TYPE, + "apns-priority": PRIORITY, + "apns-expiration": str(int(time.time()) + SECONDS_PER_DAY), + "apns-id": generate_uid(), + "content-type": "application/json", + } + + async def deliver( + self, client: httpx.AsyncClient, registration: dict[str, Any], prepared: str + ) -> Delivery: + token = (registration.get("token") or "").strip() + if not token: + return Delivery(DEAD, "missing device token") + try: + headers = self.headers() + response = await client.post( + f"https://{host()}/3/device/{token}", + headers=headers, + content=prepared.encode("utf-8"), + ) + except (httpx.HTTPError, ValueError) as exc: + return Delivery(REJECTED, str(exc)) + if response.status_code == 200: + return Delivery(ACCEPTED) + reason = _reason(response) + detail = f"{response.status_code} {reason}".strip() + if response.status_code == 410 or reason in DEAD_REASONS: + return Delivery(DEAD, detail) + return Delivery(REJECTED, detail) diff --git a/devplacepy/push/providers/base.py b/devplacepy/push/providers/base.py new file mode 100644 index 00000000..4e8d71fd --- /dev/null +++ b/devplacepy/push/providers/base.py @@ -0,0 +1,66 @@ +# retoor + +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import Any + +import httpx + +from devplacepy.database import get_setting +from devplacepy.services.base import ConfigField + +ACCEPTED = "accepted" +DEAD = "dead" +REJECTED = "rejected" + + +@dataclass(frozen=True) +class Delivery: + status: str + detail: str = "" + + +class PushProvider(ABC): + name = "" + label = "" + config_fields: list[ConfigField] = [] + + @property + def enabled_key(self) -> str: + return f"push_{self.name}_enabled" + + def enabled_field(self) -> ConfigField: + return ConfigField( + self.enabled_key, + "Enabled", + type="bool", + default=True, + help=f"Deliver notifications through {self.label}.", + group=self.label, + ) + + def all_fields(self) -> list[ConfigField]: + return [self.enabled_field(), *self.config_fields] + + def is_enabled(self) -> bool: + return get_setting(self.enabled_key, "1") == "1" + + def is_active(self) -> bool: + return self.is_enabled() and self.is_configured() + + def client_config(self) -> dict[str, Any]: + return {} + + @abstractmethod + def is_configured(self) -> bool: ... + + @abstractmethod + def parse_registration(self, body: dict[str, Any]) -> dict[str, Any] | None: ... + + @abstractmethod + def prepare(self, payload: dict[str, Any]) -> str: ... + + @abstractmethod + async def deliver( + self, client: httpx.AsyncClient, registration: dict[str, Any], prepared: str + ) -> Delivery: ... diff --git a/devplacepy/push.py b/devplacepy/push/providers/webpush.py similarity index 70% rename from devplacepy/push.py rename to devplacepy/push/providers/webpush.py index 0b665fc3..8c314f74 100644 --- a/devplacepy/push.py +++ b/devplacepy/push/providers/webpush.py @@ -7,7 +7,6 @@ import logging import os import random import time -from datetime import datetime, timezone from typing import Any from urllib.parse import urlparse @@ -20,7 +19,6 @@ from cryptography.hazmat.primitives.ciphers.aead import AESGCM from cryptography.hazmat.primitives.hashes import SHA256 from cryptography.hazmat.primitives.kdf.hkdf import HKDF -from devplacepy import stealth from devplacepy.config import ( SECONDS_PER_DAY, VAPID_PRIVATE_KEY_FILE, @@ -28,7 +26,15 @@ from devplacepy.config import ( VAPID_PUBLIC_KEY_FILE, VAPID_SUB, ) -from devplacepy.database import get_table +from devplacepy.database import get_setting +from devplacepy.push.providers.base import ( + ACCEPTED, + DEAD, + REJECTED, + Delivery, + PushProvider, +) +from devplacepy.services.base import ConfigField from devplacepy.utils import generate_uid logger = logging.getLogger(__name__) @@ -37,6 +43,8 @@ JWT_LIFETIME_SECONDS = 60 * 60 PUSH_TTL_SECONDS = str(SECONDS_PER_DAY) DEAD_SUBSCRIPTION_STATUSES = (404, 410) ACCEPTED_STATUSES = (200, 201) +SUBJECT_KEY = "push_webpush_subject" +PROVIDER_LABEL = "Web Push (VAPID)" def generate_private_key() -> None: @@ -149,13 +157,17 @@ def public_key_standard_b64() -> str: return base64.b64encode(point).decode("utf-8").rstrip("=") +def subject() -> str: + return get_setting(SUBJECT_KEY, "").strip() or VAPID_SUB + + def create_notification_authorization(push_url: str) -> str: target = urlparse(push_url) audience = f"{target.scheme}://{target.netloc}" issued_at = int(time.time()) return jwt.encode( { - "sub": VAPID_SUB, + "sub": subject(), "aud": audience, "exp": issued_at + JWT_LIFETIME_SECONDS, "nbf": issued_at, @@ -223,78 +235,76 @@ def create_notification_info_with_payload( } -def _mark_subscription_dead(subscription_id: int) -> None: - get_table("push_registration").update( - {"id": subscription_id, "deleted_at": datetime.now(timezone.utc).isoformat()}, - ["id"], - ) - logger.info("Soft-deleted dead push subscription id=%s", subscription_id) +class WebPushProvider(PushProvider): + name = "webpush" + label = PROVIDER_LABEL + config_fields = [ + ConfigField( + SUBJECT_KEY, + "VAPID subject", + type="str", + default=VAPID_SUB, + help="Contact sent as the JWT sub claim, a mailto: or https: URL. Blank uses the built-in default.", + group=PROVIDER_LABEL, + ) + ] + def is_configured(self) -> bool: + return True -async def notify_user(user_uid: str, payload: dict[str, Any]) -> None: - registrations = list( - get_table("push_registration").find(user_uid=user_uid, deleted_at=None) - ) - if not registrations: - logger.debug("No active push subscriptions for user %s", user_uid) - return + def client_config(self) -> dict[str, Any]: + try: + return {"publicKey": public_key_standard_b64()} + except Exception as exc: + logger.error("VAPID key material unavailable: %s", exc) + return {} - body = json.dumps(payload) - async with stealth.stealth_async_client(timeout=10.0) as client: - for subscription in registrations: - endpoint = subscription["endpoint"] - try: - notification_payload = create_notification_info_with_payload( - endpoint, - subscription["key_auth"], - subscription["key_p256dh"], - body, - ) - headers = {**notification_payload["headers"], "TTL": PUSH_TTL_SECONDS} - response = await client.post( - endpoint, headers=headers, content=notification_payload["data"] - ) - except (httpx.HTTPError, ValueError) as exc: - logger.warning("Push error for %s via %s: %s", user_uid, endpoint, exc) - continue + def parse_registration(self, body: dict[str, Any]) -> dict[str, Any] | None: + keys = body.get("keys") + if not isinstance(keys, dict): + return None + endpoint = body.get("endpoint") + key_auth = keys.get("auth") + key_p256dh = keys.get("p256dh") + if not ( + isinstance(endpoint, str) + and isinstance(key_auth, str) + and isinstance(key_p256dh, str) + and endpoint + and key_auth + and key_p256dh + ): + return None + return { + "endpoint": endpoint, + "key_auth": key_auth, + "key_p256dh": key_p256dh, + } - if response.status_code in ACCEPTED_STATUSES: - logger.debug("Push delivered to %s via %s", user_uid, endpoint) - elif response.status_code in DEAD_SUBSCRIPTION_STATUSES: - _mark_subscription_dead(subscription["id"]) - else: - logger.warning( - "Push rejected (%s) for %s via %s", - response.status_code, - user_uid, - endpoint, - ) + def prepare(self, payload: dict[str, Any]) -> str: + return json.dumps(payload) - -async def register( - user_uid: str, endpoint: str, key_auth: str, key_p256dh: str -) -> tuple[dict[str, Any], bool]: - table = get_table("push_registration") - existing = table.find_one( - user_uid=user_uid, - endpoint=endpoint, - key_auth=key_auth, - key_p256dh=key_p256dh, - deleted_at=None, - ) - if existing: - logger.debug("Push subscription already registered for user %s", user_uid) - return existing, False - - record = { - "uid": generate_uid(), - "user_uid": user_uid, - "endpoint": endpoint, - "key_auth": key_auth, - "key_p256dh": key_p256dh, - "created_at": datetime.now(timezone.utc).isoformat(), - "deleted_at": None, - } - table.insert(record) - logger.info("Registered push subscription for user %s", user_uid) - return record, True + async def deliver( + self, client: httpx.AsyncClient, registration: dict[str, Any], prepared: str + ) -> Delivery: + endpoint = registration.get("endpoint") or "" + if not endpoint: + return Delivery(DEAD, "missing endpoint") + try: + notification_payload = create_notification_info_with_payload( + endpoint, + registration["key_auth"], + registration["key_p256dh"], + prepared, + ) + headers = {**notification_payload["headers"], "TTL": PUSH_TTL_SECONDS} + response = await client.post( + endpoint, headers=headers, content=notification_payload["data"] + ) + except (httpx.HTTPError, ValueError, KeyError, TypeError) as exc: + return Delivery(REJECTED, str(exc)) + if response.status_code in ACCEPTED_STATUSES: + return Delivery(ACCEPTED) + if response.status_code in DEAD_SUBSCRIPTION_STATUSES: + return Delivery(DEAD, str(response.status_code)) + return Delivery(REJECTED, str(response.status_code)) diff --git a/devplacepy/push/store.py b/devplacepy/push/store.py new file mode 100644 index 00000000..24fa3eb9 --- /dev/null +++ b/devplacepy/push/store.py @@ -0,0 +1,83 @@ +# retoor + +import logging +from datetime import datetime, timezone +from typing import Any + +from devplacepy.database import db, get_table +from devplacepy.push.providers import DEFAULT_PROVIDER +from devplacepy.utils import generate_uid + +logger = logging.getLogger(__name__) + +TABLE = "push_registration" + + +def table(): + return get_table(TABLE) + + +def provider_of(registration: dict[str, Any]) -> str: + return registration.get("provider") or DEFAULT_PROVIDER + + +def active_for_user(user_uid: str) -> list[dict[str, Any]]: + return list(table().find(user_uid=user_uid, deleted_at=None)) + + +def register( + user_uid: str, provider: str, fields: dict[str, Any] +) -> tuple[dict[str, Any], bool]: + registrations = table() + existing = registrations.find_one( + user_uid=user_uid, provider=provider, deleted_at=None, **fields + ) + if existing: + logger.debug("Push subscription already registered for user %s", user_uid) + return existing, False + + record = { + "uid": generate_uid(), + "user_uid": user_uid, + "provider": provider, + "created_at": datetime.now(timezone.utc).isoformat(), + "deleted_at": None, + **fields, + } + registrations.insert(record) + logger.info("Registered %s push subscription for user %s", provider, user_uid) + return record, True + + +def mark_dead(registration_id: int) -> None: + table().update( + {"id": registration_id, "deleted_at": datetime.now(timezone.utc).isoformat()}, + ["id"], + ) + logger.info("Soft-deleted dead push subscription id=%s", registration_id) + + +def prune(cutoff: str) -> int: + if TABLE not in db.tables: + return 0 + rows = list(table().find(deleted_at={"<": cutoff})) + if not rows: + return 0 + table().delete(deleted_at={"<": cutoff}) + return len(rows) + + +def counts() -> dict[str, int]: + if TABLE not in db.tables: + return {} + totals: dict[str, int] = {"dead": 0} + for row in db.query( + f"SELECT provider AS provider, deleted_at IS NULL AS live, COUNT(*) AS total " + f"FROM {TABLE} GROUP BY provider, deleted_at IS NULL" + ): + provider = row["provider"] or DEFAULT_PROVIDER + if row["live"]: + totals[provider] = totals.get(provider, 0) + int(row["total"]) + else: + totals["dead"] += int(row["total"]) + return totals diff --git a/devplacepy/routers/push.py b/devplacepy/routers/push.py index bc9b2ed1..ad347e36 100644 --- a/devplacepy/routers/push.py +++ b/devplacepy/routers/push.py @@ -5,6 +5,7 @@ from fastapi import APIRouter, Request from fastapi.responses import FileResponse, JSONResponse from devplacepy import push from devplacepy.config import STATIC_DIR +from devplacepy.push import providers from devplacepy.utils import require_user_api from urllib.parse import urlparse from devplacepy.services.audit import record as audit @@ -22,7 +23,11 @@ WELCOME_PAYLOAD = { @router.get("/push.json") async def push_public_key() -> JSONResponse: - return JSONResponse({"publicKey": push.public_key_standard_b64()}) + configs = providers.client_config() + webpush = configs.get(providers.DEFAULT_PROVIDER, {}) + return JSONResponse( + {"publicKey": webpush.get("publicKey", ""), "providers": configs} + ) @router.post("/push.json") @@ -33,21 +38,18 @@ async def push_register(request: Request) -> JSONResponse: except ValueError: return JSONResponse({"error": "Invalid JSON"}, status_code=400) - keys = body.get("keys") if isinstance(body, dict) else None - if not ( - isinstance(keys, dict) - and body.get("endpoint") - and keys.get("p256dh") - and keys.get("auth") - ): + if not isinstance(body, dict): return JSONResponse({"error": "Invalid request"}, status_code=400) - _, created = await push.register( - user_uid=user["uid"], - endpoint=body["endpoint"], - key_auth=keys["auth"], - key_p256dh=keys["p256dh"], - ) + provider = providers.get(body.get("provider")) + if provider is None or not providers.is_active(provider): + return JSONResponse({"error": "Unknown provider"}, status_code=400) + + fields = provider.parse_registration(body) + if fields is None: + return JSONResponse({"error": "Invalid request"}, status_code=400) + + _, created = push.register(user["uid"], provider.name, fields) if created: try: @@ -62,7 +64,13 @@ async def push_register(request: Request) -> JSONResponse: target_type="user", target_uid=user["uid"], target_label=user.get("username"), - metadata={"endpoint_host": urlparse(body["endpoint"]).hostname, "created": created}, + metadata={ + "provider": provider.name, + "endpoint_host": urlparse(fields["endpoint"]).hostname + if fields.get("endpoint") + else None, + "created": created, + }, summary=f"{user.get('username')} {'registered' if created else 'updated'} a push subscription", links=[audit.target("user", user["uid"], user.get("username"))], ) diff --git a/devplacepy/services/push/__init__.py b/devplacepy/services/push/__init__.py new file mode 100644 index 00000000..7cb9df91 --- /dev/null +++ b/devplacepy/services/push/__init__.py @@ -0,0 +1,9 @@ +# retoor + +from devplacepy.services.push.service import ( + DEFAULT_RETENTION_DAYS, + RETENTION_KEY, + PushService, +) + +__all__ = ["DEFAULT_RETENTION_DAYS", "PushService", "RETENTION_KEY"] diff --git a/devplacepy/services/push/service.py b/devplacepy/services/push/service.py new file mode 100644 index 00000000..303ae1cf --- /dev/null +++ b/devplacepy/services/push/service.py @@ -0,0 +1,79 @@ +# retoor + +import logging +from datetime import datetime, timedelta, timezone + +from devplacepy.database import get_int_setting +from devplacepy.push import providers, store +from devplacepy.push.delivery import ( + DEFAULT_TIMEOUT_SECONDS, + MAX_TIMEOUT_SECONDS, + MIN_TIMEOUT_SECONDS, + TIMEOUT_KEY, +) +from devplacepy.services.base import BaseService, ConfigField + +logger = logging.getLogger(__name__) + +RETENTION_KEY = "push_dead_retention_days" +DEFAULT_RETENTION_DAYS = 30 + + +class PushService(BaseService): + title = "Push notifications" + description = ( + "Configures the push providers and prunes dead subscriptions. Delivery is " + "independent of this service and continues while it is stopped." + ) + details = ( + "Notifications reach a user through every provider they have a live subscription " + "for. A provider delivers only while its own Enabled toggle is on and its " + "configuration is complete, so an unconfigured provider is inert." + ) + default_enabled = True + min_interval = 3600 + METRICS_SECONDS = 300 + config_fields = [ + ConfigField( + RETENTION_KEY, + "Dead subscription retention (days)", + type="int", + default=DEFAULT_RETENTION_DAYS, + minimum=0, + help="Subscriptions the push services rejected as gone are removed after this many days. 0 disables pruning.", + group="General", + ), + ConfigField( + TIMEOUT_KEY, + "Delivery timeout (seconds)", + type="int", + default=DEFAULT_TIMEOUT_SECONDS, + minimum=MIN_TIMEOUT_SECONDS, + maximum=MAX_TIMEOUT_SECONDS, + help="Per request timeout used for every push provider.", + group="General", + ), + *providers.admin_fields(), + ] + + def __init__(self) -> None: + super().__init__("push", interval_seconds=86400) + + async def run_once(self) -> None: + days = get_int_setting(RETENTION_KEY, DEFAULT_RETENTION_DAYS) + if days <= 0: + self.log("Retention disabled (0 days); nothing pruned") + return + cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat() + removed = store.prune(cutoff) + self.log(f"Pruned {removed} dead push subscriptions older than {days}d") + + def collect_metrics(self) -> dict: + totals = store.counts() + metrics = {"dead": totals.get("dead", 0)} + for provider in providers.PROVIDERS.values(): + metrics[f"{provider.name}_active"] = totals.get(provider.name, 0) + metrics[f"{provider.name}_ready"] = ( + 1 if providers.is_active(provider) else 0 + ) + return metrics diff --git a/devplacepy/static/js/PushManager.js b/devplacepy/static/js/PushManager.js index a2db8fe9..b566a158 100644 --- a/devplacepy/static/js/PushManager.js +++ b/devplacepy/static/js/PushManager.js @@ -54,6 +54,9 @@ export class PushManager { } const keyData = await Http.getJson("/push.json"); + if (!keyData.publicKey) { + return; + } const applicationServerKey = Uint8Array.from(atob(keyData.publicKey), (c) => c.charCodeAt(0)); const subscription = await registration.pushManager.subscribe({