diff --git a/tests/test_coffre_client.py b/tests/test_coffre_client.py new file mode 100644 index 0000000..be498e1 --- /dev/null +++ b/tests/test_coffre_client.py @@ -0,0 +1,227 @@ +"""Tests du client de distribution coffre ObsiGate (Phase 4, issue ObsiGate #22). + +Aucune instance ObsiGate réelle n'est contactée : le « faux endpoint » est +un ``httpx.MockTransport`` local, et toutes les valeurs distribuées dans +ces tests sont fictives (``TEST-VALUE-XXXX``), jamais des secrets réels. +""" + +import hashlib +import hmac +import json + +import httpx +import pytest + +import app.main as main +from app import models +from app.coffre_client import ( + CoffreAccesRefuse, + CoffreConfig, + CoffreContexteManquant, + CoffreIndisponible, + CoffreNonConfigure, + CoffreReferenceInvalide, + distribuer, + distribuer_async, + est_reference, + parse_reference, + usage_distribue, +) + +# Configuration fictive, réservée aux tests. +CONFIG_TEST = CoffreConfig( + base_url="https://obsigate.test", + jeton_consommateur="TEST-CONSUMER-TOKEN", +) +REFERENCE_TEST = "coffre:TEST-SLUG/TEST-NOM" +VALEUR_TEST = "TEST-VALUE-XXXX" + + +def _transport_faux(statut=200, corps=None, verifier=None): + def handler(request: httpx.Request) -> httpx.Response: + if verifier is not None: + verifier(request) + return httpx.Response(statut, json=corps if corps is not None else {}) + + return httpx.MockTransport(handler) + + +def _verifier_requete(request: httpx.Request): + assert request.url.path == "/api/coffres/distribuer" + assert request.headers["Authorization"] == "Bearer TEST-CONSUMER-TOKEN" + corps = json.loads(request.content.decode()) + assert corps["reference"] == REFERENCE_TEST + assert corps["contexte_type"] == "webhook" + assert corps["contexte_id"] == "gitea" + + +@pytest.mark.parametrize( + "reference, attendu", + [ + ("coffre:personnel/gitea", True), + ("coffre:TEST-SLUG/TEST-NOM", True), + ("coffre:a/b/c", False), # un seul « / » après le préfixe + ("coffre:/nom", False), + ("coffre:slug/", False), + ("coffre:slug", False), + ("gitea", False), + ("", False), + (None, False), + ], +) +def test_reference_validation(reference, attendu): + assert est_reference(reference) is attendu + if attendu: + assert parse_reference(reference)[0] + else: + with pytest.raises(CoffreReferenceInvalide): + parse_reference(reference or "") + + +def test_distribuer_succes_sync(): + transport = _transport_faux( + 200, + {"reference": REFERENCE_TEST, "valeur": VALEUR_TEST, "version": 1}, + verifier=_verifier_requete, + ) + assert ( + distribuer(REFERENCE_TEST, "webhook", "gitea", config=CONFIG_TEST, transport=transport) + == VALEUR_TEST + ) + + +async def test_distribuer_succes_async(): + transport = _transport_faux( + 200, + {"reference": REFERENCE_TEST, "valeur": VALEUR_TEST, "version": 1}, + verifier=_verifier_requete, + ) + assert ( + await distribuer_async( + REFERENCE_TEST, "webhook", "gitea", config=CONFIG_TEST, transport=transport + ) + == VALEUR_TEST + ) + + +@pytest.mark.parametrize( + "statut, detail, message_attendu", + [ + (403, "octroi_revoque", "ré-accorder"), + (403, "octroi_expire", "renouveler"), + (403, "octroi_absent", "octroi"), + (401, "consommateur_revoque", "ré-accorder"), + (401, "consommateur_invalide", "invalide"), + (410, "secret_expire", "expiré"), + ], +) +def test_distribuer_refus_sans_repli(statut, detail, message_attendu): + transport = _transport_faux(statut, {"detail": detail}) + with pytest.raises(CoffreAccesRefuse) as excinfo: + distribuer(REFERENCE_TEST, "webhook", "gitea", config=CONFIG_TEST, transport=transport) + assert message_attendu in str(excinfo.value) + assert VALEUR_TEST not in str(excinfo.value) + + +def test_distribuer_indisponible_503_et_reseau(): + transport = _transport_faux(503, {"detail": "kek_absente"}) + with pytest.raises(CoffreIndisponible): + distribuer(REFERENCE_TEST, "webhook", "gitea", config=CONFIG_TEST, transport=transport) + + def _panne(request: httpx.Request): + raise httpx.ConnectError("faux ObsiGate injoignable", request=request) + + with pytest.raises(CoffreIndisponible): + distribuer( + REFERENCE_TEST, "webhook", "gitea", + config=CONFIG_TEST, transport=httpx.MockTransport(_panne), + ) + + +def test_distribuer_sans_valeur_malgre_200(): + transport = _transport_faux(200, {"reference": REFERENCE_TEST}) + with pytest.raises(CoffreIndisponible): + distribuer(REFERENCE_TEST, "webhook", "gitea", config=CONFIG_TEST, transport=transport) + + +def test_distribuer_reference_invalide_sans_appel(): + def _interdit(request: httpx.Request): + raise AssertionError("aucun appel HTTP ne doit partir") + + with pytest.raises(CoffreReferenceInvalide): + distribuer("valeur-brute", "webhook", "gitea", + config=CONFIG_TEST, transport=httpx.MockTransport(_interdit)) + + +def test_distribuer_contexte_manquant_sans_appel(): + with pytest.raises(CoffreContexteManquant): + distribuer(REFERENCE_TEST, "", "gitea", config=CONFIG_TEST) + + +def test_distribuer_non_configure(monkeypatch): + monkeypatch.delenv("KANBANTE_OBSIGATE_URL", raising=False) + monkeypatch.delenv("KANBANTE_COFFRE_CONSOMMATEUR_TOKEN", raising=False) + with pytest.raises(CoffreNonConfigure): + distribuer(REFERENCE_TEST, "webhook", "gitea") + + +def test_usage_distribue_ne_conserve_que_le_resultat(): + transport = _transport_faux(200, {"valeur": VALEUR_TEST}) + # Le bloc ne sert qu'à l'usage : ici, dériver un HMAC, comme le ferait + # la vérification d'un webhook. Seul le résultat de l'usage est + # conservé après le bloc, jamais la valeur distribuée. + with usage_distribue( + REFERENCE_TEST, "webhook", "gitea", + config=CONFIG_TEST, transport=transport, + ) as valeur: + signature = hmac.new(valeur.encode(), b"corps-test", hashlib.sha256).hexdigest() + assert len(signature) == 64 + + +def test_webhook_reference_octroi_revoque_sans_repli(client, monkeypatch): + """Référence webhook + octroi révoqué -> 401 explicite, sans repli.""" + models.create_connection( + "coffre-webhook", "gitea", "Coffre webhook", "https://gitea.test", + token_encrypted="", account_login="", webhook_secret=REFERENCE_TEST, + ) + + async def _refuse(*args, **kwargs): + raise CoffreAccesRefuse("Accès révoqué dans ObsiGate pour cette référence.") + + monkeypatch.setattr(main.coffre_client, "distribuer_async", _refuse) + reponse = client.post( + "/webhooks/coffre-webhook", + content=b"{}", + headers={"x-gitea-signature": "0" * 64}, + ) + assert reponse.status_code == 401 + assert "ObsiGate" in reponse.json()["error"] + + +def test_webhook_reference_distribuee_puis_oubliee(client, monkeypatch): + """Référence webhook distribuée -> HMAC vérifié, valeur non persistée.""" + models.create_connection( + "coffre-webhook-ok", "gitea", "Coffre webhook OK", "https://gitea.test", + token_encrypted="", account_login="", webhook_secret=REFERENCE_TEST, + ) + + async def _distribue(reference, contexte_type, contexte_id): + assert reference == REFERENCE_TEST + assert contexte_type == "webhook" + assert contexte_id == "coffre-webhook-ok" + return VALEUR_TEST + + monkeypatch.setattr(main.coffre_client, "distribuer_async", _distribue) + corps = b"{}" + signature = hmac.new(VALEUR_TEST.encode(), corps, hashlib.sha256).hexdigest() + reponse = client.post( + "/webhooks/coffre-webhook-ok", + content=corps, + headers={"x-gitea-signature": signature}, + ) + assert reponse.status_code == 200 + # La valeur distribuée n'a jamais été persistée : la ligne stocke + # toujours la référence, jamais la valeur fictive. + ligne = models.get_connection_by_slug("coffre-webhook-ok") + assert ligne["webhook_secret"] == REFERENCE_TEST + assert VALEUR_TEST not in (ligne["webhook_secret"] or "")