191 lines
6.9 KiB
Python
191 lines
6.9 KiB
Python
"""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 résolues par
|
|
``app.services.draco_urls.toutes()`` (explicite > hôte > dérivation
|
|
> défaut officiel), **plus** les origines prod ou ``-dev``
|
|
officielles correspondant à l'hôte de la requête — ce second
|
|
ensemble est nécessaire car ce middleware s'exécute avant le
|
|
middleware CSP qui pose le contexte d'hôte de la requête pour
|
|
``draco_urls`` ; sans lui, une page servie par ``flowdeck-dev``
|
|
verrait ses sœurs ``-dev`` refusées quand la configuration seule
|
|
ne les déclare pas ;
|
|
- avec credentials (le cookie SSO ``draco_session``, accepté par
|
|
``/api/search`` depuis cette livraison, ou la session native).
|
|
|
|
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 CORS globale de l'application.
|
|
"""
|
|
|
|
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_urls.toutes() 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 static/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 ``draco_urls.toutes()``.
|
|
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 app.services import draco_urls
|
|
|
|
resolved = draco_urls.toutes()
|
|
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)
|