Recherche partout (#68) : CORS résolue avec l'hôte de la requête (ordre middlewares)
CI / test (push) Failing after 1s
CI / test (pull_request) Failing after 0s

This commit is contained in:
2026-10-28 12:13:03 -04:00
parent b2d2fd1eee
commit 39750851b0
+23 -10
View File
@@ -66,18 +66,24 @@ def origin_of(url: str | None) -> str | None:
return f"{parsed.scheme}://{host}{port}" return f"{parsed.scheme}://{host}{port}"
def sister_origins(resolved: dict | None = None) -> frozenset: def sister_origins(
resolved: dict | None = None, request_host: str | None = None
) -> frozenset:
"""Origines autorisées pour la lecture cross-origin de /api/search. """Origines autorisées pour la lecture cross-origin de /api/search.
``resolved`` : dictionnaire ``draco_urls`` rendu par ``render()``. ``resolved`` : dictionnaire ``draco_urls`` rendu par ``render()``.
Sans argument, le résolveur est appelé (import paresseux) ; en cas Sans argument, le résolveur est appelé (import paresseux) avec
d'échec, repli sur prod + dev officiels (premier parti uniquement). 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: if resolved is None:
try: try:
from app.config import get_settings, resoudre_urls_draco from app.config import get_settings, resoudre_urls_draco
urls = resoudre_urls_draco(base=get_settings()) urls = resoudre_urls_draco(hote_requete=request_host, base=get_settings())
resolved = { resolved = {
"obsigate": urls.sso_url, "obsigate": urls.sso_url,
"flowdeck": urls.flowdeck_url, "flowdeck": urls.flowdeck_url,
@@ -94,11 +100,13 @@ def sister_origins(resolved: dict | None = None) -> frozenset:
return frozenset(origins) return frozenset(origins)
def cors_headers_for(origin: str | None, resolved: dict | None = None) -> dict | None: 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.""" """En-têtes CORS d'une réponse de recherche, ou None si refusée."""
if not origin: if not origin:
return None return None
if origin.strip() not in sister_origins(resolved): if origin.strip() not in sister_origins(resolved, request_host):
return None return None
return { return {
"Access-Control-Allow-Origin": origin.strip(), "Access-Control-Allow-Origin": origin.strip(),
@@ -107,9 +115,11 @@ def cors_headers_for(origin: str | None, resolved: dict | None = None) -> dict |
} }
def preflight_headers(origin: str | None, resolved: dict | None = None) -> dict | None: 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.""" """En-têtes d'une réponse de preflight OPTIONS pour /api/search."""
headers = cors_headers_for(origin, resolved) headers = cors_headers_for(origin, resolved, request_host)
if headers is None: if headers is None:
return None return None
headers.update({ headers.update({
@@ -176,8 +186,11 @@ class RechercheCorsMiddleware:
for k, v in scope.get("headers", []) for k, v in scope.get("headers", [])
} }
origin = headers_in.get("origin", "") 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": if path == SEARCH_PATH and scope.get("method") == "OPTIONS":
preflight = preflight_headers(origin) preflight = preflight_headers(origin, request_host=request_host)
if preflight is not None: if preflight is not None:
body = b"" body = b""
headers_out = [(k.lower().encode("latin-1"), v.encode("latin-1")) headers_out = [(k.lower().encode("latin-1"), v.encode("latin-1"))
@@ -193,7 +206,7 @@ class RechercheCorsMiddleware:
async def send_with_cors(message): async def send_with_cors(message):
if message["type"] == "http.response.start": if message["type"] == "http.response.start":
cors = cors_headers_for(origin) cors = cors_headers_for(origin, request_host=request_host)
if cors is not None: if cors is not None:
message["headers"] = list(message.get("headers", [])) + [ message["headers"] = list(message.get("headers", [])) + [
(k.lower().encode("latin-1"), v.encode("latin-1")) (k.lower().encode("latin-1"), v.encode("latin-1"))