feat: #107 configuration - gestion des clés API & MCP (création/révocation, expiration 1j/1mois/6mois/1an/sans fin, une clé pour API REST + serveur MCP, dernière utilisation, store sans secret persisté; fix révocation longue durée) + script token MCP

tests/test_api_tokens.py

@@ -0,0 +1,271 @@
+# 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")