272 lines
11 KiB
Python
272 lines
11 KiB
Python
# 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")
|