"""Recherche fédérée (fusion-trois-apps #68) : JSON de recherche, CORS bornée et conversion des groupes d'issues. Ce module ne dépend d'aucun autre module applicatif à l'import (les résolveurs sont appelés paresseusement) : les fonctions pures sont testables sans base ni serveur. - ``GET /api/search`` (route dans ``app/main.py``) réutilise la recherche Gitea de la page ``/search`` (titres + corps, tous états, tous projets du workspace) et la regroupe par dépôt. - La CORS bornée n'autorise que ``GET /api/search``, et seulement pour les origines sœurs officielles **prod ET dev, listées explicitement** (``PROD_ORIGINS | DEV_ORIGINS``), **plus** les origines du service d'agents (``AGENT_ORIGINS`` : console Spark prod ET dev — fédération console, défaut 2), **plus** les origines résolues par ``app.config`` — jamais ``*``. Correctif 2026-10-28 (fusion #68/#69, matrice constatée par Bruno) : l'ancienne allowlist ne contenait que l'environnement résolu (prod OU dev), donc une page de l'autre environnement voyait sa section afficher « Injoignable depuis cet hôte (CORS ou réseau) ». Le middleware est un ASGI pur : il répond lui-même aux preflights OPTIONS (qui ne portent jamais de cookies, donc qu'``AuthMiddleware`` refuserait sinon) et ajoute les en-têtes aux réponses de recherche, sans toucher aux autres routes. """ from __future__ import annotations from urllib.parse import urlparse SEARCH_PATH = "/api/search" _SNIPPET_LEN = 160 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", }) # Origines du service d'agents (console Spark) : la page de # recherche de la console appelle /api/search depuis le navigateur # avec le cookie SSO — sans ces deux origines, le navigateur refuse # la lecture (fédération console, défaut 2). Jamais "*". AGENT_ORIGINS = frozenset({ "https://agent.dracodev.net", "https://agent-dev.dracodev.net", }) # Clés du dictionnaire rendu par render() (app/main.py). _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 static/js/recherche.js).""" if not hostname: return False h = hostname.lower() if not h.endswith(".dracodev.net"): return False first = h.split(".")[0] return len(first) > 4 and 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 ``draco_urls`` rendu par ``render()``. Sans argument, le résolveur est appelé (import paresseux) avec l'hôte de la requête quand il est connu — ce middleware s'exécute avant tout poseur de contexte d'hôte, et sans lui la résolution retomberait sur la configuration seule et raterait l'environnement ``-dev`` ; en cas d'échec, repli sur prod + dev officiels (premier parti uniquement). """ if resolved is None: try: from app.config import get_settings, resoudre_urls_draco urls = resoudre_urls_draco(hote_requete=request_host, base=get_settings()) resolved = { "obsigate": urls.sso_url, "flowdeck": urls.flowdeck_url, "tiktak": urls.tiktak_url, "kanbante": urls.public_url, } except Exception: return PROD_ORIGINS | DEV_ORIGINS | AGENT_ORIGINS origins = set() for key in _SOEUR_KEYS: origin = origin_of((resolved or {}).get(key)) if origin: origins.add(origin) # Correctif 2026-10-28 : prod ET dev toujours, quelle que soit # la résolution — ``resolved`` n'ajoute que des origines de # configuration sans jamais retirer une origine officielle. return frozenset(origins) | PROD_ORIGINS | DEV_ORIGINS | AGENT_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 def groupes_vers_json(groups, limit: int = 20): """Convertit la sortie de ``_group_issue_results`` en JSON public. Seuls les champs utiles sont exposés (numéro, titre, état, corps tronqué à 160 caractères, labels, colonne). Retourne ``(groupes, total)`` où ``total`` est le nombre de lignes retournées (plafonné à ``limit``). """ out: list[dict] = [] total = 0 restant = max(1, min(int(limit), 50)) for group in groups or []: rows: list[dict] = [] for row in (group or {}).get("rows", []) or []: if restant <= 0: break issue = (row or {}).get("issue", {}) or {} labels = [] for label in issue.get("labels") or []: if isinstance(label, dict): labels.append({"name": label.get("name"), "color": label.get("color")}) rows.append({ "issue": { "number": issue.get("number"), "title": issue.get("title"), "state": issue.get("state"), "body": (issue.get("body") or "")[:_SNIPPET_LEN], "labels": labels, }, "column": (row or {}).get("column"), }) restant -= 1 total += 1 if rows: out.append({"full_name": (group or {}).get("full_name"), "rows": rows}) if restant <= 0: break return out, total 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("host", "") or headers_in.get("x-forwarded-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: body = b"" 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": body}) 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)