diff --git a/tests/test_draco_correspondance.py b/tests/test_draco_correspondance.py new file mode 100644 index 0000000..2460dd4 --- /dev/null +++ b/tests/test_draco_correspondance.py @@ -0,0 +1,396 @@ +"""HUB-16 (0.46.0) — correspondance draco PAR UTILISATEUR. + +Le pont SSO ne désigne plus « l'opérateur » : chaque identité +draco (claim ``sub``) désigne SON compte Kanbante, via la colonne +users.draco_sub. Couvert ici : + +- amorçage : l'unique compte admin sans liaison se lie au + premier passage (le compte existant garde son accès) ; +- JIT : deux identités distinctes → deux comptes distincts, + rôle membre, mot de passe local non divulgué/inutilisable ; +- le JIT n'adopte jamais un compte préexistant non lié, même + homonyme : il crée un compte nouveau, suffixé ; +- liaison / déliaison par l'administrateur depuis /admin, + conflit de sub refusé ; +- compte désactivé : correspondance refusée, aucun contournement ; +- /api/search : un jeton draco sans correspondance ne suffit plus, + un jeton lié passe ; +- non-régression : un jeton personnel kbt_ reste accepté sur les + surfaces machine pendant que l'authentification est active. + +La base de test est partagée par toute la session (voir +conftest) : les comptes créés ici portent des noms/subs uniques +« hub16-… » et sont supprimés au teardown de chaque test. +""" + +import base64 +import hashlib +import hmac +import json +import time + +import pytest +from fastapi.testclient import TestClient + +import app.main as main +from app import accounts, draco, models + +SECRET = "secret-partage-hub16" + +SUBS_TEST = ( + "hub16-alice", "hub16-bob", "hub16-carole", "hub16-dave", + "hub16-erin", "hub16-franck", "hub16-amorce", "hub16-sans-lien", + "hub16-recherche", +) + + +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 _nettoyer(): + """Supprime les comptes créés par ces tests (et leurs traces).""" + 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() + ] + ids += [ + r[0] for r in conn.execute( + "SELECT id FROM users WHERE username IN" + " ('hub16-dave', 'hub16-erin', 'carole-exemple')" + ).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,)) + + +def _restaurer_temoin(): + """Délie le compte témoin d'amorçage (jamais supprimé).""" + temoin = models.get_user_by_draco("hub16-temoin-amorce") + if temoin is not None: + models.update_user(temoin["id"], draco_sub=None) + + +@pytest.fixture() +def client_hub16(monkeypatch, fake_gitea): + 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() + _restaurer_temoin() + _nettoyer() + with TestClient(main.app) as client: + yield client + _restaurer_temoin() + _nettoyer() + draco._revoked_cache.clear() + + +def _pont(client, sub, nom="Alice Exemple"): + client.cookies.set("draco_session", _jeton(sub, nom)) + return client.get("/auth/draco", follow_redirects=False) + + +def _profil_username(client) -> str: + r = client.get("/profile") + assert r.status_code == 200 + # « · rôle : … » dans l'en-tête de la page profil. + return r.text.split("page-sub\">", 1)[1].split(" ·", 1)[0].strip() + + +def _unique_admin_sans_liaison(): + """(compte, vrai) si l'état partagé est le cas d'amorçage pur.""" + users = models.list_users() + if len(users) == 1 and users[0]["role"] == "admin" \ + and not users[0].get("draco_sub"): + return users[0], True + return None, False + + +# ── Amorçage ────────────────────────────────────────────────── + +def test_amorcage_unique_admin_se_lie_au_premier_passage(client_hub16): + admin, pur = _unique_admin_sans_liaison() + if not pur: + pytest.skip("base partagée : le cas d'amorçage pur n'est pas en place") + avant = models.count_users() + r = _pont(client_hub16, "hub16-amorce", "Bruno") + assert r.status_code == 303 and r.headers["location"] == "/" + assert "kanbante_session" in r.headers.get("set-cookie", "") + # Aucun compte créé : le sub s'est lié au compte admin existant. + assert models.count_users() == avant + lie = models.get_user_by_draco("hub16-amorce") + assert lie is not None and lie["id"] == admin["id"] + assert lie["role"] == "admin" + assert _profil_username(client_hub16) == admin["username"] + # Nettoyage propre à ce test : rendre l'amorçage à son état. + models.update_user(admin["id"], draco_sub=None) + + +# ── Création JIT ────────────────────────────────────────────── + +def test_jit_deux_identites_deux_comptes_distincts(client_hub16): + admin, pur = _unique_admin_sans_liaison() + if pur: + # Neutraliser l'amorçage : les scénarios JIT doivent être purs. + models.update_user(admin["id"], draco_sub="hub16-temoin-amorce") + try: + r = _pont(client_hub16, "hub16-alice", "Alice Exemple") + assert r.status_code == 303 and r.headers["location"] == "/" + assert _profil_username(client_hub16) == "alice-exemple" + alice = models.get_user_by_draco("hub16-alice") + assert alice is not None + assert alice["username"] == "alice-exemple" + assert alice["display_name"] == "Alice Exemple" + assert alice["role"] == "member" + # Mot de passe local aléatoire : ni le mot de passe + # d'environnement ni le nom ne l'ouvrent. + assert accounts.verify_password("motdepasse", alice["password_hash"]) is False + assert alice["password_hash"].startswith("pbkdf2$") + + client_hub16.cookies.delete("kanbante_session") + r = _pont(client_hub16, "hub16-bob", "Bob Exemple") + assert r.status_code == 303 and r.headers["location"] == "/" + assert _profil_username(client_hub16) == "bob-exemple" + bob = models.get_user_by_draco("hub16-bob") + assert bob is not None and bob["id"] != alice["id"] + + # Deuxième passage d'Alice : MÊME compte, pas de doublon. + client_hub16.cookies.delete("kanbante_session") + avant = models.count_users() + _pont(client_hub16, "hub16-alice", "Alice Exemple") + assert models.count_users() == avant + assert models.get_user_by_draco("hub16-alice")["id"] == alice["id"] + assert _profil_username(client_hub16) == "alice-exemple" + finally: + if pur: + models.update_user(admin["id"], draco_sub=None) + + +def test_jit_n_adopte_pas_un_compte_homonyme_non_lie(client_hub16): + admin, pur = _unique_admin_sans_liaison() + if pur: + models.update_user(admin["id"], draco_sub="hub16-temoin-amorce") + try: + local_id = models.create_user( + "carole-exemple", accounts.hash_password("local-carole"), + role="member", display_name="Carole Exemple", + ) + r = _pont(client_hub16, "hub16-carole", "Carole Exemple") + assert r.status_code == 303 and r.headers["location"] == "/" + # Le compte préexistant n'a PAS été lié ni utilisé… + local = models.get_user(local_id) + assert local["draco_sub"] is None + # …un compte nouveau, suffixé, porte la correspondance. + cree = models.get_user_by_draco("hub16-carole") + assert cree is not None and cree["id"] != local_id + assert cree["username"] == "carole-exemple-2" + assert _profil_username(client_hub16) == "carole-exemple-2" + finally: + if pur: + models.update_user(admin["id"], draco_sub=None) + + +def test_compte_desactive_refuse_sans_contournement(client_hub16): + admin, pur = _unique_admin_sans_liaison() + if pur: + models.update_user(admin["id"], draco_sub="hub16-temoin-amorce") + try: + _pont(client_hub16, "hub16-franck", "Franck Exemple") + franck = models.get_user_by_draco("hub16-franck") + assert franck is not None + models.update_user(franck["id"], active=0) + client_hub16.cookies.delete("kanbante_session") + avant = models.count_users() + r = _pont(client_hub16, "hub16-franck", "Franck Exemple") + # Refusé au login, aucune session, aucun nouveau compte. + assert r.headers["location"] == "/login" + assert "kanbante_session" not in r.headers.get("set-cookie", "") + assert models.count_users() == avant + finally: + if pur: + models.update_user(admin["id"], draco_sub=None) + + +# ── Liaison par l'administrateur ────────────────────────────── + +def _jeton_csrf_via_login(client) -> str: + """Charge la page de connexion (elle pose le cookie CSRF). + + csrf_from() charge « / », qui répond 401 JSON sans cookie + quand l'authentification est active et qu'aucune session + n'est ouverte — la page /login, elle, est rendue (render() + pose le cookie) dans tous les cas. + """ + client.get("/login") + return client.cookies.get("kanbante_csrf") or "" + + +def _connecter_admin(client): + token = _jeton_csrf_via_login(client) + r = client.post( + "/login", + data={"username": "bruno", "password": "motdepasse", + "next": "/", "csrf_token": token}, + follow_redirects=False, + ) + assert r.status_code == 303, r.text + assert "kanbante_session" in r.headers.get("set-cookie", "") + + +def test_liaison_par_administrateur(client_hub16): + dave_id = models.create_user( + "hub16-dave", accounts.hash_password("local-dave"), + role="member", display_name="Dave", + ) + erin_id = models.create_user( + "hub16-erin", accounts.hash_password("local-erin"), + role="member", display_name="Erin", + ) + _connecter_admin(client_hub16) + # L'écran d'administration expose le mécanisme de liaison. + r = client_hub16.get("/admin") + assert r.status_code == 200 + assert "/draco" in r.text and "Lier" in r.text + + # Lier Dave au sub hub16-dave. + r = client_hub16.post( + f"/admin/users/{dave_id}/draco", + data={"draco_sub": "hub16-dave", + "csrf_token": client_hub16.cookies.get("kanbante_csrf")}, + follow_redirects=False, + ) + assert r.status_code == 303 and r.headers["location"].startswith("/admin?msg=") + assert models.get_user(dave_id)["draco_sub"] == "hub16-dave" + + # Conflit : le même sub pour Erin est refusé, Erin inchangée. + r = client_hub16.post( + f"/admin/users/{erin_id}/draco", + data={"draco_sub": "hub16-dave", + "csrf_token": client_hub16.cookies.get("kanbante_csrf")}, + follow_redirects=False, + ) + assert r.status_code == 303 and "err=" in r.headers["location"] + assert models.get_user(erin_id)["draco_sub"] is None + + # Le pont route désormais ce sub vers Dave, et seulement lui. + client_hub16.cookies.delete("kanbante_session") + r = _pont(client_hub16, "hub16-dave", "Dave") + assert r.status_code == 303 and r.headers["location"] == "/" + assert _profil_username(client_hub16) == "hub16-dave" + + # Déliaison par l'administrateur (champ vide). + _connecter_admin(client_hub16) + r = client_hub16.post( + f"/admin/users/{dave_id}/draco", + data={"draco_sub": "", + "csrf_token": client_hub16.cookies.get("kanbante_csrf")}, + follow_redirects=False, + ) + assert r.status_code == 303 + assert models.get_user(dave_id)["draco_sub"] is None + + +def test_liaison_refusee_aux_non_admins(client_hub16): + dave_id = models.create_user( + "hub16-dave", accounts.hash_password("local-dave"), + role="member", display_name="Dave", + ) + # Connexion native de Dave (membre) : aucun droit d'administration. + token = _jeton_csrf_via_login(client_hub16) + r = client_hub16.post( + "/login", + data={"username": "hub16-dave", "password": "local-dave", + "next": "/", "csrf_token": token}, + follow_redirects=False, + ) + assert r.status_code == 303 + r = client_hub16.post( + f"/admin/users/{dave_id}/draco", + data={"draco_sub": "hub16-dave", + "csrf_token": client_hub16.cookies.get("kanbante_csrf")}, + follow_redirects=False, + ) + assert r.status_code == 403 + assert models.get_user(dave_id)["draco_sub"] is None + + +# ── /api/search par utilisateur ─────────────────────────────── + +def test_api_search_draco_par_utilisateur(client_hub16): + admin, pur = _unique_admin_sans_liaison() + if pur: + models.update_user(admin["id"], draco_sub="hub16-temoin-amorce") + try: + # Jeton draco valide mais SANS correspondance : refusé + # (il ne désigne plus l'opérateur). + client_hub16.cookies.set( + "draco_session", _jeton("hub16-sans-lien", "Sans Lien") + ) + r = client_hub16.get("/api/search", params={"q": "demo"}) + assert r.status_code == 401 + # Après création JIT par le pont, le même chemin passe. + _pont(client_hub16, "hub16-recherche", "Recherche Exemple") + client_hub16.cookies.delete("kanbante_session") + client_hub16.cookies.set( + "draco_session", _jeton("hub16-recherche", "Recherche Exemple") + ) + r = client_hub16.get("/api/search", params={"q": "demo"}) + assert r.status_code == 200 + assert r.json()["query"] == "demo" + finally: + if pur: + models.update_user(admin["id"], draco_sub=None) + + +# ── Non-régression : jetons kbt_ ────────────────────────────── + +def test_jeton_kbt_toujours_accepte(client_hub16): + uid = models.create_user( + "hub16-erin", accounts.hash_password("local-erin"), + role="member", display_name="Erin Jetons", + ) + user = models.get_user(uid) + _row, clair = models.create_api_token(uid, "service-agents") + assert clair.startswith("kbt_") + corps = {"jsonrpc": "2.0", "id": 1, "method": "initialize", + "params": {"protocolVersion": "2024-11-05"}} + # Sans jeton : 401 ; avec le jeton Bearer : le handshake passe. + assert client_hub16.post("/mcp", json=corps).status_code == 401 + r = client_hub16.post( + "/mcp", json=corps, headers={"Authorization": f"Bearer {clair}"} + ) + assert r.status_code == 200 + assert r.json()["result"]["serverInfo"]["name"] == "kanbante" + # Le jeton sert aussi /api/search (surface machine)… + r = client_hub16.get( + "/api/search", params={"q": "demo"}, + headers={"Authorization": f"Bearer {clair}"}, + ) + assert r.status_code == 200 + # …mais jamais l'interface web. + assert client_hub16.get( + "/", headers={"Authorization": f"Bearer {clair}", + "Accept": "application/json"}, + ).status_code == 401 + assert user["username"] == "hub16-erin" diff --git a/tests/test_draco_sso.py b/tests/test_draco_sso.py index 38930b9..8c13ec3 100644 --- a/tests/test_draco_sso.py +++ b/tests/test_draco_sso.py @@ -2,9 +2,12 @@ Le cookie émis par ObsiGate (ACC-3.1) est vérifié localement (secret partagé, HS256) et converti en session Kanbante native par -GET /auth/draco. Kanbante est mono-utilisateur : un jeton valide -ouvre la session de l'opérateur configuré. Sans -DRACO_SESSION_SECRET, le pont renvoie simplement au login. +GET /auth/draco. Depuis la 0.46.0 (HUB-16), la correspondance est +par utilisateur (users.draco_sub, création JIT) — voir +tests/test_draco_correspondance.py. Les cas couverts ici restent +valides : dans ce module, le seul compte est l'admin d'amorçage, +auquel le premier jeton valide se lie (cas d'amorçage documenté). +Sans DRACO_SESSION_SECRET, le pont renvoie simplement au login. """ import base64