Recherche partout (#68) : JSON de recherche + CORS bornée

app/recherche.py

@@ -0,0 +1,204 @@
+"""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 résolues par ``app.config`` (explicite >
+  hôte de la requête > dérivation > défaut officiel — donc ``-dev``
+  reste ``-dev``), jamais ``*``. 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",
+})
+
+# 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) -> 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) ; 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(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
+    origins = set()
+    for key in _SOEUR_KEYS:
+        origin = origin_of((resolved or {}).get(key))
+        if origin:
+            origins.add(origin)
+    return frozenset(origins)
+
+
+def cors_headers_for(origin: str | None, resolved: dict | 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):
+        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) -> dict | None:
+    """En-têtes d'une réponse de preflight OPTIONS pour /api/search."""
+    headers = cors_headers_for(origin, resolved)
+    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", "")
+        if path == SEARCH_PATH and scope.get("method") == "OPTIONS":
+            preflight = preflight_headers(origin)
+            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)
+                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)