Kanbante 0.46.0 : tests de correspondance par utilisateur (amorçage, JIT x2 identités, homonyme non adopté, liaison admin, désactivé, /api/search, kbt_) (tests/test_draco_correspondance.py, tests/test_draco_sso.py)

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
+    # « <username> · 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"

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