Recherche partout (#68, rebase sur main post-#70 et post-#69)
This commit is contained in:
@@ -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)
|
||||
Reference in New Issue
Block a user