Kanbante 0.46.2 : tests rôles Hub (client, cache/TTL, refus sans création, /admin gardé, session native non soumise) (tests/test_hub_roles.py)

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