294 lines
11 KiB
Python
294 lines
11 KiB
Python
"""Fusion #145 (0.46.5) — adoption CONDITIONNÉE du compte d'amorçage.
|
|
|
|
Avant ce correctif, sur une instance fraîche, le premier visiteur
|
|
du pont draco adoptait le compte administrateur d'amorçage
|
|
(« liaison au premier passage ») — un simple utilisateur de la
|
|
suite (« dan », audit « compte partout » du 2026-11-06) l'a fait
|
|
en pile de contrôle. Désormais, l'adoption n'a lieu que si :
|
|
|
|
(a) le ``sub`` entrant est « administrateur » pour kanbante au
|
|
registre des rôles du Hub (repli propre : Hub muet = rôle
|
|
inconnu = PAS d'adoption) ;
|
|
(b) ou l'adoption est initiée depuis une session native déjà
|
|
authentifiée (liaison explicite par l'utilisateur connecté).
|
|
|
|
Couvert ici, sur instance RÉELLEMENT fraîche (base SQLite
|
|
temporaire propre à chaque test : au démarrage, seul le compte
|
|
d'amorçage créé depuis l'environnement existe, sans liaison) :
|
|
|
|
(i) sub simple utilisateur → compte membre JIT propre,
|
|
amorçage intact et toujours sans liaison ;
|
|
(ii) sub administrateur au registre → adoption de l'amorçage ;
|
|
(ii-bis) Hub muet (rôle inconnu) → pas d'adoption, JIT membre ;
|
|
(iii) liaison existante (cas de production) → résolution
|
|
inchangée, quel que soit le rôle du registre ;
|
|
(iv) session native authentifiée → liaison explicite de
|
|
l'amorçage possible (pont et résolution directe) ;
|
|
|
|
plus la part Kanbante de la fusion #144 : l'échec du formulaire
|
|
local suggère la voie « Se connecter avec ObsiGate (SSO) ».
|
|
|
|
Le faux Hub suit le patron de tests/test_hub_roles.py : l'appel
|
|
``httpx.get`` du client du registre est simulé, les rôles sont
|
|
lus dans le cookie relayé.
|
|
"""
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import time
|
|
|
|
import httpx
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
import app.main as main
|
|
from app import accounts, draco, hub_roles, models
|
|
|
|
SECRET = "secret-amorcage-145"
|
|
|
|
SUB_SIMPLE = "fus145-simple"
|
|
SUB_ADMIN = "fus145-admin"
|
|
SUB_INCONNU = "fus145-inconnu"
|
|
SUB_LIE = "fus145-deja-lie"
|
|
SUB_NATIF = "fus145-natif"
|
|
SUB_NATIF_DIRECT = "fus145-natif-direct"
|
|
|
|
|
|
def _jeton(sub, nom="Visiteur 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 _sub_depuis_cookie(cookie: str) -> str:
|
|
for morceau in (cookie or "").split(";"):
|
|
nom, _, valeur = morceau.strip().partition("=")
|
|
if nom == "draco_session" and valeur.count(".") == 2:
|
|
segment = valeur.split(".")[1]
|
|
segment += "=" * (-len(segment) % 4)
|
|
try:
|
|
return json.loads(base64.urlsafe_b64decode(segment))["sub"]
|
|
except Exception:
|
|
return ""
|
|
return ""
|
|
|
|
|
|
class _Reponse:
|
|
def __init__(self, status_code=200, donnees=None):
|
|
self.status_code = status_code
|
|
self._donnees = donnees
|
|
|
|
def json(self):
|
|
return self._donnees
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _cache_propre():
|
|
hub_roles._reset_cache()
|
|
yield
|
|
hub_roles._reset_cache()
|
|
|
|
|
|
@pytest.fixture()
|
|
def faux_hub(monkeypatch):
|
|
"""Faux registre du Hub : rôles par ``sub``, lus dans le cookie relayé."""
|
|
etat = {"roles": {}, "mode": "ok"}
|
|
|
|
def faux_get(url, headers=None, timeout=None):
|
|
if etat["mode"] == "panne":
|
|
raise httpx.ConnectError("Hub injoignable (simulation)")
|
|
sub = _sub_depuis_cookie((headers or {}).get("Cookie", ""))
|
|
role = etat["roles"].get(sub, "utilisateur")
|
|
return _Reponse(200, {
|
|
"app": "kanbante", "sub": sub, "role": role,
|
|
"explicite": True, "ttl_secondes": 60,
|
|
})
|
|
|
|
monkeypatch.setattr(httpx, "get", faux_get)
|
|
return etat
|
|
|
|
|
|
@pytest.fixture()
|
|
def instance_fraiche(monkeypatch, tmp_path, fake_gitea, faux_hub):
|
|
"""Instance neuve : base temporaire, amorçage seul et non lié.
|
|
|
|
Au démarrage (TestClient), ``ensure_bootstrap_admin`` crée
|
|
l'unique compte « proprio » (admin) depuis l'environnement —
|
|
c'est exactement l'état de l'audit : un seul compte, aucune
|
|
liaison draco.
|
|
"""
|
|
monkeypatch.setenv("KANBANTE_DB_PATH", str(tmp_path / "fraiche.db"))
|
|
monkeypatch.setenv("KANBANTE_AUTH_USER", "proprio")
|
|
monkeypatch.setenv("KANBANTE_AUTH_PASSWORD", "mdp-amorce")
|
|
monkeypatch.setenv("DRACO_SESSION_SECRET", SECRET)
|
|
monkeypatch.setattr(draco, "_refresh_revocations", lambda: None)
|
|
draco._revoked_cache.clear()
|
|
with TestClient(main.app) as client:
|
|
users = models.list_users()
|
|
assert len(users) == 1, users
|
|
assert users[0]["username"] == "proprio"
|
|
assert users[0]["role"] == "admin"
|
|
assert not users[0].get("draco_sub")
|
|
yield client, faux_hub, users[0]
|
|
draco._revoked_cache.clear()
|
|
|
|
|
|
def _pont(client, sub, nom="Visiteur 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
|
|
return r.text.split("page-sub\">", 1)[1].split(" ·", 1)[0].strip()
|
|
|
|
|
|
def _csrf_via_login(client) -> str:
|
|
client.get("/login")
|
|
return client.cookies.get("kanbante_csrf") or ""
|
|
|
|
|
|
# ── (i) Simple utilisateur : JIT membre, amorçage intact ─────
|
|
|
|
def test_simple_utilisateur_jit_membre_amorcage_intact(instance_fraiche):
|
|
client, hub, amorce = instance_fraiche
|
|
hub["roles"][SUB_SIMPLE] = "utilisateur"
|
|
r = _pont(client, SUB_SIMPLE, "Dan Simple")
|
|
assert r.status_code == 303 and r.headers["location"] == "/"
|
|
assert "kanbante_session" in r.headers.get("set-cookie", "")
|
|
# L'amorçage est intact : toujours là, admin, TOUJOURS sans liaison.
|
|
apres = models.get_user(amorce["id"])
|
|
assert apres["role"] == "admin"
|
|
assert not apres.get("draco_sub")
|
|
# Le visiteur a reçu SON compte : membre, distinct, lié à son sub.
|
|
cree = models.get_user_by_draco(SUB_SIMPLE)
|
|
assert cree is not None and cree["id"] != amorce["id"]
|
|
assert cree["role"] == "member"
|
|
assert models.count_users() == 2
|
|
assert _profil_username(client) == cree["username"]
|
|
|
|
|
|
# ── (ii) Administrateur au registre : adoption ───────────────
|
|
|
|
def test_administrateur_registre_adopte_amorcage(instance_fraiche):
|
|
client, hub, amorce = instance_fraiche
|
|
hub["roles"][SUB_ADMIN] = "administrateur"
|
|
r = _pont(client, SUB_ADMIN, "Chef Suite")
|
|
assert r.status_code == 303 and r.headers["location"] == "/"
|
|
# Aucun compte créé : le sub s'est lié au compte d'amorçage.
|
|
assert models.count_users() == 1
|
|
lie = models.get_user_by_draco(SUB_ADMIN)
|
|
assert lie is not None and lie["id"] == amorce["id"]
|
|
assert lie["role"] == "admin"
|
|
assert _profil_username(client) == "proprio"
|
|
|
|
|
|
def test_hub_muet_pas_d_adoption(instance_fraiche):
|
|
"""Repli propre : rôle inconnu (Hub injoignable) = pas d'adoption."""
|
|
client, hub, amorce = instance_fraiche
|
|
hub["mode"] = "panne"
|
|
r = _pont(client, SUB_INCONNU, "Inconnu Exemple")
|
|
assert r.status_code == 303 and r.headers["location"] == "/"
|
|
assert not models.get_user(amorce["id"]).get("draco_sub")
|
|
cree = models.get_user_by_draco(SUB_INCONNU)
|
|
assert cree is not None and cree["role"] == "member"
|
|
assert cree["id"] != amorce["id"]
|
|
|
|
|
|
# ── (iii) Liaison existante : résolution inchangée ───────────
|
|
|
|
def test_liaison_existante_resolution_inchangee(instance_fraiche):
|
|
client, hub, amorce = instance_fraiche
|
|
# Cas de production : l'amorçage est DÉJÀ lié (à Bruno).
|
|
models.update_user(amorce["id"], draco_sub=SUB_LIE)
|
|
# Même avec un rôle moindre au registre, la correspondance
|
|
# existante désigne son compte, comme avant le correctif.
|
|
hub["roles"][SUB_LIE] = "utilisateur"
|
|
claims = {"sub": SUB_LIE, "name": "Bruno"}
|
|
resolu = accounts.resolve_draco_user(claims)
|
|
assert resolu is not None and resolu["id"] == amorce["id"]
|
|
r = _pont(client, SUB_LIE, "Bruno")
|
|
assert r.status_code == 303 and r.headers["location"] == "/"
|
|
assert models.count_users() == 1
|
|
assert _profil_username(client) == "proprio"
|
|
|
|
|
|
# ── (iv) Session native authentifiée : liaison explicite ─────
|
|
|
|
def test_session_native_lie_explicitement_via_le_pont(instance_fraiche):
|
|
client, hub, amorce = instance_fraiche
|
|
hub["roles"][SUB_NATIF] = "utilisateur"
|
|
# Connexion native de l'amorçage (mot de passe du compte).
|
|
token = _csrf_via_login(client)
|
|
r = client.post(
|
|
"/login",
|
|
data={"username": "proprio", "password": "mdp-amorce",
|
|
"next": "/", "csrf_token": token},
|
|
follow_redirects=False,
|
|
)
|
|
assert r.status_code == 303
|
|
assert "kanbante_session" in r.headers.get("set-cookie", "")
|
|
# Le passage du pont, session native ouverte, lie le sub libre
|
|
# au compte de la session — sans rôle administrateur au registre.
|
|
r = _pont(client, SUB_NATIF, "Proprio Lui-Même")
|
|
assert r.status_code == 303
|
|
lie = models.get_user_by_draco(SUB_NATIF)
|
|
assert lie is not None and lie["id"] == amorce["id"]
|
|
assert models.count_users() == 1
|
|
|
|
|
|
def test_resolution_directe_avec_session_native(instance_fraiche):
|
|
_client, _hub, amorce = instance_fraiche
|
|
claims = {"sub": SUB_NATIF_DIRECT, "name": "Proprio"}
|
|
# Sans drapeau : aucune adoption (résolution « à froid »).
|
|
assert accounts.resolve_draco_user(claims) is None
|
|
assert not models.get_user(amorce["id"]).get("draco_sub")
|
|
# Avec le drapeau « session native authentifiée » : adoption.
|
|
resolu = accounts.resolve_draco_user(
|
|
claims, liaison_session_native=True
|
|
)
|
|
assert resolu is not None and resolu["id"] == amorce["id"]
|
|
assert models.get_user_by_draco(SUB_NATIF_DIRECT)["id"] == amorce["id"]
|
|
|
|
|
|
# ── Fusion #144 : l'échec du formulaire suggère le SSO ───────
|
|
|
|
def test_echec_login_identifiant_sans_compte_local_suggere_sso(instance_fraiche):
|
|
client, _hub, _amorce = instance_fraiche
|
|
token = _csrf_via_login(client)
|
|
r = client.post(
|
|
"/login",
|
|
data={"username": "compte-du-hub", "password": "peu-importe",
|
|
"next": "/", "csrf_token": token},
|
|
follow_redirects=False,
|
|
)
|
|
assert r.status_code == 401
|
|
assert "aucun compte local" in r.text
|
|
assert "Se connecter avec ObsiGate (SSO)" in r.text
|
|
assert "kanbante_session" not in r.headers.get("set-cookie", "")
|
|
|
|
|
|
def test_echec_login_mauvais_mot_de_passe_suggere_sso(instance_fraiche):
|
|
client, _hub, _amorce = instance_fraiche
|
|
token = _csrf_via_login(client)
|
|
r = client.post(
|
|
"/login",
|
|
data={"username": "proprio", "password": "mauvais",
|
|
"next": "/", "csrf_token": token},
|
|
follow_redirects=False,
|
|
)
|
|
assert r.status_code == 401
|
|
assert "Identifiants incorrects" in r.text
|
|
assert "Se connecter avec ObsiGate (SSO)" in r.text
|