diff --git a/CHANGELOG.md b/CHANGELOG.md index c6af518..acc9c89 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,13 @@ +## 0.46.2 — 2026-11-06 — Rôles du Hub propagés à Kanbante (HUB-19, suite du pilote ShaarIt 0.7.0) + +Version **corrective** au sens du patron du dépôt (pas de fonction nouvelle côté métier : un garde d'accès branché sur le registre du Hub, contrat D3(a) du plan Hub v2). + +- **Client du registre** (`app/hub_roles.py`, portage Python du pilote `hub_roles.rs` de ShaarIt) : à l'ouverture de session draco, consultation de `GET {hub}/api/roles/kanbante` en relayant l'en-tête Cookie complet de la requête (`Accept: application/json`, délai 2 s) → rôle ∈ {`aucun`, `utilisateur`, `administrateur`}. Cache mémoire par `sub`, TTL 60 s, horodatage posé **avant** l'appel (échec = ancien cache conservé, pas de rafale). URL du Hub résolue comme la barre commune (`resoudre_urls_draco` : `DRACO_HUB_URL` explicite > hôte public de la requête > dérivation par l'hôte sœur de l'agent > défaut officiel) — jamais d'URL dev figée. **Toute erreur = rôle inconnu → comportement actuel** : le registre ne crée jamais un verrouillage nouveau. +- **Pont `/auth/draco` — consultation d'abord** : le registre est consulté **avant** la liaison à une session native, avant `resolve_draco_user()` et donc avant toute création JIT (HUB-16). Rôle connu `aucun` → refus explicite en français (403, page d'erreur) : **aucune session Kanbante ouverte par draco pour ce `sub`, aucun compte créé, aucune liaison posée**. La session ouverte par le pont porte désormais un marqueur « établie par draco » **couvert par la signature** du cookie de session (`user.expiry.draco.hmac` ; le format natif historique reste lu tel quel). +- **`/admin` gardé par le registre pour les sessions draco** : `_require_admin` (qui couvre `/admin` et toutes ses actions) exige, pour une session établie par draco dont le rôle du registre est **connu**, le niveau `administrateur` du registre **en plus** du rôle local admin. Rôle inconnu ou session native → règles locales actuelles, inchangées. +- **Non touchés** : session native préexistante, connexion native (mot de passe), Basic d'environnement, jetons `kbt_`, `/api/search`, droits internes et réconciliation Gitea par étiquettes (le piège « exactement une étiquette reconnue » n'est pas le sujet de cette carte). Détails : `docs/correspondance-draco.md`, section « Rôles du Hub (HUB-19) ». +- Tests : `tests/test_hub_roles.py` — client (contrat d'appel, cookie relayé, cache et TTL, échec sans rafale avec ancien cache conservé, réponses illisibles, résolution d'URL par hôte dev et par variable explicite), marqueur de session signé, pont (refus `aucun` sans création ni liaison, JIT `utilisateur`, repli rôle inconnu) et garde `/admin` (session draco `utilisateur` refusée malgré le rôle local admin, `administrateur` accepté, inconnu = repli local, session native non soumise). + ## 0.46.1 — 2026-11-06 — Barre commune v0.5.0 : thème commun (HUB-18, fusion #132) > **Hypothèse d'ordre** : cette version suppose la fusion préalable de la PR #84 (Kanbante 0.46.0, HUB-16) ; branchée sur `main` (0.45.12), elle vise **0.46.1**. diff --git a/VERSION b/VERSION index 620104d..43c125a 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.46.1 +0.46.2 diff --git a/app/hub_roles.py b/app/hub_roles.py new file mode 100644 index 0000000..8a60760 --- /dev/null +++ b/app/hub_roles.py @@ -0,0 +1,120 @@ +"""Rôles par application — consommation du registre du Hub (HUB-19). + +Portage Python du pilote ShaarIt 0.7.0 (``src/hub_roles.rs``), +contrat D3(a) du plan Hub v2 : à l'ouverture de session draco, +Kanbante demande au Hub le niveau du porteur du cookie pour +l'application ``kanbante`` — ``GET {hub}/api/roles/kanbante``, +en-tête Cookie complet de la requête relayé tel quel, +``Accept: application/json``, délai 2 s → ``{"app", "sub", +"role", "explicite", "ttl_secondes": 60}``, rôle ∈ {``aucun``, +``utilisateur``, ``administrateur``}. + +Cache mémoire par ``sub``, TTL 60 s, sur le patron du cache de +révocations de ``app.draco`` : l'horodatage est posé **avant** +l'appel — un échec ne déclenche pas de rafale et l'ancien cache +reste en vigueur (il resservira aux appels suivants pendant le +TTL posé par la tentative). L'URL du Hub est résolue comme celle +de la barre commune (``app.config.resoudre_urls_draco`` : +variable explicite ``DRACO_HUB_URL`` > hôte public de la requête +> dérivation par l'hôte sœur de l'agent > défaut officiel) — +jamais d'URL dev figée. + +**Repli obligatoire** : Hub injoignable, en erreur ou réponse +illisible → ``None`` (rôle **inconnu**), et l'appelant garde son +comportement actuel. Le registre ne crée jamais un verrouillage +nouveau. +""" + +from __future__ import annotations + +import logging +import threading +import time + +import httpx + +from app.config import resoudre_urls_draco + +logger = logging.getLogger(__name__) + +#: Identifiant de Kanbante dans le registre des applications du Hub. +APP = "kanbante" + +#: TTL du cache, aligné sur ``ttl_secondes`` annoncé par le Hub +#: et sur le cache de révocations draco. +CACHE_TTL = 60.0 # secondes + +ROLES_CONNUS = ("aucun", "utilisateur", "administrateur") + +_verrou = threading.Lock() +_cache: dict[str, str] = {} +_cache_fetched_at: float = 0.0 + + +def _reset_cache() -> None: + """Vide le cache (tests, ou redémarrage de processus).""" + global _cache_fetched_at + with _verrou: + _cache.clear() + _cache_fetched_at = 0.0 + + +def _consulter(hub_url: str, cookie: str) -> str | None: + """Un appel au registre ; rôle connu ou ``None`` (inconnu).""" + url = f"{hub_url.rstrip('/')}/api/roles/{APP}" + try: + reponse = httpx.get( + url, + headers={"Cookie": cookie, "Accept": "application/json"}, + timeout=2.0, + ) + except Exception as exc: # Hub injoignable : repli, jamais un blocage + logger.debug("Registre des rôles du Hub injoignable (%s) : %s", url, exc) + return None + if getattr(reponse, "status_code", None) != 200: + logger.debug( + "Registre des rôles du Hub en erreur (%s) : statut %s", + url, getattr(reponse, "status_code", "?"), + ) + return None + try: + donnees = reponse.json() + except Exception: + return None + if not isinstance(donnees, dict): + return None + role = donnees.get("role") + return role if role in ROLES_CONNUS else None + + +def role_pour(cookie: str, sub: str, hote: str | None = None) -> str | None: + """Niveau connu pour ce ``sub`` : un rôle ou ``None`` (inconnu). + + ``cookie`` est l'en-tête Cookie complet de la requête, relayé + tel quel au Hub ; ``hote`` est l'hôte public de la requête, + pour résoudre l'URL du Hub comme la barre commune. + """ + global _cache_fetched_at + sub = str(sub or "") + if not sub: + return None + now = time.monotonic() + with _verrou: + if now - _cache_fetched_at < CACHE_TTL and sub in _cache: + return _cache[sub] + # Horodatage posé avant l'appel (patron révocations) : + # l'ancien cache est conservé tel quel pendant l'appel. + _cache_fetched_at = now + hub_url = resoudre_urls_draco(hote).hub_url + resultat = _consulter(hub_url, cookie) + if resultat is not None: + with _verrou: + _cache[sub] = resultat + _cache_fetched_at = time.monotonic() + else: + logger.debug( + "Registre des rôles du Hub indisponible (%s/api/roles/%s)" + " — comportement actuel conservé", + hub_url, APP, + ) + return resultat diff --git a/app/main.py b/app/main.py index 4d7fbb3..b795f10 100644 --- a/app/main.py +++ b/app/main.py @@ -263,23 +263,48 @@ def _session_secret() -> bytes: -def _make_session(user: str) -> str: - """Signed session cookie (0.22.0): user.expiry.hmac.""" +def _make_session(user: str, via_draco: bool = False) -> str: + """Signed session cookie (0.22.0): user.expiry.hmac. + + HUB-19 (0.46.2) : une session **établie par le pont draco** + porte un segment ``draco`` couvert par la signature + (``user.expiry.draco.hmac``) — c'est ce marqueur infalsifiable + qui soumet ensuite ``/admin`` au registre des rôles du Hub + (voir ``_require_admin``). Les sessions natives gardent le + format historique, et les deux formats restent lisibles. + """ exp = int(time.time()) + SESSION_DAYS * 86400 + charge = f"{user}.{exp}.draco" if via_draco else f"{user}.{exp}" sig = hmac.new( - _session_secret(), f"{user}.{exp}".encode("utf-8"), hashlib.sha256 + _session_secret(), charge.encode("utf-8"), hashlib.sha256 ).hexdigest() - return f"{user}.{exp}.{sig}" + return f"{charge}.{sig}" -def _session_user(request: Request) -> str | None: +def _session_info(request: Request) -> tuple[str, bool] | None: + """(utilisateur, établie par draco) de la session, ou None. + + Vérifie signature et expiration, dans les deux formats + (natif historique et marqué draco de la 0.46.2). + """ raw = request.cookies.get(SESSION_COOKIE, "") - user, _, rest = raw.partition(".") - exp, _, sig = rest.partition(".") + morceaux = raw.split(".") + if len(morceaux) < 3: + return None + sig = morceaux[-1] + via_draco = len(morceaux) >= 4 and morceaux[-2] == "draco" + if via_draco: + exp = morceaux[-3] + user = ".".join(morceaux[:-3]) + charge = f"{user}.{exp}.draco" + else: + exp = morceaux[-2] + user = ".".join(morceaux[:-2]) + charge = f"{user}.{exp}" if not user or not exp or not sig: return None expected = hmac.new( - _session_secret(), f"{user}.{exp}".encode("utf-8"), hashlib.sha256 + _session_secret(), charge.encode("utf-8"), hashlib.sha256 ).hexdigest() if not secrets.compare_digest(sig, expected): return None @@ -288,6 +313,14 @@ def _session_user(request: Request) -> str | None: return None except ValueError: return None + return user, via_draco + + +def _session_user(request: Request) -> str | None: + info = _session_info(request) + if info is None: + return None + user = info[0] # 0.24.0 : la session peut appartenir à n'importe quel # compte de la base (résolu ensuite par _current_user) ou, # en mode env pur, à l'opérateur d'environnement. @@ -299,6 +332,12 @@ def _session_user(request: Request) -> str | None: return None +def _session_via_draco(request: Request) -> bool: + """Vrai si la session présentée a été établie par le pont draco.""" + info = _session_info(request) + return bool(info and info[1] and _session_user(request) is not None) + + def _basic_ok(request: Request) -> bool: settings = get_settings() header = request.headers.get("authorization", "") @@ -597,9 +636,21 @@ def draco_bridge(request: Request, next: str = "/"): Un ``sub`` inconnu n'ouvre plus jamais la session de l'opérateur d'environnement ni celle d'un compte non lié. Inerte sans ``DRACO_SESSION_SECRET``. + + HUB-19 (0.46.2, contrat D3(a)) : le registre des rôles du Hub + est consulté **en premier** — avant la liaison à une session + native, avant la résolution de la correspondance et avant + toute création JIT. Rôle connu ``aucun`` → le pont refuse : + aucune session Kanbante n'est ouverte par draco pour ce + ``sub`` et aucun compte n'est créé. Rôle inconnu (Hub + injoignable) → comportement actuel, inchangé. Les autres + accès ne passent pas par ce pont et ne sont pas touchés : + session native préexistante, connexion native (mot de + passe), Basic d'environnement, jetons ``kbt_``. """ from app import accounts as accounts_service from app import draco + from app import hub_roles settings = get_settings() claims = draco.claims_for(request) @@ -608,6 +659,28 @@ def draco_bridge(request: Request, next: str = "/"): sub = str(claims.get("sub") or "") if not sub: return RedirectResponse("/login", status_code=303) + # HUB-19 : consultation du registre AVANT toute résolution + # ou création (resolve_draco_user peut créer le compte JIT : + # un sub refusé ne doit jamais en recevoir un). + role_hub = hub_roles.role_pour( + request.headers.get("cookie", ""), sub, + hote=request.headers.get("host"), + ) + if role_hub == "aucun": + return render( + request, "error.html", status_code=403, + message=( + "Accès refusé : le registre des rôles de la suite " + "(Hub Dracodev) attribue le niveau « aucun » à " + "votre compte pour Kanbante — l'ouverture de " + "session par draco est refusée et aucun compte " + "Kanbante n'a été créé. L'administrateur de la " + "suite peut modifier ce niveau dans la matrice " + "des rôles du Hub. Vos autres accès à Kanbante " + "(connexion par mot de passe, session déjà " + "ouverte) ne sont pas touchés." + ), + ) # Déjà connecté nativement : lier le sub à ce compte. current = _current_user(request) if current is not None and current.get("id") is not None: @@ -618,12 +691,15 @@ def draco_bridge(request: Request, next: str = "/"): return RedirectResponse(_safe_next(next), status_code=303) # Correspondance, amorçage ou création JIT : ouvrir la # session du compte que ce sub désigne — et seulement lui. + # La session est marquée « établie par draco » (segment signé, + # voir _make_session) : c'est elle, et elle seule, que le + # garde de /admin soumettra au registre du Hub. user = accounts_service.resolve_draco_user(claims, provisionner=True) if user is not None: response = RedirectResponse(_safe_next(next), status_code=303) response.set_cookie( SESSION_COOKIE, - _make_session(user["username"]), + _make_session(user["username"], via_draco=True), max_age=SESSION_DAYS * 86400, httponly=True, samesite="lax", @@ -1006,6 +1082,28 @@ def _require_admin(request: Request) -> dict | None: user = _current_user(request) if user is None or user.get("role") != "admin": return None + # HUB-19 (0.46.2) : pour une session **établie par le pont + # draco** (marqueur signé, voir _make_session), si le rôle du + # registre du Hub est connu, il faut AUSSI « administrateur » + # dans le registre — en plus du rôle local ci-dessus. Rôle + # inconnu (Hub injoignable) ou session native (mot de passe, + # Basic, jeton) : règles locales actuelles, inchangées. Ce + # garde couvre /admin et toutes ses actions (même fonction). + if _session_via_draco(request): + from app import draco + from app import hub_roles + + sub = str(user.get("draco_sub") or "") + if not sub: + claims = draco.claims_for(request) + sub = str((claims or {}).get("sub") or "") + if sub: + role_hub = hub_roles.role_pour( + request.headers.get("cookie", ""), sub, + hote=request.headers.get("host"), + ) + if role_hub is not None and role_hub != "administrateur": + return None return user diff --git a/docs/correspondance-draco.md b/docs/correspondance-draco.md index 6415a04..0a1a256 100644 --- a/docs/correspondance-draco.md +++ b/docs/correspondance-draco.md @@ -97,8 +97,43 @@ Chaque utilisateur peut aussi voir sa propre liaison dans **Mon profil → Liaison draco (SSO)** et la délier lui-même. > Le registre matriciel des rôles par application côté Hub -> (carte HUB-19) lira cette correspondance ; la présente carte ne -> livre que le côté Kanbante, qui est la source de vérité locale. +> (carte HUB-19) lit cette correspondance ; la 0.46.0 ne livrait +> que le côté Kanbante, qui reste la source de vérité locale. +> La consommation du registre par Kanbante est décrite à la +> section suivante (0.46.2). + +## Rôles du Hub (HUB-19, 0.46.2) + +Le registre des rôles du Hub (matrice par application, contrat +D3(a)) est consulté par `app/hub_roles.py` — portage du pilote +ShaarIt 0.7.0 : `GET {hub}/api/roles/kanbante`, en-tête Cookie +complet de la requête relayé, délai 2 s, cache mémoire par +`sub` de 60 s avec horodatage posé avant l'appel (un échec +conserve l'ancien cache et ne déclenche pas de rafale). L'URL +du Hub est celle de la barre commune (`resoudre_urls_draco`). +Toute erreur rend un rôle **inconnu**, qui vaut toujours +« comportement actuel » — jamais un verrouillage nouveau. + +Effets, pour les seules sessions **établies par le pont +draco** : + +1. **Pont `/auth/draco`** : la consultation précède la + liaison, la résolution de la correspondance et la création + JIT. Rôle connu `aucun` → refus explicite (403, en + français) : aucune session ouverte par draco pour ce + `sub`, aucun compte créé. La session que le pont ouvre + porte un marqueur « draco » couvert par la signature du + cookie (`user.expiry.draco.hmac`). +2. **`/admin`** (et ses actions, même garde + `_require_admin`) : session établie par draco + rôle du + registre connu ⇒ il faut `administrateur` dans le + registre, **en plus** du rôle local. Rôle inconnu ou + session native ⇒ règles locales, inchangées. + +Les autres accès ne sont pas draco et ne sont pas touchés : +session native préexistante, connexion native (mot de +passe), Basic d'environnement, jetons `kbt_`. Ni les droits +internes ni la réconciliation par étiquettes ne changent. ## Surfaces voisines diff --git a/tests/test_hub_roles.py b/tests/test_hub_roles.py new file mode 100644 index 0000000..0db4a53 --- /dev/null +++ b/tests/test_hub_roles.py @@ -0,0 +1,375 @@ +"""HUB-19 (0.46.2) — rôles du Hub propagés à Kanbante (contrat D3(a)). + +Portage du pilote ShaarIt 0.7.0. Couvert ici : + +- client ``app.hub_roles`` : appel ``GET {hub}/api/roles/kanbante`` + (Cookie complet relayé, ``Accept: application/json``), URL du + Hub résolue comme la barre (explicite > hôte de la requête > + défaut officiel), cache 60 s par ``sub``, horodatage posé avant + l'appel (échec = ancien cache conservé, pas de rafale), tout + échec = rôle inconnu (``None``) ; +- pont ``/auth/draco`` : la consultation précède la résolution et + la création JIT — rôle connu ``aucun`` → refus explicite 403, + **aucune session ouverte, aucun compte créé**, aucune liaison à + une session native préexistante ; +- ``/admin`` : session établie par draco + rôle connu ≠ + ``administrateur`` → refusé malgré le rôle local admin ; rôle + ``administrateur`` → accepté ; rôle inconnu → repli local ; + session **native** (même admin, même cookie draco présent) → + règles locales inchangées. +""" + +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 draco, hub_roles, models + +SECRET = "secret-hub19-partage" + +SUBS_TEST = ( + "hub19-refuse", "hub19-membre", "hub19-chef", "hub19-inconnu", + "hub19-sans-liaison", +) + + +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 _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): + if isinstance(self._donnees, Exception): + raise self._donnees + 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``, lu dans le cookie relayé.""" + etat = {"roles": {}, "appels": [], "mode": "ok"} + + def faux_get(url, headers=None, timeout=None): + etat["appels"].append( + {"url": url, "headers": dict(headers or {}), "timeout": timeout} + ) + if etat["mode"] == "panne": + raise httpx.ConnectError("Hub injoignable (simulation)") + if etat["mode"] == "erreur500": + return _Reponse(500, {"detail": "erreur"}) + 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 + + +# ── Client : contrat d'appel et résolution d'URL ───────────── + +def test_client_appel_contrat_et_cache(faux_hub): + faux_hub["roles"]["sub-a"] = "administrateur" + cookie = f"draco_session={_jeton('sub-a')}; autre=1" + assert hub_roles.role_pour(cookie, "sub-a") == "administrateur" + # Deuxième appel dans le TTL : servi par le cache, aucun appel. + assert hub_roles.role_pour(cookie, "sub-a") == "administrateur" + assert len(faux_hub["appels"]) == 1 + appel = faux_hub["appels"][0] + assert appel["url"] == "https://hub.dracodev.net/api/roles/kanbante" + assert appel["headers"]["Cookie"] == cookie + assert appel["headers"]["Accept"] == "application/json" + assert appel["timeout"] == 2.0 + + +def test_client_url_hub_par_hote_dev(faux_hub): + hub_roles.role_pour( + f"draco_session={_jeton('sub-b')}", "sub-b", + hote="kanbante-dev.dracodev.net", + ) + assert faux_hub["appels"][0]["url"] == ( + "https://hub-dev.dracodev.net/api/roles/kanbante" + ) + + +def test_client_url_hub_explicite_prioritaire(faux_hub, monkeypatch): + monkeypatch.setenv("DRACO_HUB_URL", "https://hub.exemple.test/") + hub_roles.role_pour( + f"draco_session={_jeton('sub-c')}", "sub-c", + hote="kanbante-dev.dracodev.net", + ) + assert faux_hub["appels"][0]["url"] == ( + "https://hub.exemple.test/api/roles/kanbante" + ) + + +def test_client_ttl_expire_reconsulte(faux_hub): + cookie = f"draco_session={_jeton('sub-d')}" + faux_hub["roles"]["sub-d"] = "utilisateur" + assert hub_roles.role_pour(cookie, "sub-d") == "utilisateur" + hub_roles._cache_fetched_at -= hub_roles.CACHE_TTL + 1 + faux_hub["roles"]["sub-d"] = "administrateur" + assert hub_roles.role_pour(cookie, "sub-d") == "administrateur" + assert len(faux_hub["appels"]) == 2 + + +def test_client_echec_inconnu_et_ancien_cache_conserve(faux_hub): + cookie = f"draco_session={_jeton('sub-e')}" + faux_hub["roles"]["sub-e"] = "utilisateur" + assert hub_roles.role_pour(cookie, "sub-e") == "utilisateur" + # TTL expiré, Hub en panne : l'appel tenté rend « inconnu »… + hub_roles._cache_fetched_at -= hub_roles.CACHE_TTL + 1 + faux_hub["mode"] = "panne" + assert hub_roles.role_pour(cookie, "sub-e") is None + appels = len(faux_hub["appels"]) + # …mais l'horodatage posé avant l'appel garde l'ancien cache : + # l'appel suivant (dans le nouveau TTL) le ressert sans rafale. + assert hub_roles.role_pour(cookie, "sub-e") == "utilisateur" + assert len(faux_hub["appels"]) == appels + + +def test_client_reponses_illisibles_inconnues(monkeypatch): + def faux_get(url, headers=None, timeout=None): + return _Reponse(200, {"role": "super-utilisateur"}) + + monkeypatch.setattr(httpx, "get", faux_get) + assert hub_roles.role_pour("c=1", "sub-f") is None + hub_roles._reset_cache() + + def faux_get_403(url, headers=None, timeout=None): + return _Reponse(403, {"role": "aucun"}) + + monkeypatch.setattr(httpx, "get", faux_get_403) + assert hub_roles.role_pour("c=1", "sub-f") is None + + +# ── Intégration : pont draco et garde /admin ───────────────── + +def _nettoyer(): + 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() + ] + 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,)) + + +@pytest.fixture() +def client_hub19(monkeypatch, fake_gitea, faux_hub): + 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() + _nettoyer() + with TestClient(main.app) as client: + admin = next(u for u in models.list_users() if u["role"] == "admin") + sub_avant = admin.get("draco_sub") + try: + yield client, faux_hub, admin + finally: + models.update_user(admin["id"], draco_sub=sub_avant) + _nettoyer() + draco._revoked_cache.clear() + + +def _neutraliser_amorcage(admin) -> bool: + """Vrai si l'amorçage pur a été neutralisé (témoin lié à l'admin). + + Même patron que tests/test_draco_correspondance.py : les + scénarios JIT exigent que l'unique admin sans liaison ne + capture pas le premier sub de passage. + """ + users = models.list_users() + if len(users) == 1 and users[0]["role"] == "admin" \ + and not users[0].get("draco_sub"): + models.update_user(users[0]["id"], draco_sub="hub19-temoin-amorce") + return True + return False + + +def _pont(client, sub, nom="Alice Exemple"): + client.cookies.set("draco_session", _jeton(sub, nom)) + return client.get("/auth/draco", follow_redirects=False) + + +def test_pont_role_aucun_refuse_sans_creation(client_hub19): + client, hub, _admin = client_hub19 + hub["roles"]["hub19-refuse"] = "aucun" + avant = models.count_users() + r = _pont(client, "hub19-refuse", "Refusée Exemple") + assert r.status_code == 403 + assert "Accès refusé" in r.text + assert "« aucun »" in r.text + # Aucune session Kanbante ouverte par draco… + assert "kanbante_session" not in r.headers.get("set-cookie", "") + # …et aucun compte JIT créé pour ce sub. + assert models.count_users() == avant + assert models.get_user_by_draco("hub19-refuse") is None + + +def test_pont_role_aucun_ne_lie_pas_une_session_native(client_hub19): + client, hub, admin = client_hub19 + hub["roles"]["hub19-sans-liaison"] = "aucun" + # Session native préexistante (format historique, sans marqueur). + client.cookies.set( + "kanbante_session", main._make_session(admin["username"]) + ) + r = _pont(client, "hub19-sans-liaison") + assert r.status_code == 403 + # La consultation a précédé la liaison : rien n'a été lié. + assert models.get_user_by_draco("hub19-sans-liaison") is None + # La session native elle-même n'est pas touchée. + assert client.get("/").status_code == 200 + + +def test_pont_role_utilisateur_ouvre_session_jit(client_hub19): + client, hub, admin = client_hub19 + hub["roles"]["hub19-membre"] = "utilisateur" + pur = _neutraliser_amorcage(admin) + try: + r = _pont(client, "hub19-membre", "Membre Exemple") + assert r.status_code == 303 and r.headers["location"] == "/" + assert "kanbante_session" in r.headers.get("set-cookie", "") + cree = models.get_user_by_draco("hub19-membre") + assert cree is not None and cree["role"] == "member" + assert client.get("/").status_code == 200 + finally: + if pur: + models.update_user(admin["id"], draco_sub=None) + + +def test_pont_role_inconnu_comportement_actuel(client_hub19): + client, hub, admin = client_hub19 + hub["mode"] = "panne" + pur = _neutraliser_amorcage(admin) + try: + r = _pont(client, "hub19-inconnu", "Inconnu Exemple") + assert r.status_code == 303 and r.headers["location"] == "/" + assert models.get_user_by_draco("hub19-inconnu") is not None + finally: + if pur: + models.update_user(admin["id"], draco_sub=None) + + +def _admin_lie_au_pont(client, hub, admin, role_hub): + """Lie le compte admin local au sub chef, puis ouvre sa session par le pont.""" + models.update_user(admin["id"], draco_sub="hub19-chef") + hub["roles"]["hub19-chef"] = role_hub + hub_roles._reset_cache() + r = _pont(client, "hub19-chef", "Chef Exemple") + assert r.status_code == 303 and r.headers["location"] == "/" + return r + + +def test_admin_session_draco_exige_administrateur_du_registre(client_hub19): + client, hub, admin = client_hub19 + _admin_lie_au_pont(client, hub, admin, "utilisateur") + # Rôle local admin, mais registre connu « utilisateur » : refusé. + r = client.get("/admin") + assert r.status_code == 403 + # Registre passé à « administrateur » : accepté. + hub["roles"]["hub19-chef"] = "administrateur" + hub_roles._reset_cache() + r = client.get("/admin") + assert r.status_code == 200 + + +def test_admin_session_draco_role_inconnu_repli_local(client_hub19): + client, hub, admin = client_hub19 + _admin_lie_au_pont(client, hub, admin, "administrateur") + assert client.get("/admin").status_code == 200 + # Hub en panne et cache expiré : rôle inconnu → règles locales. + hub["mode"] = "panne" + hub_roles._cache_fetched_at -= hub_roles.CACHE_TTL + 1 + # L'appel de garde rend « inconnu » (l'ancien cache ne ressert + # qu'aux appels suivants, patron du pilote) : le repli local joue. + r = client.get("/admin") + assert r.status_code == 200 + + +def test_admin_session_native_non_soumise_au_registre(client_hub19): + client, hub, admin = client_hub19 + models.update_user(admin["id"], draco_sub="hub19-chef") + hub["roles"]["hub19-chef"] = "utilisateur" + # Session NATIVE de l'admin (mot de passe / format historique) : + # même avec un cookie draco présent et un rôle registre connu + # « utilisateur », /admin reste régi par les règles locales. + client.cookies.set( + "kanbante_session", main._make_session(admin["username"]) + ) + client.cookies.set("draco_session", _jeton("hub19-chef", "Chef Exemple")) + r = client.get("/admin") + assert r.status_code == 200 + + +def test_marqueur_session_draco_signe(): + """Le marqueur « établie par draco » est couvert par la signature.""" + from starlette.requests import Request + + def requete(valeur): + scope = { + "type": "http", "method": "GET", "path": "/", + "headers": [(b"cookie", f"kanbante_session={valeur}".encode())], + } + return Request(scope) + + natif = main._make_session("bruno") + assert main._session_info(requete(natif)) == ("bruno", False) + via = main._make_session("bruno", via_draco=True) + assert main._session_info(requete(via)) == ("bruno", True) + # Marqueur ajouté à la main sur un jeton natif : signature fausse. + pirate = natif.rsplit(".", 1)[0] + ".draco." + natif.rsplit(".", 1)[1] + assert main._session_info(requete(pirate)) is None