flowdeck/tests/test_hub_roles.py

"""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