Flowdeck 7.85.2 — HUB-19 : propagation des rôles du Hub (registre D3(a))

tests/test_hub_roles.py

@@ -0,0 +1,308 @@
+"""HUB-19 — propagation des rôles du Hub à Flowdeck (suite du pilote
+ShaarIt 0.7.0) : client ``app.services.hub_roles`` et ses effets.
+
+Contrat D3(a) : à l'ouverture de session draco, consultation de
+``GET {hub}/api/roles/flowdeck`` (cookie complet relayé, délai 2 s),
+cache mémoire par ``sub`` TTL 60 s, horodatage posé avant l'appel.
+Toute erreur = rôle inconnu → comportement actuel conservé.
+
+Effets vérifiés ici :
+- rôle connu ``aucun`` → 403 en français au pont ``GET /auth/draco``,
+  avant la création de la session locale et avant le provisionnement
+  JIT HUB-15 (aucun compte créé pour un ``sub`` refusé) ;
+- session locale préexistante exemptée (accès de secours) ;
+- surfaces d'administration d'instance (``/api/admin/*``) : session
+  draco + rôle connu → ``administrateur`` exigé ;
+- rôle inconnu (Hub injoignable/erreur/rôle hors ensemble) → repli
+  intégral.
+"""
+
+import base64
+import hashlib
+import hmac
+import json
+import time
+from types import SimpleNamespace
+
+import pytest
+from conftest import anon
+
+from app.db import get_conn
+from app.services import draco, hub_roles
+
+SECRET = "secret-partage-de-test"
+
+
+def _jeton(sub="admin-1", secret=SECRET, exp=None, jti="jti-hub19", name="Bruno"):
+    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": name, "iat": int(time.time()),
+        "exp": exp if exp is not None else int(time.time()) + 3600,
+        "jti": jti,
+    }).encode())
+    sig = hmac.new(secret.encode(), f"{header}.{payload}".encode(),
+                   hashlib.sha256).digest()
+    return f"{header}.{payload}.{b64(sig)}"
+
+
+def _requete(cookie_complet="", draco_token=None):
+    cookies = {}
+    headers = {}
+    if draco_token is not None:
+        cookies["draco_session"] = draco_token
+        complet = f"draco_session={draco_token}"
+        headers["cookie"] = complet if not cookie_complet else cookie_complet
+    elif cookie_complet:
+        headers["cookie"] = cookie_complet
+    return SimpleNamespace(cookies=cookies, headers=headers)
+
+
+@pytest.fixture(autouse=True)
+def _caches_propres(monkeypatch):
+    monkeypatch.setenv("DRACO_SESSION_SECRET", SECRET)
+    monkeypatch.setattr(draco, "_refresh_revocations", lambda: None)
+    draco._revoked_cache.clear()
+    hub_roles._cache.clear()
+    hub_roles._cache_fetched_at = 0.0
+    yield
+    draco._revoked_cache.clear()
+    hub_roles._cache.clear()
+    hub_roles._cache_fetched_at = 0.0
+
+
+def _pose_role(monkeypatch, role, compteur=None):
+    """Simule la réponse du Hub pour ``role_pour`` (appel réseau évité)."""
+    def faux_consulter(cookie):
+        if compteur is not None:
+            compteur.append(cookie)
+        return role
+    monkeypatch.setattr(hub_roles, "_consulter_hub", faux_consulter)
+
+
+# ── Client : cache par sub, TTL, horodatage avant l'appel ───────────
+
+def test_role_connu_mis_en_cache_par_sub(monkeypatch):
+    appels = []
+    _pose_role(monkeypatch, "utilisateur", appels)
+    assert hub_roles.role_pour("cookie-a", "sub-1") == "utilisateur"
+    assert hub_roles.role_pour("cookie-a", "sub-1") == "utilisateur"
+    assert len(appels) == 1                      # 2e lecture : cache
+    assert hub_roles.role_pour("cookie-b", "sub-2") == "utilisateur"
+    assert len(appels) == 2                      # autre sub : nouvel appel
+
+
+def test_echec_ancien_cache_conserve_pas_de_rafale(monkeypatch):
+    appels = []
+    _pose_role(monkeypatch, "utilisateur", appels)
+    assert hub_roles.role_pour("c", "sub-1") == "utilisateur"
+    # Fenêtre expirée, Hub désormais en échec : l'ancien cache reste.
+    hub_roles._cache_fetched_at = time.monotonic() - 61
+    _pose_role(monkeypatch, None, appels)
+    assert hub_roles.role_pour("c", "sub-1") == "utilisateur"
+    assert len(appels) == 2
+    # Pas de rafale : dans la fenêtre qui suit l'échec, aucun rappel.
+    assert hub_roles.role_pour("c", "sub-1") == "utilisateur"
+    assert len(appels) == 2
+
+
+def test_echec_sans_cache_role_inconnu(monkeypatch):
+    # Fidèle au pilote : sans entrée en cache pour ce sub, l'échec
+    # rend « inconnu » et la demande suivante retente (la protection
+    # anti-rafale de l'horodatage posé avant l'appel ne sert que les
+    # subs déjà en cache — test précédent).
+    appels = []
+    _pose_role(monkeypatch, None, appels)
+    assert hub_roles.role_pour("c", "sub-x") is None
+    assert hub_roles.role_pour("c", "sub-x") is None
+    assert len(appels) == 2
+
+
+def test_role_sans_sub_inconnu(monkeypatch):
+    _pose_role(monkeypatch, "aucun")
+    assert hub_roles.role_pour("c", "") is None
+
+
+# ── Client : validation de la réponse du Hub ────────────────────────
+
+class _Reponse:
+    def __init__(self, status_code, payload=None):
+        self.status_code = status_code
+        self._payload = payload
+
+    def json(self):
+        if isinstance(self._payload, Exception):
+            raise self._payload
+        return self._payload
+
+
+def _pose_http(monkeypatch, reponse_ou_exception, appels=None):
+    import httpx
+
+    from app.services import draco_urls
+    monkeypatch.setattr(draco_urls, "hub_url", lambda: "https://hub.example")
+
+    def faux_get(url, headers=None, timeout=None):
+        if appels is not None:
+            appels.append((url, headers, timeout))
+        if isinstance(reponse_ou_exception, Exception):
+            raise reponse_ou_exception
+        return reponse_ou_exception
+
+    monkeypatch.setattr(httpx, "get", faux_get)
+
+
+def test_consultation_contrat_exact(monkeypatch):
+    appels = []
+    _pose_http(monkeypatch, _Reponse(200, {
+        "app": "flowdeck", "sub": "sub-1", "role": "administrateur",
+        "explicite": True, "ttl_secondes": 60}), appels)
+    cookie = "draco_session=jeton; autre=1"
+    assert hub_roles._consulter_hub(cookie) == "administrateur"
+    url, headers, timeout = appels[0]
+    assert url == "https://hub.example/api/roles/flowdeck"
+    assert headers["Cookie"] == cookie           # cookie complet relayé
+    assert headers["Accept"] == "application/json"
+    assert timeout == 2.0
+
+
+@pytest.mark.parametrize("reponse", [
+    _Reponse(401, {}), _Reponse(404, {}), _Reponse(500, {}),
+    _Reponse(200, {"role": "superadmin"}),       # hors ensemble
+    _Reponse(200, {"role": None}),
+    _Reponse(200, ValueError("json illisible")),
+    ConnectionError("hub injoignable"),
+])
+def test_consultation_erreurs_role_inconnu(monkeypatch, reponse):
+    _pose_http(monkeypatch, reponse)
+    assert hub_roles._consulter_hub("c") is None
+
+
+# ── role_draco / admin_hub_ok ───────────────────────────────────────
+
+def test_role_draco_sans_cookie_inconnu():
+    assert hub_roles.role_draco(_requete()) is None
+
+
+def test_role_draco_avec_cookie(monkeypatch):
+    _pose_role(monkeypatch, "administrateur")
+    assert hub_roles.role_draco(_requete(draco_token=_jeton())) == "administrateur"
+
+
+def test_admin_hub_ok_sans_draco_comportement_actuel():
+    assert hub_roles.admin_hub_ok(_requete()) is True
+
+
+@pytest.mark.parametrize("role,attendu", [
+    ("administrateur", True), ("utilisateur", False), ("aucun", False),
+    (None, True),                                # inconnu → repli
+])
+def test_admin_hub_ok_selon_role(monkeypatch, role, attendu):
+    _pose_role(monkeypatch, role)
+    assert hub_roles.admin_hub_ok(_requete(draco_token=_jeton())) is attendu
+
+
+# ── Intégration : pont /auth/draco ──────────────────────────────────
+
+def _compte_utilisateurs():
+    with get_conn() as conn:
+        return conn.execute("SELECT COUNT(*) FROM users").fetchone()[0]
+
+
+def _ajoute_deuxieme_utilisateur():
+    with get_conn() as conn:
+        conn.execute(
+            "INSERT INTO users (login, email, password_hash, full_name, is_active) "
+            "VALUES ('deuxieme', '[email protected]', 'x', 'Deuxième', 1)")
+        conn.commit()
+
+
+def test_pont_refus_aucun_403_en_francais(client, monkeypatch):
+    c = anon(client)
+    _pose_role(monkeypatch, "aucun")
+    c.cookies.set("draco_session", _jeton(sub="admin-1"))
+    r = c.get("/auth/draco", follow_redirects=False)
+    assert r.status_code == 403
+    assert "registre des rôles" in r.text and "Hub" in r.text
+    assert "« aucun »" in r.text
+    assert "flowdeck_session" not in r.headers.get("set-cookie", "")
+    assert c.get("/auth/user").json()["authenticated"] is False
+
+
+def test_pont_refus_aucun_avant_provisionnement_jit(client, monkeypatch):
+    c = anon(client)
+    _ajoute_deuxieme_utilisateur()
+    avant = _compte_utilisateurs()
+    _pose_role(monkeypatch, "aucun")
+    c.cookies.set("draco_session", _jeton(sub="og-refuse", name="Refusée"))
+    r = c.get("/auth/draco", follow_redirects=False)
+    assert r.status_code == 403
+    # Ordre vérifié : refus avant JIT — aucun compte, aucun lien créés.
+    assert _compte_utilisateurs() == avant
+    with get_conn() as conn:
+        assert conn.execute(
+            "SELECT COUNT(*) FROM draco_links WHERE sub = 'og-refuse'"
+        ).fetchone()[0] == 0
+
+
+def test_pont_role_utilisateur_jit_inchange(client, monkeypatch):
+    c = anon(client)
+    _ajoute_deuxieme_utilisateur()
+    avant = _compte_utilisateurs()
+    _pose_role(monkeypatch, "utilisateur")
+    c.cookies.set("draco_session", _jeton(sub="og-nouveau", name="Nouvelle"))
+    r = c.get("/auth/draco", follow_redirects=False)
+    assert r.status_code == 302
+    assert r.headers["location"] == "/workspaces"
+    assert _compte_utilisateurs() == avant + 1   # JIT HUB-15 intact
+
+
+def test_pont_role_inconnu_repli_comportement_actuel(client, monkeypatch):
+    c = anon(client)
+    _pose_role(monkeypatch, None)                # Hub injoignable
+    c.cookies.set("draco_session", _jeton(sub="admin-1"))
+    r = c.get("/auth/draco", follow_redirects=False)
+    assert r.status_code == 302
+    assert r.headers["location"] == "/workspaces"
+
+
+def test_pont_session_locale_preexistante_exemptee(client, monkeypatch):
+    # Le client de la fixture porte déjà une session locale (admin de
+    # test) : patron du pilote — l'accès de secours n'est pas refusé.
+    _pose_role(monkeypatch, "aucun")
+    client.cookies.set("draco_session", _jeton(sub="admin-1"))
+    r = client.get("/auth/draco", follow_redirects=False)
+    assert r.status_code == 302
+    assert r.headers["location"] == "/workspaces"
+
+
+# ── Intégration : surfaces d'administration (/api/admin/*) ──────────
+
+def test_admin_instance_role_utilisateur_refuse(client, monkeypatch):
+    _pose_role(monkeypatch, "utilisateur")
+    client.cookies.set("draco_session", _jeton(sub="admin-1"))
+    r = client.get("/api/admin/users")
+    assert r.status_code == 403
+    assert "registre" in r.text and "Hub" in r.text
+
+
+def test_admin_instance_role_administrateur_accepte(client, monkeypatch):
+    _pose_role(monkeypatch, "administrateur")
+    client.cookies.set("draco_session", _jeton(sub="admin-1"))
+    r = client.get("/api/admin/users")
+    assert r.status_code == 200
+
+
+def test_admin_instance_role_inconnu_repli(client, monkeypatch):
+    _pose_role(monkeypatch, None)
+    client.cookies.set("draco_session", _jeton(sub="admin-1"))
+    r = client.get("/api/admin/users")
+    assert r.status_code == 200                  # comportement actuel
+
+
+def test_admin_instance_sans_draco_inchange(client, monkeypatch):
+    _pose_role(monkeypatch, "aucun")             # jamais consulté sans draco
+    r = client.get("/api/admin/users")
+    assert r.status_code == 200