diff --git a/backend/recherche_cors.py b/backend/recherche_cors.py new file mode 100644 index 0000000..af90390 --- /dev/null +++ b/backend/recherche_cors.py @@ -0,0 +1,189 @@ +"""CORS bornée pour la recherche fédérée (fusion-trois-apps #68). + +La page « Rechercher partout » (``/recherche``) appelle, depuis le +navigateur, ``GET /api/search`` sur chaque application sœur. Sans +en-têtes CORS, le navigateur refuse la lecture des réponses +cross-origin ; sans borne, on ouvrirait la lecture à tout le web. +Ce module n'autorise donc que : + +- le chemin ``/api/search`` (jamais les autres routes, jamais les + écritures) ; +- les origines sœurs : celles de ``backend.draco_config`` + (``urls_publiques()`` : explicite > dérivation par l'hôte de + l'agent > défaut officiel), **plus** les origines prod ou ``-dev`` + officielles correspondant à l'hôte de la requête — ce second + ensemble couvre le cas où la configuration seule ne déclare pas + l'environnement courant ; +- avec credentials (le cookie SSO ``draco_session``, accepté par + ``/api/search`` via le repli de ``backend.auth.middleware``, ou + le cookie natif ``access_token``). + +Jamais ``*``. Le middleware est un ASGI pur : il répond lui-même +aux preflights OPTIONS (qui ne portent jamais de cookies) et ajoute +les en-têtes aux réponses de recherche, sans toucher aux autres +routes ni à la politique same-origin explicite du CORSMiddleware +global (``allow_origins=[]``), qui reste la règle par défaut. +""" + +from __future__ import annotations + +from urllib.parse import urlparse + +SEARCH_PATH = "/api/search" + +PROD_ORIGINS = frozenset({ + "https://og.dracodev.net", + "https://flowdeck.dracodev.net", + "https://tiktak.dracodev.net", + "https://kanbante.dracodev.net", +}) + +DEV_ORIGINS = frozenset({ + "https://og-dev.dracodev.net", + "https://flowdeck-dev.dracodev.net", + "https://tiktak-dev.dracodev.net", + "https://kanbante-dev.dracodev.net", +}) + +# Clés de draco_config.urls_publiques() qui sont des applications sœurs. +_SOEUR_KEYS = ("obsigate", "flowdeck", "tiktak", "kanbante") + + +def is_dev_host(hostname: str | None) -> bool: + """Même règle que la barre v0.1.6 (voir frontend/js/recherche.js).""" + if not hostname: + return False + h = hostname.lower().split(":")[0] + if not h.endswith(".dracodev.net"): + return False + first = h.split(".")[0] + return len(first) > 4 and first.endswith("-dev") + + +def is_prod_host(hostname: str | None) -> bool: + """Hôte dracodev.net qui n'est ni ``-dev`` ni un piège de suffixe.""" + if not hostname: + return False + h = hostname.lower().split(":")[0] + if not h.endswith(".dracodev.net"): + return False + first = h.split(".")[0] + return len(first) > 0 and not first.endswith("-dev") + + +def origin_of(url: str | None) -> str | None: + """Origine (schéma + hôte) d'une URL, sans chemin ni ``/`` final.""" + if not url: + return None + parsed = urlparse(url.strip()) + if parsed.scheme not in ("http", "https") or not parsed.hostname: + return None + host = parsed.hostname.lower() + port = f":{parsed.port}" if parsed.port else "" + return f"{parsed.scheme}://{host}{port}" + + +def sister_origins(resolved: dict | None = None, request_host: str | None = None) -> frozenset: + """Origines autorisées pour la lecture cross-origin de /api/search. + + ``resolved`` : dictionnaire rendu par ``urls_publiques()``. + Sans argument, le résolveur est appelé (import paresseux) ; en cas + d'échec, repli sur prod + dev officiels (premier parti uniquement). + ``request_host`` ajoute l'ensemble officiel de l'environnement de + la requête (voir le docstring du module). + """ + origins = set() + if resolved is None: + try: + from backend.draco_config import urls_publiques + + resolved = urls_publiques() + except Exception: + resolved = {} + for key in _SOEUR_KEYS: + origin = origin_of((resolved or {}).get(key)) + if origin: + origins.add(origin) + if is_dev_host(request_host): + origins |= DEV_ORIGINS + elif is_prod_host(request_host): + origins |= PROD_ORIGINS + if not origins: + return PROD_ORIGINS | DEV_ORIGINS + return frozenset(origins) + + +def cors_headers_for( + origin: str | None, resolved: dict | None = None, request_host: str | None = None +) -> dict | None: + """En-têtes CORS d'une réponse de recherche, ou None si refusée.""" + if not origin: + return None + if origin.strip() not in sister_origins(resolved, request_host): + return None + return { + "Access-Control-Allow-Origin": origin.strip(), + "Access-Control-Allow-Credentials": "true", + "Vary": "Origin", + } + + +def preflight_headers( + origin: str | None, resolved: dict | None = None, request_host: str | None = None +) -> dict | None: + """En-têtes d'une réponse de preflight OPTIONS pour /api/search.""" + headers = cors_headers_for(origin, resolved, request_host) + if headers is None: + return None + headers.update({ + "Access-Control-Allow-Methods": "GET, OPTIONS", + "Access-Control-Allow-Headers": "Accept, Content-Type, Authorization", + "Access-Control-Max-Age": "600", + }) + return headers + + +class RechercheCorsMiddleware: + """ASGI pur : CORS bornée pour ``GET /api/search`` (voir module).""" + + def __init__(self, app): + self.app = app + + async def __call__(self, scope, receive, send): + if scope.get("type") != "http": + await self.app(scope, receive, send) + return + path = scope.get("path", "") + headers_in = { + k.decode("latin-1").lower(): v.decode("latin-1") + for k, v in scope.get("headers", []) + } + origin = headers_in.get("origin", "") + request_host = ( + headers_in.get("x-forwarded-host", "") or headers_in.get("host", "") + ).split(",")[0].strip() + if path == SEARCH_PATH and scope.get("method") == "OPTIONS": + preflight = preflight_headers(origin, request_host=request_host) + if preflight is not None: + headers_out = [(k.lower().encode("latin-1"), v.encode("latin-1")) + for k, v in preflight.items()] + headers_out.append((b"content-length", b"0")) + await send({"type": "http.response.start", "status": 204, + "headers": headers_out}) + await send({"type": "http.response.body", "body": b""}) + return + if path != SEARCH_PATH: + await self.app(scope, receive, send) + return + + async def send_with_cors(message): + if message["type"] == "http.response.start": + cors = cors_headers_for(origin, request_host=request_host) + if cors is not None: + message["headers"] = list(message.get("headers", [])) + [ + (k.lower().encode("latin-1"), v.encode("latin-1")) + for k, v in cors.items() + ] + await send(message) + + await self.app(scope, receive, send_with_cors)