292 lines
11 KiB
Python
292 lines
11 KiB
Python
# tests/test_hub_roles.py
|
|
# Hub HUB-19 (propagation, #228) — Rôles par application : client du
|
|
# registre du Hub (parsing, cache 60 s par sub, repli systématique) et
|
|
# effets sur le chemin draco d'ObsiGate (refus « aucun », garde
|
|
# administrateur des routes /api/auth/admin/* pour le canal draco).
|
|
# Contrat posé par l'hôte pilote ShaarIt 0.7.0 (décision D3(a)).
|
|
|
|
import io
|
|
import json
|
|
import time
|
|
import urllib.error
|
|
|
|
import pytest
|
|
|
|
from backend import draco_session, hub_roles
|
|
|
|
USER = {"id": "admin-1", "username": "admin", "display_name": "Admin"}
|
|
USER_SIMPLE = {"id": "user-1", "username": "normaluser", "display_name": "normal"}
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _hub_env(monkeypatch):
|
|
"""Cache vierge + Hub de test explicite pour chaque test."""
|
|
hub_roles.reinitialiser_cache()
|
|
monkeypatch.setenv("DRACO_HUB_URL", "https://hub-test.dracodev.net")
|
|
yield
|
|
hub_roles.reinitialiser_cache()
|
|
|
|
|
|
def _reponse(role, statut=200, corps=None):
|
|
"""Fausse réponse urlopen portant un payload du contrat D3(a)."""
|
|
if corps is None:
|
|
corps = {"app": "obsigate", "sub": "admin-1", "role": role,
|
|
"explicite": True, "ttl_secondes": 60}
|
|
charge = json.dumps(corps).encode("utf-8")
|
|
|
|
class _Resp:
|
|
status = statut
|
|
|
|
def read(self):
|
|
return charge
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
return _Resp()
|
|
|
|
|
|
def _patch_hub(monkeypatch, role, appels):
|
|
"""Remplace la consultation HTTP par un rôle fixe (compte les appels)."""
|
|
def _faux(url, cookie):
|
|
appels.append((url, cookie))
|
|
return role
|
|
monkeypatch.setattr(hub_roles, "_consulter_hub", _faux)
|
|
|
|
|
|
# ── Client : parsing du contrat ─────────────────────────────────────────
|
|
|
|
def test_consultation_parse_le_contrat(monkeypatch):
|
|
vu = {}
|
|
|
|
def _faux_urlopen(requete, timeout=None):
|
|
vu["url"] = requete.full_url
|
|
vu["cookie"] = requete.get_header("Cookie")
|
|
vu["accept"] = requete.get_header("Accept")
|
|
vu["timeout"] = timeout
|
|
return _reponse("administrateur")
|
|
|
|
monkeypatch.setattr(hub_roles.urllib.request, "urlopen", _faux_urlopen)
|
|
role = hub_roles.role_pour("admin-1", "draco_session=jeton; autre=1")
|
|
assert role == "administrateur"
|
|
assert vu["url"] == "https://hub-test.dracodev.net/api/roles/obsigate"
|
|
assert vu["cookie"] == "draco_session=jeton; autre=1" # Cookie complet relayé
|
|
assert vu["accept"] == "application/json"
|
|
assert vu["timeout"] == 2 # délai du contrat
|
|
|
|
|
|
@pytest.mark.parametrize("role", ["aucun", "utilisateur", "administrateur"])
|
|
def test_consultation_roles_reconnus(monkeypatch, role):
|
|
monkeypatch.setattr(hub_roles.urllib.request, "urlopen",
|
|
lambda requete, timeout=None: _reponse(role))
|
|
assert hub_roles.role_pour("admin-1", "c=1") == role
|
|
|
|
|
|
def test_consultation_role_hors_contrat_inconnu(monkeypatch):
|
|
monkeypatch.setattr(hub_roles.urllib.request, "urlopen",
|
|
lambda requete, timeout=None: _reponse("super-admin"))
|
|
assert hub_roles.role_pour("admin-1", "c=1") is None
|
|
|
|
|
|
def test_consultation_json_illisible_inconnu(monkeypatch):
|
|
class _Resp:
|
|
status = 200
|
|
|
|
def read(self):
|
|
return b"pas du json"
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
monkeypatch.setattr(hub_roles.urllib.request, "urlopen",
|
|
lambda requete, timeout=None: _Resp())
|
|
assert hub_roles.role_pour("admin-1", "c=1") is None
|
|
|
|
|
|
def test_consultation_erreur_http_inconnu(monkeypatch):
|
|
def _erreur(requete, timeout=None):
|
|
raise urllib.error.HTTPError(requete.full_url, 500, "boom", {}, io.BytesIO(b""))
|
|
|
|
monkeypatch.setattr(hub_roles.urllib.request, "urlopen", _erreur)
|
|
assert hub_roles.role_pour("admin-1", "c=1") is None
|
|
|
|
|
|
def test_consultation_hub_injoignable_inconnu(monkeypatch):
|
|
def _panne(requete, timeout=None):
|
|
raise urllib.error.URLError("injoignable")
|
|
|
|
monkeypatch.setattr(hub_roles.urllib.request, "urlopen", _panne)
|
|
assert hub_roles.role_pour("admin-1", "c=1") is None
|
|
|
|
|
|
def test_url_hub_derivee_par_hote_soeur(monkeypatch):
|
|
# Sans variable explicite : dérivation depuis l'hôte de l'agent
|
|
# (agent-dev → hub-dev), comme la barre commune (#223).
|
|
monkeypatch.delenv("DRACO_HUB_URL")
|
|
monkeypatch.setenv("DRACO_AGENT_URL", "https://agent-dev.dracodev.net")
|
|
appels = []
|
|
_patch_hub(monkeypatch, "utilisateur", appels)
|
|
assert hub_roles.role_pour("admin-1", "c=1") == "utilisateur"
|
|
assert appels[0][0] == "https://hub-dev.dracodev.net/api/roles/obsigate"
|
|
|
|
|
|
def test_url_hub_defaut_officiel(monkeypatch):
|
|
monkeypatch.delenv("DRACO_HUB_URL")
|
|
monkeypatch.delenv("DRACO_AGENT_URL", raising=False)
|
|
monkeypatch.delenv("AGENT_EVENEMENTS_URL", raising=False)
|
|
appels = []
|
|
_patch_hub(monkeypatch, "utilisateur", appels)
|
|
hub_roles.role_pour("admin-1", "c=1")
|
|
assert appels[0][0] == "https://hub.dracodev.net/api/roles/obsigate"
|
|
|
|
|
|
# ── Client : cache 60 s par sub, horodatage avant l'appel ──────────────
|
|
|
|
def test_cache_par_sub_et_ttl(monkeypatch):
|
|
appels = []
|
|
_patch_hub(monkeypatch, "utilisateur", appels)
|
|
assert hub_roles.role_pour("admin-1", "c=1") == "utilisateur"
|
|
assert hub_roles.role_pour("admin-1", "c=1") == "utilisateur"
|
|
assert len(appels) == 1 # 2e lecture : cache
|
|
assert hub_roles.role_pour("user-1", "c=2") == "utilisateur"
|
|
assert len(appels) == 2 # autre sub : nouvel appel
|
|
# TTL expiré → nouvel appel.
|
|
hub_roles._cache_at = time.monotonic() - hub_roles.CACHE_TTL_SECONDS - 1
|
|
assert hub_roles.role_pour("admin-1", "c=1") == "utilisateur"
|
|
assert len(appels) == 3
|
|
|
|
|
|
def test_echec_conserve_ancien_cache_sans_rafale(monkeypatch):
|
|
appels = []
|
|
_patch_hub(monkeypatch, "administrateur", appels)
|
|
assert hub_roles.role_pour("admin-1", "c=1") == "administrateur"
|
|
assert len(appels) == 1
|
|
# Le Hub tombe ; le TTL expire : un appel est tenté, échoue, et
|
|
# l'horodatage posé avant l'appel empêche la rafale — l'ancien
|
|
# rôle connu n'est ni effacé ni réinterrogé aussitôt.
|
|
hub_roles._cache_at = time.monotonic() - hub_roles.CACHE_TTL_SECONDS - 1
|
|
_patch_hub(monkeypatch, None, appels)
|
|
assert hub_roles.role_pour("user-1", "c=2") is None
|
|
assert len(appels) == 2
|
|
# Dans la fenêtre qui suit l'échec, l'ancien cache sert toujours.
|
|
assert hub_roles.role_pour("admin-1", "c=1") == "administrateur"
|
|
assert len(appels) == 2
|
|
|
|
|
|
def test_sub_vide_aucun_appel(monkeypatch):
|
|
appels = []
|
|
_patch_hub(monkeypatch, "utilisateur", appels)
|
|
assert hub_roles.role_pour("", "c=1") is None
|
|
assert appels == []
|
|
|
|
|
|
# ── Intégration : chemin draco (admin_client, secret draco posé) ──────
|
|
|
|
@pytest.fixture()
|
|
def _draco_env(monkeypatch, tmp_path):
|
|
monkeypatch.setenv("DRACO_SESSION_SECRET", "secret-de-test")
|
|
monkeypatch.setattr(draco_session, "REVOKED_FILE",
|
|
tmp_path / "draco_revoked.json")
|
|
monkeypatch.setattr(draco_session, "ISSUED_FILE",
|
|
tmp_path / "draco_issued.json")
|
|
draco_session._revoked = None
|
|
draco_session._issued = None
|
|
yield
|
|
draco_session._revoked = None
|
|
draco_session._issued = None
|
|
|
|
|
|
def _pose_cookie_draco(client, user):
|
|
token, _jti, _exp = draco_session.issue_token(user)
|
|
client.cookies.set("draco_session", token)
|
|
return token
|
|
|
|
|
|
def test_draco_aucun_refuse_session_explicite(admin_client, _draco_env, monkeypatch):
|
|
appels = []
|
|
_patch_hub(monkeypatch, "aucun", appels)
|
|
_pose_cookie_draco(admin_client, USER)
|
|
r = admin_client.get("/api/auth/me")
|
|
assert r.status_code == 403
|
|
assert "niveau « aucun »" in r.json()["detail"]
|
|
assert "registre des rôles du Hub" in r.json()["detail"]
|
|
assert appels and appels[0][1].find("draco_session=") >= 0 # Cookie relayé
|
|
|
|
|
|
def test_draco_aucun_mais_connexion_native_intacte(admin_client, _draco_env, monkeypatch):
|
|
# L'authentification native (émettrice, accès de secours) n'est
|
|
# jamais touchée par le registre : le login par mot de passe du
|
|
# même compte passe, même avec un niveau « aucun » au Hub.
|
|
_patch_hub(monkeypatch, "aucun", [])
|
|
r = admin_client.post("/api/auth/login",
|
|
json={"username": "admin", "password": "chab30"})
|
|
assert r.status_code == 200, r.text
|
|
# Et la session native ainsi établie ouvre /me (canal natif, le
|
|
# rôle du Hub ne s'applique pas).
|
|
r = admin_client.get("/api/auth/me")
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["username"] == "admin"
|
|
|
|
|
|
def test_draco_role_inconnu_comportement_actuel(admin_client, _draco_env, monkeypatch):
|
|
_patch_hub(monkeypatch, None, []) # Hub injoignable
|
|
_pose_cookie_draco(admin_client, USER)
|
|
r = admin_client.get("/api/auth/me")
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["username"] == "admin"
|
|
# Canal draco + rôle inconnu : les routes admin gardent les
|
|
# règles actuelles (le rôle local admin suffit).
|
|
r = admin_client.get("/api/auth/admin/users")
|
|
assert r.status_code == 200, r.text
|
|
|
|
|
|
def test_draco_utilisateur_admin_local_refuse_admin(admin_client, _draco_env, monkeypatch):
|
|
# Compte local admin authentifié par draco, mais niveau Hub
|
|
# « utilisateur » : /api/auth/admin/* exige « administrateur ».
|
|
_patch_hub(monkeypatch, "utilisateur", [])
|
|
_pose_cookie_draco(admin_client, USER)
|
|
r = admin_client.get("/api/auth/me")
|
|
assert r.status_code == 200, r.text # session draco OK
|
|
r = admin_client.get("/api/auth/admin/users")
|
|
assert r.status_code == 403
|
|
assert "administrateur" in r.json()["detail"]
|
|
|
|
|
|
def test_draco_administrateur_hub_admin_ok(admin_client, _draco_env, monkeypatch):
|
|
_patch_hub(monkeypatch, "administrateur", [])
|
|
_pose_cookie_draco(admin_client, USER)
|
|
r = admin_client.get("/api/auth/admin/users")
|
|
assert r.status_code == 200, r.text
|
|
assert any(u["username"] == "admin" for u in r.json())
|
|
|
|
|
|
def test_draco_utilisateur_simple_admin_refuse_comme_avant(admin_client, _draco_env, monkeypatch):
|
|
# Compte local non admin, niveau Hub « administrateur » : le
|
|
# registre n'élève aucun privilège local — refus par la règle
|
|
# actuelle (rôle local), comme avant HUB-19.
|
|
_patch_hub(monkeypatch, "administrateur", [])
|
|
_pose_cookie_draco(admin_client, USER_SIMPLE)
|
|
r = admin_client.get("/api/auth/admin/users")
|
|
assert r.status_code == 403
|
|
assert r.json()["detail"] == "Accès admin requis"
|
|
|
|
|
|
def test_natif_admin_sans_draco_jamais_bloque_par_hub(admin_client, _draco_env, monkeypatch):
|
|
# Session native (login) + niveau Hub « utilisateur » : le canal
|
|
# n'est pas draco, la garde du registre ne s'applique pas.
|
|
_patch_hub(monkeypatch, "utilisateur", [])
|
|
r = admin_client.post("/api/auth/login",
|
|
json={"username": "admin", "password": "chab30"})
|
|
assert r.status_code == 200, r.text
|
|
# Le jeton natif valide prime : get_current_user ne passe pas par
|
|
# le repli draco, le dict utilisateur n'a ni _draco_sub ni
|
|
# _hub_role — la garde du registre ne peut pas s'appliquer.
|
|
r = admin_client.get("/api/auth/admin/users")
|
|
assert r.status_code == 200, r.text
|