"""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)