332 lines
13 KiB
Python
332 lines
13 KiB
Python
# tests/test_draco_session.py
|
|
# Fusion ACC-3.1 (#216) — émettrice SSO draco_session : émission à la
|
|
# connexion, vérification (signature, expiration, révocation, rotation),
|
|
# renouvellement glissant, return_to en liste blanche, révocations.
|
|
|
|
import time
|
|
|
|
import pytest
|
|
from jose import jwt as jose_jwt
|
|
|
|
from backend import draco_session
|
|
|
|
USER = {"id": "admin-1", "username": "admin", "display_name": "Admin"}
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _draco_env(monkeypatch, tmp_path):
|
|
monkeypatch.setenv("DRACO_SESSION_SECRET", "secret-de-test")
|
|
monkeypatch.setattr(draco_session, "REVOKED_FILE",
|
|
tmp_path / "draco_revoked.json")
|
|
monkeypatch.setattr(draco_session, "ISSUED_FILE",
|
|
tmp_path / "draco_issued.json")
|
|
draco_session._revoked = None
|
|
draco_session._issued = None
|
|
yield
|
|
draco_session._revoked = None
|
|
draco_session._issued = None
|
|
|
|
|
|
# ── Jeton : émission / vérification ─────────────────────────────────────
|
|
|
|
def test_roundtrip_claims_minimaux():
|
|
token, jti, exp = draco_session.issue_token(USER)
|
|
claims = draco_session.verify_token(token)
|
|
assert claims is not None
|
|
assert claims["sub"] == "admin-1"
|
|
assert claims["name"] == "Admin"
|
|
assert claims["jti"] == jti
|
|
assert exp - claims["iat"] == 12 * 3600 # durée de vie du design §5.2
|
|
assert "role" not in claims # aucun rôle dans le jeton
|
|
|
|
|
|
def test_jeton_falsifie_rejete():
|
|
token, _j, _e = draco_session.issue_token(USER)
|
|
assert draco_session.verify_token(token + "x") is None
|
|
autre = jose_jwt.encode({"sub": "admin-1"}, "mauvais-secret",
|
|
algorithm="HS256")
|
|
assert draco_session.verify_token(autre) is None
|
|
|
|
|
|
def test_jeton_expire_rejete():
|
|
now = int(time.time())
|
|
token = jose_jwt.encode(
|
|
{"sub": "admin-1", "name": "Admin", "iat": now - 7200,
|
|
"exp": now - 3600, "jti": "ancien"},
|
|
"secret-de-test", algorithm="HS256")
|
|
assert draco_session.verify_token(token) is None
|
|
|
|
|
|
def test_rotation_secret_precedent_accepte(monkeypatch):
|
|
token = jose_jwt.encode(
|
|
{"sub": "admin-1", "name": "Admin", "iat": int(time.time()),
|
|
"exp": int(time.time()) + 3600, "jti": "rotation"},
|
|
"ancien-secret", algorithm="HS256")
|
|
assert draco_session.verify_token(token) is None # pas encore configuré
|
|
monkeypatch.setenv("DRACO_SESSION_SECRET_PREV", "ancien-secret")
|
|
assert draco_session.verify_token(token)["jti"] == "rotation"
|
|
|
|
|
|
def test_revocation():
|
|
token, jti, exp = draco_session.issue_token(USER)
|
|
assert draco_session.verify_token(token) is not None
|
|
draco_session.revoke(jti, exp)
|
|
assert draco_session.verify_token(token) is None
|
|
assert {"jti": jti, "exp": exp} in draco_session.revoked_list()
|
|
|
|
|
|
def test_inerte_sans_secret(monkeypatch):
|
|
monkeypatch.delenv("DRACO_SESSION_SECRET")
|
|
assert draco_session.enabled() is False
|
|
assert draco_session.verify_token("peu-importe") is None
|
|
|
|
|
|
def test_fenetre_de_renouvellement():
|
|
assert draco_session.needs_renewal({"exp": int(time.time()) + 1800})
|
|
assert not draco_session.needs_renewal({"exp": int(time.time()) + 12 * 3600})
|
|
|
|
|
|
# ── return_to : liste blanche ────────────────────────────────────────────
|
|
|
|
@pytest.mark.parametrize("url,ok", [
|
|
("https://tiktak.dracodev.net/?task=3", True),
|
|
("https://og.dracodev.net/", True),
|
|
("https://dracodev.net/x", True),
|
|
("http://tiktak.dracodev.net/", False), # https seulement
|
|
("https://evil.com/", False),
|
|
("https://tiktak.dracodev.net.evil.com/", False),
|
|
("https://notdracodev.net/", False),
|
|
("", False),
|
|
])
|
|
def test_validate_return_to(url, ok):
|
|
assert (draco_session.validate_return_to(url) is not None) == ok
|
|
|
|
|
|
# ── Intégration : connexion / déconnexion / renouvellement ──────────────
|
|
|
|
def _login(client):
|
|
r = client.post("/api/auth/login",
|
|
json={"username": "admin", "password": "chab30"})
|
|
assert r.status_code == 200, r.text
|
|
return r
|
|
|
|
|
|
def _draco_cookie(response):
|
|
for raw in response.headers.get_list("set-cookie"):
|
|
if raw.startswith("draco_session="):
|
|
return raw
|
|
return None
|
|
|
|
|
|
def test_login_pose_le_cookie(admin_client):
|
|
r = _login(admin_client)
|
|
cookie = _draco_cookie(r)
|
|
assert cookie is not None
|
|
assert "Domain=.dracodev.net" in cookie
|
|
assert "HttpOnly" in cookie and "SameSite=lax" in cookie
|
|
token = cookie.split(";")[0].split("=", 1)[1]
|
|
assert draco_session.verify_token(token)["sub"] == "admin-1"
|
|
|
|
|
|
def test_login_sans_secret_aucun_cookie(admin_client, monkeypatch):
|
|
monkeypatch.delenv("DRACO_SESSION_SECRET")
|
|
r = _login(admin_client)
|
|
assert _draco_cookie(r) is None
|
|
|
|
|
|
def _token_depuis_login(client):
|
|
# Le jar du TestClient refuse le cookie Domain=.dracodev.net (hôte
|
|
# de test différent) : on lit le jeton dans l'en-tête set-cookie.
|
|
cookie = _draco_cookie(_login(client))
|
|
assert cookie is not None
|
|
return cookie.split(";")[0].split("=", 1)[1]
|
|
|
|
|
|
def test_logout_revoque_le_jti(admin_client):
|
|
token = _token_depuis_login(admin_client)
|
|
assert draco_session.verify_token(token)
|
|
r = admin_client.post("/api/auth/logout",
|
|
headers={"Cookie": f"draco_session={token}"})
|
|
assert r.status_code == 200, r.text
|
|
assert draco_session.verify_token(token) is None # révoqué
|
|
|
|
|
|
def test_renouvellement_glissant(admin_client):
|
|
_login(admin_client)
|
|
# Cookie draco presque expiré (dans la fenêtre des 2 h) :
|
|
now = int(time.time())
|
|
vieux = jose_jwt.encode(
|
|
{"sub": "admin-1", "name": "Admin", "iat": now - 11 * 3600,
|
|
"exp": now + 600, "jti": "vieux-jti"},
|
|
"secret-de-test", algorithm="HS256")
|
|
admin_client.cookies.set("draco_session", vieux)
|
|
r = admin_client.get("/api/auth/me")
|
|
assert r.status_code == 200, r.text
|
|
cookie = _draco_cookie(r)
|
|
assert cookie is not None # réémis
|
|
nouveau = cookie.split(";")[0].split("=", 1)[1]
|
|
assert draco_session.verify_token(nouveau)["jti"] != "vieux-jti"
|
|
|
|
|
|
def test_pas_de_renouvellement_si_frais(admin_client):
|
|
token = _token_depuis_login(admin_client)
|
|
admin_client.cookies.set("draco_session", token) # cookie frais porté
|
|
r = admin_client.get("/api/auth/me")
|
|
assert r.status_code == 200
|
|
assert _draco_cookie(r) is None # cookie frais : rien
|
|
|
|
|
|
def test_sso_authorize_liste_blanche(client):
|
|
r = client.get("/sso/authorize",
|
|
params={"return_to": "https://tiktak.dracodev.net/?task=3"},
|
|
follow_redirects=False)
|
|
assert r.status_code == 302
|
|
assert "return_to=" in r.headers["location"]
|
|
r = client.get("/sso/authorize",
|
|
params={"return_to": "https://evil.com/phish"},
|
|
follow_redirects=False)
|
|
assert r.status_code == 302
|
|
assert r.headers["location"] == "/" # refusé, pas de relais
|
|
|
|
|
|
# ── Déconnexion globale navigable (BUG-122, fusion #141) ────────────────
|
|
|
|
def test_sso_logout_revoque_et_redirige(admin_client):
|
|
token = _token_depuis_login(admin_client)
|
|
assert draco_session.verify_token(token)
|
|
r = admin_client.get(
|
|
"/sso/logout",
|
|
params={"return_to": "https://tiktak.dracodev.net/"},
|
|
headers={"Cookie": f"draco_session={token}"},
|
|
follow_redirects=False)
|
|
assert r.status_code == 302
|
|
assert r.headers["location"] == "https://tiktak.dracodev.net/"
|
|
assert draco_session.verify_token(token) is None # jti révoqué
|
|
cookie = _draco_cookie(r) # cookie effacé
|
|
assert cookie is not None and "Max-Age=0" in cookie
|
|
|
|
|
|
def test_sso_logout_liste_blanche(admin_client):
|
|
r = admin_client.get("/sso/logout",
|
|
params={"return_to": "https://evil.com/phish"},
|
|
follow_redirects=False)
|
|
assert r.status_code == 302
|
|
assert r.headers["location"] == "/" # refusé, pas de relais
|
|
r = admin_client.get("/sso/logout", follow_redirects=False)
|
|
assert r.status_code == 302
|
|
assert r.headers["location"] == "/" # sans return_to
|
|
|
|
|
|
def test_sso_logout_sans_session_idempotent(client):
|
|
# Aucun cookie : la route reste une redirection propre (les autres
|
|
# applications y naviguent même si la session a déjà expiré).
|
|
r = client.get("/sso/logout",
|
|
params={"return_to": "https://hub.dracodev.net/"},
|
|
follow_redirects=False)
|
|
assert r.status_code == 302
|
|
assert r.headers["location"] == "https://hub.dracodev.net/"
|
|
|
|
|
|
# ── Validation locale (ACC-3.2, #9) ──────────────────────────────────────
|
|
|
|
def test_me_avec_cookie_draco_seul(admin_client):
|
|
# Aucune session native : le cookie partagé suffit (sub = id).
|
|
token, _jti, _exp = draco_session.issue_token(USER)
|
|
admin_client.cookies.set("draco_session", token)
|
|
r = admin_client.get("/api/auth/me")
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["username"] == "admin"
|
|
|
|
|
|
def test_me_draco_revoque_refuse(admin_client):
|
|
token, jti, exp = draco_session.issue_token(USER)
|
|
draco_session.revoke(jti, exp)
|
|
admin_client.cookies.set("draco_session", token)
|
|
r = admin_client.get("/api/auth/me")
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_me_draco_sub_inconnu_refuse(admin_client):
|
|
token, _j, _e = draco_session.issue_token(
|
|
{"id": "personne-99", "username": "x", "display_name": "X"})
|
|
admin_client.cookies.set("draco_session", token)
|
|
r = admin_client.get("/api/auth/me")
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_me_draco_inerte_sans_secret(admin_client, monkeypatch):
|
|
token, _j, _e = draco_session.issue_token(USER)
|
|
monkeypatch.delenv("DRACO_SESSION_SECRET")
|
|
admin_client.cookies.set("draco_session", token)
|
|
r = admin_client.get("/api/auth/me")
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_revocations_protegees_par_cle_machine(admin_client, monkeypatch):
|
|
token, jti, exp = draco_session.issue_token(USER)
|
|
draco_session.revoke(jti, exp)
|
|
r = admin_client.get("/api/sso/revocations")
|
|
assert r.status_code == 404 # clé non configurée
|
|
monkeypatch.setenv("DRACO_SSO_MACHINE_KEY", "cle-machine")
|
|
r = admin_client.get("/api/sso/revocations")
|
|
assert r.status_code == 401 # en-tête absent
|
|
r = admin_client.get("/api/sso/revocations",
|
|
headers={"X-Draco-Machine-Key": "cle-machine"})
|
|
assert r.status_code == 200
|
|
assert {"jti": jti, "exp": exp} in r.json()["revoked"]
|
|
|
|
|
|
# ── Registre des sessions (HUB-22, #227) ────────────────────────────────
|
|
|
|
AUTRE = {"id": "user-2", "username": "autre", "display_name": "Autre"}
|
|
|
|
|
|
def test_sessions_emises_listees_triees_et_marquees():
|
|
_t1, jti1, exp1 = draco_session.issue_token(USER)
|
|
t2, jti2, exp2 = draco_session.issue_token(USER)
|
|
draco_session.issue_token(AUTRE) # autre sub : exclu
|
|
draco_session.revoke(jti2, exp2)
|
|
|
|
sessions = draco_session.issued_sessions_for_user("admin-1")
|
|
assert [s["jti"] for s in sessions] == [jti1, jti2] # exp croissant
|
|
assert sessions[0] == {"jti": jti1, "exp": exp1, "revoquee": False}
|
|
assert sessions[1] == {"jti": jti2, "exp": exp2, "revoquee": True}
|
|
assert draco_session.verify_token(t2) is None # cohérent
|
|
|
|
|
|
def test_sessions_emises_expirees_purgees():
|
|
now = int(time.time())
|
|
with draco_session._lock:
|
|
store = draco_session._load_issued()
|
|
store["admin-1"] = {"ancien": now - 10, "frais": now + 600}
|
|
assert [s["jti"] for s in
|
|
draco_session.issued_sessions_for_user("admin-1")] == ["frais"]
|
|
with draco_session._lock:
|
|
assert "ancien" not in draco_session._load_issued()["admin-1"]
|
|
assert draco_session.issued_sessions_for_user("personne") == []
|
|
assert draco_session.issued_sessions_for_user("") == []
|
|
|
|
|
|
def test_revoquer_une_session_appartenante():
|
|
token, jti, _exp = draco_session.issue_token(USER)
|
|
assert draco_session.revoke_one_for_user("admin-1", jti) is True
|
|
assert draco_session.verify_token(token) is None
|
|
assert draco_session.is_revoked(jti)
|
|
|
|
|
|
def test_revoquer_une_session_cross_utilisateur_refusee():
|
|
token, jti, _exp = draco_session.issue_token(USER)
|
|
# Le jti existe, mais pas dans l'ensemble émis de cet autre sub.
|
|
assert draco_session.revoke_one_for_user("user-2", jti) is False
|
|
assert draco_session.verify_token(token) is not None
|
|
assert draco_session.revoke_one_for_user("admin-1", "inconnu") is False
|
|
assert draco_session.revoke_one_for_user("", jti) is False
|
|
|
|
|
|
def test_revoquer_une_session_expiree_refusee():
|
|
now = int(time.time())
|
|
with draco_session._lock:
|
|
draco_session._load_issued()["admin-1"] = {"perime": now - 10}
|
|
assert draco_session.revoke_one_for_user("admin-1", "perime") is False
|
|
assert not draco_session.is_revoked("perime")
|