bruno
34253c7b3f
FlowDeck CI / lint (push) Failing after 1s
FlowDeck CI / test (push) Failing after 1s
FlowDeck CI / docker (push) Skipped
FlowDeck CI / lint (pull_request) Failing after 1m49s
FlowDeck CI / test (pull_request) Failing after 32m50s
FlowDeck CI / docker (pull_request) Skipped
121 lines
5.0 KiB
Python
121 lines
5.0 KiB
Python
"""Défaut 3b — /api/search Flowdeck depuis la console agent-dev.
|
|
|
|
Cause prouvée en déployé (flowdeck-dev, 2026-10-28) : sur une réponse
|
|
GET /api/search, le CORSMiddleware global pose déjà
|
|
``Access-Control-Allow-Credentials: true`` (comportement Starlette
|
|
avec ``allow_credentials=True``, même hors allowlist), et l'ancien
|
|
middleware de recherche EMPILAIT ses en-têtes par-dessus — d'où, pour
|
|
l'origine agent-dev, ``Allow-Credentials`` en double (« true, true »
|
|
≠ « true » pour le contrôle CORS du navigateur → fetch rejeté →
|
|
« injoignable (CORS ou réseau) » dans la console, alors que TikTak et
|
|
Kanbante, sans ce cumul, répondent avec des en-têtes uniques) et,
|
|
pour une origine sœur déjà dans la allowlist globale, ``Allow-Origin``
|
|
en double (rejet certain). Le correctif : le middleware de recherche
|
|
REMPLACE les en-têtes CORS existants au lieu de les empiler (même
|
|
patron que app/recherche_cors.py de la console).
|
|
|
|
Ces tests pilotent le middleware en ASGI pur, avec une application
|
|
interne qui imite le CORSMiddleware global (un en-tête
|
|
Allow-Credentials déjà posé, et, pour une sœur, Allow-Origin aussi).
|
|
``sister_origins`` est figé sur les listes officielles (les fonctions
|
|
pures sont déjà couvertes par tests/test_recherche.py) pour garder le
|
|
test hermétique (aucune résolution draco_urls, aucune base).
|
|
"""
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from app.middleware import recherche_cors as rc
|
|
|
|
ORIGINES = rc.PROD_ORIGINS | rc.DEV_ORIGINS | rc.AGENT_ORIGINS
|
|
|
|
|
|
def _lancer(middleware, scope):
|
|
"""Exécute le middleware ASGI et retourne les messages envoyés."""
|
|
recus = []
|
|
|
|
async def recevoir():
|
|
return {"type": "http.request", "body": b"", "more_body": False}
|
|
|
|
async def envoyer(message):
|
|
recus.append(message)
|
|
|
|
asyncio.run(middleware(scope, recevoir, envoyer))
|
|
return recus
|
|
|
|
|
|
def _application_interne(entetes):
|
|
async def appli(scope, recevoir, envoyer):
|
|
await envoyer({"type": "http.response.start", "status": 401,
|
|
"headers": list(entetes)})
|
|
await envoyer({"type": "http.response.body", "body": b"{}"})
|
|
return appli
|
|
|
|
|
|
def _scope(methode, origine):
|
|
return {"type": "http", "method": methode, "path": "/api/search",
|
|
"headers": [(b"origin", origine.encode("latin-1")),
|
|
(b"host", b"flowdeck-dev.dracodev.net")]}
|
|
|
|
|
|
def _compter(entetes, nom):
|
|
return sum(1 for k, _v in entetes if k == nom.encode("latin-1"))
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _origines_figees(monkeypatch):
|
|
monkeypatch.setattr(rc, "sister_origins", lambda *a, **k: ORIGINES)
|
|
|
|
|
|
def test_get_agent_dev_un_seul_jeu_d_entetes():
|
|
"""Cas du défaut : globale (ACAC seul) + recherche → un seul jeu."""
|
|
interne = _application_interne([
|
|
(b"content-type", b"application/json"),
|
|
(b"access-control-allow-credentials", b"true")])
|
|
recus = _lancer(rc.RechercheCorsMiddleware(interne),
|
|
_scope("GET", "https://agent-dev.dracodev.net"))
|
|
entetes = recus[0]["headers"]
|
|
assert _compter(entetes, "access-control-allow-credentials") == 1
|
|
assert _compter(entetes, "access-control-allow-origin") == 1
|
|
assert _compter(entetes, "vary") == 1
|
|
assert (b"access-control-allow-origin",
|
|
b"https://agent-dev.dracodev.net") in entetes
|
|
assert (b"content-type", b"application/json") in entetes
|
|
assert recus[0]["status"] == 401 # le statut traverse inchangé
|
|
|
|
|
|
def test_get_origine_soeur_pas_d_allow_origin_double():
|
|
"""Sœur déjà dans la allowlist globale : pas de doublon non plus."""
|
|
interne = _application_interne([
|
|
(b"access-control-allow-origin", b"https://og-dev.dracodev.net"),
|
|
(b"access-control-allow-credentials", b"true"),
|
|
(b"vary", b"Origin")])
|
|
recus = _lancer(rc.RechercheCorsMiddleware(interne),
|
|
_scope("GET", "https://og-dev.dracodev.net"))
|
|
entetes = recus[0]["headers"]
|
|
assert _compter(entetes, "access-control-allow-origin") == 1
|
|
assert _compter(entetes, "access-control-allow-credentials") == 1
|
|
assert _compter(entetes, "vary") == 1
|
|
|
|
|
|
def test_origine_non_autorisee_traverse_sans_modification():
|
|
"""Une origine refusée ne reçoit aucun Allow-Origin ajouté."""
|
|
interne = _application_interne([
|
|
(b"access-control-allow-credentials", b"true")])
|
|
recus = _lancer(rc.RechercheCorsMiddleware(interne),
|
|
_scope("GET", "https://example.com"))
|
|
entetes = recus[0]["headers"]
|
|
assert _compter(entetes, "access-control-allow-origin") == 0
|
|
assert entetes == [(b"access-control-allow-credentials", b"true")]
|
|
|
|
|
|
def test_preflight_agent_dev_toujours_unique():
|
|
"""Le preflight (court-circuit avant l'interne) reste propre."""
|
|
recus = _lancer(rc.RechercheCorsMiddleware(_application_interne([])),
|
|
_scope("OPTIONS", "https://agent-dev.dracodev.net"))
|
|
entetes = recus[0]["headers"]
|
|
assert recus[0]["status"] == 204
|
|
assert _compter(entetes, "access-control-allow-origin") == 1
|
|
assert _compter(entetes, "access-control-allow-credentials") == 1
|