125 lines
3.9 KiB
Python
125 lines
3.9 KiB
Python
"""Validation locale du cookie SSO ``draco_session`` — Fusion ACC-3.2.
|
|
|
|
Kanbante valide localement le jeton émis par ObsiGate (l'émettrice,
|
|
ACC-3.1) : JWS HS256 vérifié avec le secret partagé
|
|
(``DRACO_SESSION_SECRET``, ``_PREV`` accepté en rotation),
|
|
algorithme épinglé, expiration, et liste de révocation de
|
|
l'émettrice (cache 60 s, interrogée seulement si un cookie draco
|
|
est présenté). Vérification hmac/hashlib — mêmes sémantiques que
|
|
les validateurs de TikTak (0.52.0) et Flowdeck (7.74.0).
|
|
|
|
Le pont ``GET /auth/draco`` (app/main.py) convertit un cookie valide
|
|
en session Kanbante native. Kanbante est mono-utilisateur
|
|
(identifiants configurés par l'environnement) : un jeton valide
|
|
désigne l'opérateur configuré — pas de table de liens. Sans secret
|
|
configuré, module inerte.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import logging
|
|
import time
|
|
|
|
from app.config import get_settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
COOKIE_NAME = "draco_session"
|
|
_REVOCATIONS_TTL = 60 # secondes de cache, comme le prévoit le design
|
|
|
|
_revoked_cache: dict[str, int] = {}
|
|
_revoked_fetched_at: float = 0.0
|
|
|
|
|
|
def enabled() -> bool:
|
|
return bool(get_settings().draco_secret)
|
|
|
|
|
|
def _b64decode(segment: str) -> bytes:
|
|
return base64.urlsafe_b64decode(segment + "=" * (-len(segment) % 4))
|
|
|
|
|
|
def verify_token(token: str) -> dict | None:
|
|
"""Vérifie un JWS HS256 (signature, alg épinglé, expiration)."""
|
|
settings = get_settings()
|
|
if not token or not settings.draco_secret:
|
|
return None
|
|
parts = token.split(".")
|
|
if len(parts) != 3:
|
|
return None
|
|
try:
|
|
header = json.loads(_b64decode(parts[0]))
|
|
if not isinstance(header, dict) or header.get("alg") != "HS256":
|
|
return None
|
|
signing_input = f"{parts[0]}.{parts[1]}".encode("ascii")
|
|
signature = _b64decode(parts[2])
|
|
claims = json.loads(_b64decode(parts[1]))
|
|
except Exception:
|
|
return None
|
|
if not isinstance(claims, dict):
|
|
return None
|
|
for secret in (settings.draco_secret, settings.draco_secret_prev):
|
|
if not secret:
|
|
continue
|
|
attendu = hmac.new(secret.encode("utf-8"), signing_input,
|
|
hashlib.sha256).digest()
|
|
if hmac.compare_digest(attendu, signature):
|
|
break
|
|
else:
|
|
return None
|
|
try:
|
|
if int(claims.get("exp", 0)) <= int(time.time()):
|
|
return None
|
|
except (TypeError, ValueError):
|
|
return None
|
|
return claims
|
|
|
|
|
|
# ── Révocations (cache 60 s alimenté par l'émettrice) ───────────────────
|
|
|
|
def _refresh_revocations() -> None:
|
|
global _revoked_fetched_at
|
|
now = time.monotonic()
|
|
if now - _revoked_fetched_at < _REVOCATIONS_TTL:
|
|
return
|
|
_revoked_fetched_at = now
|
|
settings = get_settings()
|
|
try:
|
|
import httpx
|
|
|
|
headers = {}
|
|
if settings.sso_machine_key:
|
|
headers["X-Draco-Machine-Key"] = settings.sso_machine_key
|
|
r = httpx.get(f"{settings.sso_url}/api/sso/revocations",
|
|
headers=headers, timeout=2.0)
|
|
if r.status_code == 200:
|
|
_revoked_cache.clear()
|
|
for item in r.json().get("revoked", []):
|
|
_revoked_cache[str(item["jti"])] = int(item["exp"])
|
|
except Exception as exc: # l'émettrice injoignable ne bloque pas le SSO
|
|
logger.debug("Révocations draco indisponibles: %s", exc)
|
|
|
|
|
|
def is_revoked(jti: str) -> bool:
|
|
_refresh_revocations()
|
|
return jti in _revoked_cache
|
|
|
|
|
|
def claims_for(request) -> dict | None:
|
|
"""Claims valides du cookie draco de la requête (révocation incluse)."""
|
|
if not enabled():
|
|
return None
|
|
claims = verify_token(request.cookies.get(COOKIE_NAME, ""))
|
|
if not claims:
|
|
return None
|
|
jti = str(claims.get("jti", ""))
|
|
if jti and is_revoked(jti):
|
|
return None
|
|
if not claims.get("sub"):
|
|
return None
|
|
return claims
|