Spaces:
Sleeping
Sleeping
| """ | |
| The public face of Collect (docs/blueprint/adr/0006) — the "DMZ". Exactly these routes and | |
| nothing else are reachable without signing in, and each one only calls backend/collect.py. | |
| It is an APIRouter so the main app can include it today, and `collect_app` is the same router | |
| as a standalone ASGI app for deploying on its own later. | |
| Cross-origin: the snippet sends its JSON as text/plain, which browsers treat as a "simple" | |
| request (no preflight), and this module sets Access-Control-Allow-Origin per key itself. The | |
| site-wide CORS middleware in app.py is for the signed-in app and stays out of the way. | |
| """ | |
| from __future__ import annotations | |
| import html | |
| import json | |
| from pathlib import Path | |
| from fastapi import APIRouter, FastAPI, Request | |
| from fastapi.responses import HTMLResponse, JSONResponse, Response | |
| from backend import collect | |
| router = APIRouter() | |
| _SNIPPET_PATH = Path(__file__).parent / "static" / "m8ion.js" | |
| def _client_ip(request: Request) -> str: | |
| # Behind a proxy the first X-Forwarded-For entry is the visitor. Only trusted as a rate-limit | |
| # bucket, never for anything that grants access. | |
| forwarded = request.headers.get("x-forwarded-for", "") | |
| if forwarded: | |
| return forwarded.split(",")[0].strip() | |
| return request.client.host if request.client else "unknown" | |
| def _with_cors(response: Response, public_key: str, origin: str | None) -> Response: | |
| allowed = collect.allowed_origin(public_key, origin) | |
| if allowed: | |
| response.headers["Access-Control-Allow-Origin"] = allowed | |
| response.headers["Vary"] = "Origin" | |
| return response | |
| async def submit(public_key: str, request: Request): | |
| origin = request.headers.get("origin") | |
| raw = await request.body() | |
| if len(raw) > 16_000: | |
| return _with_cors(JSONResponse({"detail": "Submission is too large"}, status_code=413), public_key, origin) | |
| try: | |
| payload = json.loads(raw) | |
| except ValueError: | |
| return _with_cors(JSONResponse({"detail": "Malformed submission"}, status_code=400), public_key, origin) | |
| try: | |
| result = collect.submit(public_key, payload, origin, _client_ip(request)) | |
| response = JSONResponse(result) | |
| except collect.CollectError as exc: | |
| response = JSONResponse({"detail": str(exc)}, status_code=exc.status) | |
| return _with_cors(response, public_key, origin) | |
| _PAGE = """<!doctype html><html lang="en"><head><meta charset="utf-8"> | |
| <meta name="viewport" content="width=device-width,initial-scale=1"><title>{title}</title> | |
| <style>body{{font-family:system-ui,sans-serif;max-width:32rem;margin:15vh auto;padding:0 1.25rem;color:#1b1f23}} | |
| h1{{font-size:1.4rem}}p{{color:#555;line-height:1.5}}</style></head> | |
| <body><h1>{title}</h1><p>{message}</p></body></html>""" | |
| _CONFIRM_TEXT = { | |
| "confirmed": ("You're on the list", "Thanks, your email address is confirmed. You can close this page."), | |
| "expired": ("This link has expired", "Please sign up again from the form to get a new confirmation email."), | |
| "invalid": ("This link is no longer valid", "It may already have been used. If you are not on the list yet, sign up again."), | |
| } | |
| def confirm(token: str): | |
| title, message = _CONFIRM_TEXT[collect.confirm(token)] | |
| return HTMLResponse(_PAGE.format(title=html.escape(title), message=html.escape(message))) | |
| def snippet(): | |
| return Response( | |
| _SNIPPET_PATH.read_text(encoding="utf-8"), | |
| media_type="application/javascript", | |
| headers={"Cache-Control": "public, max-age=300", "Access-Control-Allow-Origin": "*"}, | |
| ) | |
| collect_app = FastAPI(title="M8ion Collect") | |
| collect_app.include_router(router) | |