397 lines
16 KiB
Python
397 lines
16 KiB
Python
"""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"
|