Kanbante 0.46.2 — HUB-19 : rôles du Hub propagés (pont draco + /admin) #86

Merged
bruno merged 6 commits from hub19-roles-kanbante into main 2026-11-06 11:32:27 -05:00
6 changed files with 650 additions and 12 deletions
+10
View File
@@ -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) ## 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**. > **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**.
+1 -1
View File
@@ -1 +1 @@
0.46.1 0.46.2
+120
View File
@@ -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
+107 -9
View File
@@ -263,23 +263,48 @@ def _session_secret() -> bytes:
def _make_session(user: str) -> str: def _make_session(user: str, via_draco: bool = False) -> str:
"""Signed session cookie (0.22.0): user.expiry.hmac.""" """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 exp = int(time.time()) + SESSION_DAYS * 86400
charge = f"{user}.{exp}.draco" if via_draco else f"{user}.{exp}"
sig = hmac.new( sig = hmac.new(
_session_secret(), f"{user}.{exp}".encode("utf-8"), hashlib.sha256 _session_secret(), charge.encode("utf-8"), hashlib.sha256
).hexdigest() ).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, "") raw = request.cookies.get(SESSION_COOKIE, "")
user, _, rest = raw.partition(".") morceaux = raw.split(".")
exp, _, sig = rest.partition(".") 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: if not user or not exp or not sig:
return None return None
expected = hmac.new( expected = hmac.new(
_session_secret(), f"{user}.{exp}".encode("utf-8"), hashlib.sha256 _session_secret(), charge.encode("utf-8"), hashlib.sha256
).hexdigest() ).hexdigest()
if not secrets.compare_digest(sig, expected): if not secrets.compare_digest(sig, expected):
return None return None
@@ -288,6 +313,14 @@ def _session_user(request: Request) -> str | None:
return None return None
except ValueError: except ValueError:
return None 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 # 0.24.0 : la session peut appartenir à n'importe quel
# compte de la base (résolu ensuite par _current_user) ou, # compte de la base (résolu ensuite par _current_user) ou,
# en mode env pur, à l'opérateur d'environnement. # en mode env pur, à l'opérateur d'environnement.
@@ -299,6 +332,12 @@ def _session_user(request: Request) -> str | None:
return 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: def _basic_ok(request: Request) -> bool:
settings = get_settings() settings = get_settings()
header = request.headers.get("authorization", "") 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 Un ``sub`` inconnu n'ouvre plus jamais la session de
l'opérateur d'environnement ni celle d'un compte non lié. l'opérateur d'environnement ni celle d'un compte non lié.
Inerte sans ``DRACO_SESSION_SECRET``. 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 accounts as accounts_service
from app import draco from app import draco
from app import hub_roles
settings = get_settings() settings = get_settings()
claims = draco.claims_for(request) claims = draco.claims_for(request)
@@ -608,6 +659,28 @@ def draco_bridge(request: Request, next: str = "/"):
sub = str(claims.get("sub") or "") sub = str(claims.get("sub") or "")
if not sub: if not sub:
return RedirectResponse("/login", status_code=303) 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. # Déjà connecté nativement : lier le sub à ce compte.
current = _current_user(request) current = _current_user(request)
if current is not None and current.get("id") is not None: 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) return RedirectResponse(_safe_next(next), status_code=303)
# Correspondance, amorçage ou création JIT : ouvrir la # Correspondance, amorçage ou création JIT : ouvrir la
# session du compte que ce sub désigne — et seulement lui. # 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) user = accounts_service.resolve_draco_user(claims, provisionner=True)
if user is not None: if user is not None:
response = RedirectResponse(_safe_next(next), status_code=303) response = RedirectResponse(_safe_next(next), status_code=303)
response.set_cookie( response.set_cookie(
SESSION_COOKIE, SESSION_COOKIE,
_make_session(user["username"]), _make_session(user["username"], via_draco=True),
max_age=SESSION_DAYS * 86400, max_age=SESSION_DAYS * 86400,
httponly=True, httponly=True,
samesite="lax", samesite="lax",
@@ -1006,6 +1082,28 @@ def _require_admin(request: Request) -> dict | None:
user = _current_user(request) user = _current_user(request)
if user is None or user.get("role") != "admin": if user is None or user.get("role") != "admin":
return None 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 return user
+37 -2
View File
@@ -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. **Mon profil → Liaison draco (SSO)** et la délier lui-même.
> Le registre matriciel des rôles par application côté Hub > Le registre matriciel des rôles par application côté Hub
> (carte HUB-19) lira cette correspondance ; la présente carte ne > (carte HUB-19) lit cette correspondance ; la 0.46.0 ne livrait
> livre que le côté Kanbante, qui est la source de vérité locale. > 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 ## Surfaces voisines
+375
View File
@@ -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