File size: 2,870 Bytes
a12d188 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 | """Notification / out-of-band confirmation endpoints — a torch-free router.
Isolated from the monolithic `api.routes` (which imports the biometric models/torch)
so these endpoints and their API tests run in the lightweight, torch-free environment.
Paths, methods and request/response contracts are UNCHANGED from the previous
definitions in `api.routes`."""
from __future__ import annotations
import time
from fastapi import APIRouter, Body
from fastapi.responses import JSONResponse
from api.services import audit_event, notifications, persist_prefs
router = APIRouter()
@router.post("/notify/prefs")
def notify_set_prefs(body: dict = Body(...)) -> dict:
r = notifications.set_prefs(body["user_id"], channels=body.get("channels"),
email=body.get("email"), phone=body.get("phone"))
persist_prefs()
return r
@router.get("/notify/prefs")
def notify_get_prefs(user_id: str) -> dict:
return notifications.get_prefs(user_id)
@router.post("/notify/subscribe")
def notify_subscribe(body: dict = Body(...)) -> dict:
"""Store a Web Push subscription (from the browser) for the user."""
return notifications.subscribe_push(body["user_id"], body["subscription"])
@router.post("/notify/request")
def notify_request(body: dict = Body(...)) -> dict:
"""Create a demo out-of-band payment confirmation (dynamic-linked to amount+merchant)
using the existing notification service, so the React OOB demo can exercise the
request -> respond -> status workflow without going through /wallet/pay."""
return notifications.send_payment_confirmation(
body["user_id"], float(body.get("amount", 0.0)),
str(body.get("merchant", "")), now=time.time())
@router.get("/notify/status")
def notify_status(confirmation_id: str) -> dict:
"""Lifecycle of an OOB confirmation for UI polling: pending/approved/rejected/expired."""
return {"confirmation_id": confirmation_id,
"status": notifications.confirmation_status(confirmation_id, now=time.time())}
@router.post("/notify/respond")
def notify_respond(body: dict = Body(...)):
"""Approve/deny an OOB confirmation. Enforces terminal-state immutability: a
conflicting response after a terminal state returns HTTP 409 (oob_already_finalized)
and leaves the final state unchanged; a same-decision retry is idempotent."""
r = notifications.respond(body["confirmation_id"], bool(body.get("approve", False)),
code=body.get("code"), now=time.time())
if r.get("conflict"):
# Safe security-audit event (no code/secret) for the rejected conflicting transition.
audit_event("oob", "oob_conflict_rejected",
{"confirmation_id": r.get("confirmation_id"), "final_state": r.get("status")})
return JSONResponse(status_code=409, content=r)
return r
|