flowdeck/tests/test_fusion_draco.py

"""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. HUB-15 ajoute le
provisionnement JIT : aucun compte correspondant → le pendant local
est créé depuis les claims (rôle standard, jamais admin, sans mot
de passe local), sauf pierre tombale de dissociation.
"""

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", name="Bruno"):
    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": name, "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()
    # Le lien explicite (créé pendant une visite connectée) ouvre 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"


# ── HUB-15 : provisionnement JIT ─────────────────────────────────────

def _ajoute_deuxieme_utilisateur():
    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()


def _compte_utilisateurs():
    with get_conn() as conn:
        return conn.execute("SELECT COUNT(*) FROM users").fetchone()[0]


def test_jit_cree_le_pendant_local_en_multi_utilisateurs(client):
    _ajoute_deuxieme_utilisateur()
    avant = _compte_utilisateurs()
    # Sub inconnu, sans lien, en multi-utilisateurs : avant HUB-15 le
    # pont refusait ; désormais le pendant local est créé en JIT.
    client.cookies.set("draco_session", _jeton(sub="og-nouveau", name="Nouvelle"))
    r = client.get("/auth/draco", follow_redirects=False)
    assert r.headers["location"] == "/workspaces"
    assert _compte_utilisateurs() == avant + 1
    with get_conn() as conn:
        user = conn.execute(
            "SELECT * FROM users WHERE login = 'draco-og-nouveau'").fetchone()
        assert user is not None
        assert user["full_name"] == "Nouvelle"
        assert user["is_admin"] == 0            # jamais admin
        assert user["is_active"] == 1
        assert user["password_hash"] is None    # aucun mot de passe local
        assert user["auth_method"] == "draco"
        lien = conn.execute(
            "SELECT * FROM draco_links WHERE sub = 'og-nouveau'").fetchone()
        assert lien is not None and lien["user_id"] == user["id"]
        assert lien["unlinked_at"] is None
    # Deuxième arrivée : même compte réutilisé, aucune création de plus.
    client.cookies.delete("flowdeck_session")
    r = client.get("/auth/draco", follow_redirects=False)
    assert r.headers["location"] == "/workspaces"
    assert _compte_utilisateurs() == avant + 1
    # Le compte JIT ne peut pas se connecter par mot de passe local.
    r = client.post("/auth/local-login",
                    json={"email": "draco-og-nouveau", "password": "nimporte"})
    assert r.status_code == 401


def test_jit_pierre_tombale_sans_recreation(client):
    _ajoute_deuxieme_utilisateur()
    client.cookies.set("draco_session", _jeton(sub="og-ephemere"))
    r = client.get("/auth/draco", follow_redirects=False)
    assert r.headers["location"] == "/workspaces"   # compte JIT créé
    # La session courante est celle du compte JIT : dissociation.
    r = client.delete("/auth/draco-link")
    assert r.json()["ok"] is True
    apres_creation = _compte_utilisateurs()
    # Nouvelle arrivée du même sub : la pierre tombale interdit toute
    # re-création automatique — retour à la connexion locale.
    client.cookies.delete("flowdeck_session")
    r = client.get("/auth/draco", follow_redirects=False)
    assert r.headers["location"] == "/auth/login?provider=local"
    assert _compte_utilisateurs() == apres_creation