Files
flowdeck/tests/test_fusion_draco.py
T
Spark 1ce9f691a6
FlowDeck CI / lint (push) Failing after 2m8s
FlowDeck CI / test (push) Failing after 20m43s
FlowDeck CI / docker (push) Skipped
feat: Fusion ACC-3.2 — pont SSO draco_session (/auth/draco), liens sub → compte (migration 51) (#21)
2026-10-20 23:56:24 -04:00

114 lines
4.3 KiB
Python

"""Fusion ACC-3.2 — validation locale draco_session et pont de session.
Le cookie émis par ObsiGate (ACC-3.1) est vérifié localement
(secret partagé, HS256), résolu vers le compte Flowdeck
(draco_links, migration 51 ; repli mono-utilisateur), et converti en
session native par GET /auth/draco. Sans DRACO_SESSION_SECRET, le
pont renvoie simplement à la connexion locale.
"""
import base64
import hashlib
import hmac
import json
import time
import pytest
from app.db import get_conn
from app.services import draco
SECRET = "secret-partage-de-test"
def _jeton(sub="admin-1", secret=SECRET, exp=None, jti="jti-test"):
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": "Bruno", "iat": int(time.time()),
"exp": exp if exp is not None else int(time.time()) + 3600,
"jti": jti,
}).encode())
sig = hmac.new(secret.encode(), f"{header}.{payload}".encode(),
hashlib.sha256).digest()
return f"{header}.{payload}.{b64(sig)}"
@pytest.fixture(autouse=True)
def _env(monkeypatch):
monkeypatch.setenv("DRACO_SESSION_SECRET", SECRET)
monkeypatch.setattr(draco, "_refresh_revocations", lambda: None)
draco._revoked_cache.clear()
yield
draco._revoked_cache.clear()
def test_verify_token():
assert draco.verify_token(_jeton())["sub"] == "admin-1"
assert draco.verify_token(_jeton(secret="mauvais")) is None
assert draco.verify_token(_jeton(exp=int(time.time()) - 5)) is None
assert draco.verify_token("pas.un.jeton") is None
def test_pont_sans_cookie_vers_login(client):
r = client.get("/auth/draco", follow_redirects=False)
assert r.status_code == 302
assert r.headers["location"] == "/auth/login?provider=local"
def test_pont_cookie_valide_cree_session(client):
client.cookies.set("draco_session", _jeton(sub="admin-1"))
r = client.get("/auth/draco", follow_redirects=False)
assert r.status_code == 302
assert r.headers["location"] == "/workspaces"
assert "flowdeck_session" in r.headers.get("set-cookie", "")
me = client.get("/auth/user")
assert me.json()["authenticated"] is True
def test_pont_cookie_falsifie_vers_login(client):
client.cookies.set("draco_session", _jeton(secret="mauvais"))
r = client.get("/auth/draco", follow_redirects=False)
assert r.headers["location"] == "/auth/login?provider=local"
def test_pont_inerte_sans_secret(client, monkeypatch):
monkeypatch.delenv("DRACO_SESSION_SECRET")
client.cookies.set("draco_session", _jeton())
r = client.get("/auth/draco", follow_redirects=False)
assert r.headers["location"] == "/auth/login?provider=local"
def test_lien_automatique_et_gestion(client):
client.cookies.set("draco_session", _jeton(sub="og-bruno-1"))
client.get("/auth/user") # visite doublement connectée
r = client.get("/auth/draco-link")
assert r.json()["sub"] == "og-bruno-1"
r = client.delete("/auth/draco-link")
assert r.json()["ok"] is True
assert client.get("/auth/draco-link").json()["sub"] is None
client.get("/auth/user") # ne recrée pas le lien
assert client.get("/auth/draco-link").json()["sub"] is None
r = client.post("/auth/draco-link") # ré-association explicite
assert r.json()["sub"] == "og-bruno-1"
def test_lien_requis_en_multi_utilisateurs(client):
with get_conn() as conn:
conn.execute(
"INSERT INTO users (login, email, password_hash, full_name, is_active) "
"VALUES ('deuxieme', '[email protected]', 'x', 'Deuxième', 1)")
conn.commit()
# Sans lien et avec deux utilisateurs actifs : le pont refuse.
client.cookies.set("draco_session", _jeton(sub="og-inconnu"))
r = client.get("/auth/draco", follow_redirects=False)
assert r.headers["location"] == "/auth/login?provider=local"
# Le lien explicite (créé pendant une visite connectée) rouvre la porte.
client.cookies.set("draco_session", _jeton(sub="og-bruno-1"))
assert client.post("/auth/draco-link").json()["sub"] == "og-bruno-1"
client.cookies.delete("flowdeck_session")
r = client.get("/auth/draco", follow_redirects=False)
assert r.headers["location"] == "/workspaces"