diff --git a/tests/test_hub_roles.py b/tests/test_hub_roles.py new file mode 100644 index 0000000..0db4a53 --- /dev/null +++ b/tests/test_hub_roles.py @@ -0,0 +1,375 @@ +"""HUB-19 (0.46.2) — rôles du Hub propagés à Kanbante (contrat D3(a)). + +Portage du pilote ShaarIt 0.7.0. Couvert ici : + +- client ``app.hub_roles`` : appel ``GET {hub}/api/roles/kanbante`` + (Cookie complet relayé, ``Accept: application/json``), URL du + Hub résolue comme la barre (explicite > hôte de la requête > + défaut officiel), cache 60 s par ``sub``, horodatage posé avant + l'appel (échec = ancien cache conservé, pas de rafale), tout + échec = rôle inconnu (``None``) ; +- pont ``/auth/draco`` : la consultation précède la résolution et + la création JIT — rôle connu ``aucun`` → refus explicite 403, + **aucune session ouverte, aucun compte créé**, aucune liaison à + une session native préexistante ; +- ``/admin`` : session établie par draco + rôle connu ≠ + ``administrateur`` → refusé malgré le rôle local admin ; rôle + ``administrateur`` → accepté ; rôle inconnu → repli local ; + session **native** (même admin, même cookie draco présent) → + règles locales inchangées. +""" + +import base64 +import hashlib +import hmac +import json +import time + +import httpx +import pytest +from fastapi.testclient import TestClient + +import app.main as main +from app import draco, hub_roles, models + +SECRET = "secret-hub19-partage" + +SUBS_TEST = ( + "hub19-refuse", "hub19-membre", "hub19-chef", "hub19-inconnu", + "hub19-sans-liaison", +) + + +def _jeton(sub, nom="Alice Exemple", secret=SECRET): + def b64(d: bytes) -> str: + return base64.urlsafe_b64encode(d).rstrip(b"=").decode() + + header = b64(json.dumps({"alg": "HS256", "typ": "JWT"}).encode()) + payload = b64(json.dumps({ + "sub": sub, "name": nom, "iat": int(time.time()), + "exp": int(time.time()) + 3600, "jti": f"jti-{sub}", + }).encode()) + sig = hmac.new(secret.encode(), f"{header}.{payload}".encode(), + hashlib.sha256).digest() + return f"{header}.{payload}.{b64(sig)}" + + +def _sub_depuis_cookie(cookie: str) -> str: + for morceau in (cookie or "").split(";"): + nom, _, valeur = morceau.strip().partition("=") + if nom == "draco_session" and valeur.count(".") == 2: + segment = valeur.split(".")[1] + segment += "=" * (-len(segment) % 4) + try: + return json.loads(base64.urlsafe_b64decode(segment))["sub"] + except Exception: + return "" + return "" + + +class _Reponse: + def __init__(self, status_code=200, donnees=None): + self.status_code = status_code + self._donnees = donnees + + def json(self): + if isinstance(self._donnees, Exception): + raise self._donnees + return self._donnees + + +@pytest.fixture(autouse=True) +def _cache_propre(): + hub_roles._reset_cache() + yield + hub_roles._reset_cache() + + +@pytest.fixture() +def faux_hub(monkeypatch): + """Faux registre du Hub : rôles par ``sub``, lu dans le cookie relayé.""" + etat = {"roles": {}, "appels": [], "mode": "ok"} + + def faux_get(url, headers=None, timeout=None): + etat["appels"].append( + {"url": url, "headers": dict(headers or {}), "timeout": timeout} + ) + if etat["mode"] == "panne": + raise httpx.ConnectError("Hub injoignable (simulation)") + if etat["mode"] == "erreur500": + return _Reponse(500, {"detail": "erreur"}) + sub = _sub_depuis_cookie((headers or {}).get("Cookie", "")) + role = etat["roles"].get(sub, "utilisateur") + return _Reponse(200, { + "app": "kanbante", "sub": sub, "role": role, + "explicite": True, "ttl_secondes": 60, + }) + + monkeypatch.setattr(httpx, "get", faux_get) + return etat + + +# ── Client : contrat d'appel et résolution d'URL ───────────── + +def test_client_appel_contrat_et_cache(faux_hub): + faux_hub["roles"]["sub-a"] = "administrateur" + cookie = f"draco_session={_jeton('sub-a')}; autre=1" + assert hub_roles.role_pour(cookie, "sub-a") == "administrateur" + # Deuxième appel dans le TTL : servi par le cache, aucun appel. + assert hub_roles.role_pour(cookie, "sub-a") == "administrateur" + assert len(faux_hub["appels"]) == 1 + appel = faux_hub["appels"][0] + assert appel["url"] == "https://hub.dracodev.net/api/roles/kanbante" + assert appel["headers"]["Cookie"] == cookie + assert appel["headers"]["Accept"] == "application/json" + assert appel["timeout"] == 2.0 + + +def test_client_url_hub_par_hote_dev(faux_hub): + hub_roles.role_pour( + f"draco_session={_jeton('sub-b')}", "sub-b", + hote="kanbante-dev.dracodev.net", + ) + assert faux_hub["appels"][0]["url"] == ( + "https://hub-dev.dracodev.net/api/roles/kanbante" + ) + + +def test_client_url_hub_explicite_prioritaire(faux_hub, monkeypatch): + monkeypatch.setenv("DRACO_HUB_URL", "https://hub.exemple.test/") + hub_roles.role_pour( + f"draco_session={_jeton('sub-c')}", "sub-c", + hote="kanbante-dev.dracodev.net", + ) + assert faux_hub["appels"][0]["url"] == ( + "https://hub.exemple.test/api/roles/kanbante" + ) + + +def test_client_ttl_expire_reconsulte(faux_hub): + cookie = f"draco_session={_jeton('sub-d')}" + faux_hub["roles"]["sub-d"] = "utilisateur" + assert hub_roles.role_pour(cookie, "sub-d") == "utilisateur" + hub_roles._cache_fetched_at -= hub_roles.CACHE_TTL + 1 + faux_hub["roles"]["sub-d"] = "administrateur" + assert hub_roles.role_pour(cookie, "sub-d") == "administrateur" + assert len(faux_hub["appels"]) == 2 + + +def test_client_echec_inconnu_et_ancien_cache_conserve(faux_hub): + cookie = f"draco_session={_jeton('sub-e')}" + faux_hub["roles"]["sub-e"] = "utilisateur" + assert hub_roles.role_pour(cookie, "sub-e") == "utilisateur" + # TTL expiré, Hub en panne : l'appel tenté rend « inconnu »… + hub_roles._cache_fetched_at -= hub_roles.CACHE_TTL + 1 + faux_hub["mode"] = "panne" + assert hub_roles.role_pour(cookie, "sub-e") is None + appels = len(faux_hub["appels"]) + # …mais l'horodatage posé avant l'appel garde l'ancien cache : + # l'appel suivant (dans le nouveau TTL) le ressert sans rafale. + assert hub_roles.role_pour(cookie, "sub-e") == "utilisateur" + assert len(faux_hub["appels"]) == appels + + +def test_client_reponses_illisibles_inconnues(monkeypatch): + def faux_get(url, headers=None, timeout=None): + return _Reponse(200, {"role": "super-utilisateur"}) + + monkeypatch.setattr(httpx, "get", faux_get) + assert hub_roles.role_pour("c=1", "sub-f") is None + hub_roles._reset_cache() + + def faux_get_403(url, headers=None, timeout=None): + return _Reponse(403, {"role": "aucun"}) + + monkeypatch.setattr(httpx, "get", faux_get_403) + assert hub_roles.role_pour("c=1", "sub-f") is None + + +# ── Intégration : pont draco et garde /admin ───────────────── + +def _nettoyer(): + with models.get_db() as conn: + ids = [ + r[0] for r in conn.execute( + "SELECT id FROM users WHERE draco_sub IN ({})".format( + ",".join("?" * len(SUBS_TEST)) + ), + SUBS_TEST, + ).fetchall() + ] + for uid in ids: + conn.execute("DELETE FROM api_tokens WHERE user_id = ?", (uid,)) + conn.execute( + "DELETE FROM workspace_members WHERE user_id = ?", (uid,) + ) + conn.execute("DELETE FROM users WHERE id = ?", (uid,)) + + +@pytest.fixture() +def client_hub19(monkeypatch, fake_gitea, faux_hub): + monkeypatch.setenv("KANBANTE_AUTH_USER", "bruno") + monkeypatch.setenv("KANBANTE_AUTH_PASSWORD", "motdepasse") + monkeypatch.setenv("DRACO_SESSION_SECRET", SECRET) + monkeypatch.setattr(draco, "_refresh_revocations", lambda: None) + draco._revoked_cache.clear() + _nettoyer() + with TestClient(main.app) as client: + admin = next(u for u in models.list_users() if u["role"] == "admin") + sub_avant = admin.get("draco_sub") + try: + yield client, faux_hub, admin + finally: + models.update_user(admin["id"], draco_sub=sub_avant) + _nettoyer() + draco._revoked_cache.clear() + + +def _neutraliser_amorcage(admin) -> bool: + """Vrai si l'amorçage pur a été neutralisé (témoin lié à l'admin). + + Même patron que tests/test_draco_correspondance.py : les + scénarios JIT exigent que l'unique admin sans liaison ne + capture pas le premier sub de passage. + """ + users = models.list_users() + if len(users) == 1 and users[0]["role"] == "admin" \ + and not users[0].get("draco_sub"): + models.update_user(users[0]["id"], draco_sub="hub19-temoin-amorce") + return True + return False + + +def _pont(client, sub, nom="Alice Exemple"): + client.cookies.set("draco_session", _jeton(sub, nom)) + return client.get("/auth/draco", follow_redirects=False) + + +def test_pont_role_aucun_refuse_sans_creation(client_hub19): + client, hub, _admin = client_hub19 + hub["roles"]["hub19-refuse"] = "aucun" + avant = models.count_users() + r = _pont(client, "hub19-refuse", "Refusée Exemple") + assert r.status_code == 403 + assert "Accès refusé" in r.text + assert "« aucun »" in r.text + # Aucune session Kanbante ouverte par draco… + assert "kanbante_session" not in r.headers.get("set-cookie", "") + # …et aucun compte JIT créé pour ce sub. + assert models.count_users() == avant + assert models.get_user_by_draco("hub19-refuse") is None + + +def test_pont_role_aucun_ne_lie_pas_une_session_native(client_hub19): + client, hub, admin = client_hub19 + hub["roles"]["hub19-sans-liaison"] = "aucun" + # Session native préexistante (format historique, sans marqueur). + client.cookies.set( + "kanbante_session", main._make_session(admin["username"]) + ) + r = _pont(client, "hub19-sans-liaison") + assert r.status_code == 403 + # La consultation a précédé la liaison : rien n'a été lié. + assert models.get_user_by_draco("hub19-sans-liaison") is None + # La session native elle-même n'est pas touchée. + assert client.get("/").status_code == 200 + + +def test_pont_role_utilisateur_ouvre_session_jit(client_hub19): + client, hub, admin = client_hub19 + hub["roles"]["hub19-membre"] = "utilisateur" + pur = _neutraliser_amorcage(admin) + try: + r = _pont(client, "hub19-membre", "Membre Exemple") + assert r.status_code == 303 and r.headers["location"] == "/" + assert "kanbante_session" in r.headers.get("set-cookie", "") + cree = models.get_user_by_draco("hub19-membre") + assert cree is not None and cree["role"] == "member" + assert client.get("/").status_code == 200 + finally: + if pur: + models.update_user(admin["id"], draco_sub=None) + + +def test_pont_role_inconnu_comportement_actuel(client_hub19): + client, hub, admin = client_hub19 + hub["mode"] = "panne" + pur = _neutraliser_amorcage(admin) + try: + r = _pont(client, "hub19-inconnu", "Inconnu Exemple") + assert r.status_code == 303 and r.headers["location"] == "/" + assert models.get_user_by_draco("hub19-inconnu") is not None + finally: + if pur: + models.update_user(admin["id"], draco_sub=None) + + +def _admin_lie_au_pont(client, hub, admin, role_hub): + """Lie le compte admin local au sub chef, puis ouvre sa session par le pont.""" + models.update_user(admin["id"], draco_sub="hub19-chef") + hub["roles"]["hub19-chef"] = role_hub + hub_roles._reset_cache() + r = _pont(client, "hub19-chef", "Chef Exemple") + assert r.status_code == 303 and r.headers["location"] == "/" + return r + + +def test_admin_session_draco_exige_administrateur_du_registre(client_hub19): + client, hub, admin = client_hub19 + _admin_lie_au_pont(client, hub, admin, "utilisateur") + # Rôle local admin, mais registre connu « utilisateur » : refusé. + r = client.get("/admin") + assert r.status_code == 403 + # Registre passé à « administrateur » : accepté. + hub["roles"]["hub19-chef"] = "administrateur" + hub_roles._reset_cache() + r = client.get("/admin") + assert r.status_code == 200 + + +def test_admin_session_draco_role_inconnu_repli_local(client_hub19): + client, hub, admin = client_hub19 + _admin_lie_au_pont(client, hub, admin, "administrateur") + assert client.get("/admin").status_code == 200 + # Hub en panne et cache expiré : rôle inconnu → règles locales. + hub["mode"] = "panne" + hub_roles._cache_fetched_at -= hub_roles.CACHE_TTL + 1 + # L'appel de garde rend « inconnu » (l'ancien cache ne ressert + # qu'aux appels suivants, patron du pilote) : le repli local joue. + r = client.get("/admin") + assert r.status_code == 200 + + +def test_admin_session_native_non_soumise_au_registre(client_hub19): + client, hub, admin = client_hub19 + models.update_user(admin["id"], draco_sub="hub19-chef") + hub["roles"]["hub19-chef"] = "utilisateur" + # Session NATIVE de l'admin (mot de passe / format historique) : + # même avec un cookie draco présent et un rôle registre connu + # « utilisateur », /admin reste régi par les règles locales. + client.cookies.set( + "kanbante_session", main._make_session(admin["username"]) + ) + client.cookies.set("draco_session", _jeton("hub19-chef", "Chef Exemple")) + r = client.get("/admin") + assert r.status_code == 200 + + +def test_marqueur_session_draco_signe(): + """Le marqueur « établie par draco » est couvert par la signature.""" + from starlette.requests import Request + + def requete(valeur): + scope = { + "type": "http", "method": "GET", "path": "/", + "headers": [(b"cookie", f"kanbante_session={valeur}".encode())], + } + return Request(scope) + + natif = main._make_session("bruno") + assert main._session_info(requete(natif)) == ("bruno", False) + via = main._make_session("bruno", via_draco=True) + assert main._session_info(requete(via)) == ("bruno", True) + # Marqueur ajouté à la main sur un jeton natif : signature fausse. + pirate = natif.rsplit(".", 1)[0] + ".draco." + natif.rsplit(".", 1)[1] + assert main._session_info(requete(pirate)) is None