# tests/test_api_tokens.py — Feature #107 : jetons API/MCP gérés dans la config. """Couvre : - création / liste / révocation via /api/auth/tokens ; - le même jeton authentifie l'API REST ET le serveur MCP /mcp ; - choix d'expiration 1d/30d/180d/365d/never (revoked_tokens et exp) ; - révocation immédiate et persistante (pas de retour à la vie après 7 jours) ; - isolation par utilisateur, auth requise. """ import json import os import time import pytest from fastapi.testclient import TestClient ACCEPT = "application/json, text/event-stream" @pytest.fixture def tokens_client(tmp_path, monkeypatch): """Auth-enabled TestClient in an isolated data dir (admin / chab30).""" import shutil from pathlib import Path data_dir = tmp_path / "data" data_dir.mkdir() from backend.auth.password import hash_password users = { "version": 1, "users": { "admin": { "id": "admin-1", "username": "admin", "display_name": "admin", "password_hash": hash_password("chab30"), "role": "admin", "vaults": ["*"], "active": True, "created_at": "2026-01-01T00:00:00", }, "bob": { "id": "bob-1", "username": "bob", "display_name": "bob", "password_hash": hash_password("chab30"), "role": "user", "vaults": ["TestVault"], "active": True, "created_at": "2026-01-01T00:00:00", }, }, } (data_dir / "users.json").write_text(json.dumps(users), encoding="utf-8") src_secret = Path("data/secret.key") if src_secret.exists(): shutil.copy2(str(src_secret), str(data_dir / "secret.key")) vault = os.path.abspath("test_vault") orig_cwd = os.getcwd() os.chdir(str(tmp_path)) os.environ["VAULT_1_NAME"] = "TestVault" os.environ["VAULT_1_PATH"] = vault os.environ["OBSIGATE_AUTH_ENABLED"] = "true" os.environ["OBSIGATE_WATCHER_ENABLED"] = "false" import backend.main backend.main._load_config = lambda: {"watcher_enabled": False} from backend.indexer import build_index, index import asyncio for key in list(index.keys()): del index[key] loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) loop.run_until_complete(build_index()) from backend.search import init_inverted_index init_inverted_index() # Fresh revoked-token state per test (module caches a global map). import backend.auth.jwt_handler as jh jh._revoked_jtis_backup = getattr(jh, "_revoked_map", {}) jh._revoked_map = {} jh._revoked_loaded = False jh._touch_last_write.clear() # The MCP session manager can only run() once per instance/event-loop # (same reset as tests/test_mcp.py::mcp_client) — otherwise this file's # /mcp tests 500 when an earlier test already bound it to a dead loop. backend.main.mcp_app._manager = None backend.main.mcp_app._run_task = None backend.main.mcp_app._start_lock = None with TestClient(backend.main.app) as client: yield client backend.main.mcp_app._manager = None backend.main.mcp_app._run_task = None backend.main.mcp_app._start_lock = None jh._revoked_map = {} jh._revoked_loaded = False os.chdir(orig_cwd) shutil.rmtree(str(tmp_path), ignore_errors=True) for k in ["VAULT_1_NAME", "VAULT_1_PATH", "OBSIGATE_AUTH_ENABLED", "OBSIGATE_WATCHER_ENABLED"]: os.environ.pop(k, None) _TEST_PW = "chab" + "30" def _login(client, username="admin", password=_TEST_PW): resp = client.post("/api/auth/login", json={"username": username, "password": password}) assert resp.status_code == 200, resp.text return resp.json()["access_token"] def _hdr(token): return {"Authorization": f"Bearer {token}"} def _create(client, token, name="claude desktop", expiry="30d"): resp = client.post("/api/auth/tokens", json={"name": name, "expiry": expiry}, headers=_hdr(token)) assert resp.status_code == 200, resp.text return resp.json() # ═══════════════════════════════════════════════════════════════════ class TestCreateAndList: def test_requires_auth(self, tokens_client): assert tokens_client.get("/api/auth/tokens").status_code == 401 def test_create_returns_token_once(self, tokens_client): tok = _login(tokens_client) created = _create(tokens_client, tok) assert created["token"].count(".") == 2 # JWT assert created["name"] == "claude desktop" assert created["expires_at"] is not None listing = tokens_client.get("/api/auth/tokens", headers=_hdr(tok)).json() assert [t["jti"] for t in listing["tokens"]] == [created["jti"]] # Le secret n'est JAMAIS stocké/restitués en liste. assert "token" not in listing["tokens"][0] def test_expiry_choices(self, tokens_client): tok = _login(tokens_client) from backend.auth.jwt_handler import decode_token expected = {"1d": 86400, "30d": 2592000, "180d": 15552000, "365d": 31536000} for key, secs in expected.items(): c = _create(tokens_client, tok, name=key, expiry=key) payload = decode_token(c["token"]) assert payload["exp"] - payload["iat"] == secs never = _create(tokens_client, tok, name="never", expiry="never") payload = decode_token(never["token"]) assert "exp" not in payload and never["expires_at"] is None def test_invalid_expiry_rejected(self, tokens_client): tok = _login(tokens_client) resp = tokens_client.post("/api/auth/tokens", json={"name": "x", "expiry": "5minutes"}, headers=_hdr(tok)) assert resp.status_code == 400 class TestTokenWorksOnApiAndMcp: """Le point #107 : une seule clé pour l'API REST et le serveur MCP.""" def test_authenticates_rest_api(self, tokens_client): api_tok = _login(tokens_client) key = _create(tokens_client, api_tok)["token"] me = tokens_client.get("/api/auth/me", headers=_hdr(key)) assert me.status_code == 200 assert me.json()["username"] == "admin" def test_mcp_endpoint_rejects_anonymous(self, tokens_client): resp = tokens_client.post( "/mcp", content=json.dumps({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}}), headers={"Accept": ACCEPT, "Content-Type": "application/json"}, ) assert resp.status_code == 401 def test_same_key_authenticates_mcp(self, tokens_client): api_tok = _login(tokens_client) key = _create(tokens_client, api_tok)["token"] resp = tokens_client.post( "/mcp", content=json.dumps({ "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2025-03-26", "capabilities": {}, "clientInfo": {"name": "pytest", "version": "1.0"}}, }), headers={"Accept": ACCEPT, "Content-Type": "application/json", "Authorization": f"Bearer {key}"}, ) assert resp.status_code == 200, resp.text assert resp.headers.get("mcp-session-id") def test_api_token_vault_scope_from_login_snapshot(self, tokens_client): # bob (user role, vaults=[TestVault]) creates a token; /api/auth/me ok. bob = _login(tokens_client, "bob") key = _create(tokens_client, bob, name="bob-key")["token"] assert tokens_client.get("/api/auth/me", headers=_hdr(key)).status_code == 200 # admin's listing must not see bob's token. admin = _login(tokens_client) names = [t["name"] for t in tokens_client.get("/api/auth/tokens", headers=_hdr(admin)).json()["tokens"]] assert "bob-key" not in names class TestRevoke: def test_revoke_kills_api_and_mcp_immediately(self, tokens_client): api_tok = _login(tokens_client) created = _create(tokens_client, api_tok) key, jti = created["token"], created["jti"] assert tokens_client.get("/api/auth/me", headers=_hdr(key)).status_code == 200 resp = tokens_client.delete(f"/api/auth/tokens/{jti}", headers=_hdr(api_tok)) assert resp.status_code == 200 # API assert tokens_client.get("/api/auth/me", headers=_hdr(key)).status_code == 401 # MCP — même clé révoquée = 401 aussi mcp = tokens_client.post( "/mcp", content=json.dumps({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}}), headers={"Accept": ACCEPT, "Content-Type": "application/json", "Authorization": f"Bearer {key}"}, ) assert mcp.status_code == 401 def test_revoke_unknown_is_404(self, tokens_client): api_tok = _login(tokens_client) assert tokens_client.delete("/api/auth/tokens/nope", headers=_hdr(api_tok)).status_code == 404 def test_revocation_survives_7day_cleanup_for_long_lived(self, tokens_client): """Un jeton 'never' révoqué ne doit PAS revenir à la vie : la révocation est bornée à l'expiration du jeton lui-même (infini ici).""" import backend.auth.jwt_handler as jh api_tok = _login(tokens_client) created = _create(tokens_client, api_tok, name="forever", expiry="never") tokens_client.delete(f"/api/auth/tokens/{created['jti']}", headers=_hdr(api_tok)) until = jh._revoked_map[created["jti"]] # 30+ ans devant nous → survit à tout nettoyage "7 days max". assert until > time.time() + 365 * 24 * 3600 # Reload depuis le disque → toujours révoqué. jh._revoked_map = {} jh._revoked_loaded = False assert jh.is_token_revoked(created["jti"]) is True def test_expired_token_flagged_in_list(self, tokens_client): from backend.auth.jwt_handler import _load_api_tokens, _save_api_tokens api_tok = _login(tokens_client) created = _create(tokens_client, api_tok, name="old", expiry="1d") # Forcer l'expiration côté registre + jeton (via iat/exp passés). data = _load_api_tokens() data["tokens"][created["jti"]]["expires_at"] = int(time.time()) - 10 _save_api_tokens(data) listing = tokens_client.get("/api/auth/tokens", headers=_hdr(api_tok)).json() assert listing["tokens"][0]["expired"] is True class TestRevokedStoreFormat: def test_migration_from_list_format(self, tmp_path, monkeypatch): """Ancien format (set) et nouveau (dict jti->until) coexistent au load.""" from backend.auth.jwt_handler import ( REVOKED_TOKENS_FILE, _load_revoked, is_token_revoked, ) import backend.auth.jwt_handler as jh REVOKED_TOKENS_FILE.parent.mkdir(parents=True, exist_ok=True) future = int(time.time()) + 3600 REVOKED_TOKENS_FILE.write_text(json.dumps( {"alive": future, "dead": int(time.time()) - 10})) jh._revoked_map, jh._revoked_loaded = {}, False _load_revoked() assert is_token_revoked("alive") and not is_token_revoked("dead")