# tests/test_scoped_tokens.py — #199 (Gitea #2) : clés API à portée limitée. """Couvre : - création d'une clé limitée (``vaults`` + ``scope=files``) : portée exposée dans le record et la liste ; - la clé limitée lit/écrit les fichiers de son vault (REST) ; - tout autre vault est refusé (403, via ``check_vault_access``) ; - les endpoints hors périmètre sont refusés (403, via le middleware — y compris la gestion des clés elles-mêmes, anti-escalade) ; - les clés sans portée fonctionnent comme avant (non-régression) ; - validation à la création (scope inconnu → 400, vault inconnu → 400, vault non autorisé → 403) ; - niveau outils MCP : ``call_tool`` autorise les outils fichiers et refuse les autres (``token_scope_denied``), y compris les outils sans vault. """ import json import os import shutil from pathlib import Path import pytest from fastapi.testclient import TestClient from backend.tools.api import ToolPermissionError, call_tool from tests.test_tools import _ctx @pytest.fixture def scoped_client(tmp_path, monkeypatch): """Auth-enabled TestClient in an isolated data dir (admin / chab30).""" 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() import backend.auth.jwt_handler as jh jh._revoked_map = {} jh._revoked_loaded = False jh._touch_last_write.clear() # Pas de context manager : le lifespan relancerait un rebuild d'index en # tâche de fond qui vide l'index pendant les appels outils directs # (même pattern que tests/conftest.py). client = TestClient(backend.main.app, raise_server_exceptions=False) try: yield client finally: 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="scoped", expiry="30d", **extra): body = {"name": name, "expiry": expiry, **extra} resp = client.post("/api/auth/tokens", json=body, headers=_hdr(token)) assert resp.status_code == 200, resp.text return resp.json() class TestCreateScoped: def test_record_exposes_scope_and_vaults(self, scoped_client): admin = _login(scoped_client) created = _create(scoped_client, admin, vaults=["TestVault"], scope="files") assert created["scope"] == "files" assert created["vaults"] == ["TestVault"] listing = scoped_client.get("/api/auth/tokens", headers=_hdr(admin)).json() entry = next(t for t in listing["tokens"] if t["jti"] == created["jti"]) assert entry["scope"] == "files" and entry["vaults"] == ["TestVault"] def test_full_scope_default(self, scoped_client): admin = _login(scoped_client) created = _create(scoped_client, admin) assert created["scope"] == "full" def test_invalid_scope_rejected(self, scoped_client): admin = _login(scoped_client) resp = scoped_client.post("/api/auth/tokens", json={"name": "x", "expiry": "30d", "scope": "root"}, headers=_hdr(admin)) assert resp.status_code == 400 def test_unknown_vault_rejected(self, scoped_client): admin = _login(scoped_client) resp = scoped_client.post("/api/auth/tokens", json={"name": "x", "expiry": "30d", "vaults": ["Nope"]}, headers=_hdr(admin)) assert resp.status_code == 400 def test_vault_outside_creator_access_rejected(self, scoped_client): bob = _login(scoped_client, "bob") resp = scoped_client.post("/api/auth/tokens", json={"name": "x", "expiry": "30d", "vaults": ["Autre"]}, headers=_hdr(bob)) assert resp.status_code in (400, 403) class TestScopedRestAccess: def test_reads_and_writes_own_vault(self, scoped_client): import uuid admin = _login(scoped_client) key = _create(scoped_client, admin, vaults=["TestVault"], scope="files")["token"] assert scoped_client.get("/api/vaults", headers=_hdr(key)).status_code == 200 assert scoped_client.get("/api/vault/TestVault/paths", headers=_hdr(key)).status_code == 200 assert scoped_client.get("/api/vault/TestVault/files", headers=_hdr(key)).status_code == 200 # Fichier unique par exécution + nettoyage : le vault de test est le # vrai répertoire du dépôt, on n'y laisse aucune trace. path = f"scoped-sync-probe-{uuid.uuid4().hex[:8]}.md" try: created = scoped_client.post("/api/file/TestVault", json={"path": path, "content": "# a\n"}, headers=_hdr(key)) assert created.status_code == 200, created.text raw = scoped_client.get("/api/file/TestVault/raw", params={"path": path}, headers=_hdr(key)) assert raw.status_code == 200 saved = scoped_client.put("/api/file/TestVault/save", params={"path": path}, json={"content": "# a\nmodifié\n"}, headers=_hdr(key)) assert saved.status_code == 200, saved.text finally: scoped_client.delete("/api/file/TestVault", params={"path": path}, headers=_hdr(key)) def test_other_vault_refused(self, scoped_client): admin = _login(scoped_client) key = _create(scoped_client, admin, vaults=["TestVault"], scope="files")["token"] assert scoped_client.get("/api/browse/Nope", headers=_hdr(key)).status_code == 403 # Sanity : la clé pleine du même admin passe le garde vault (404 ensuite). full = _create(scoped_client, admin, name="full")["token"] assert scoped_client.get("/api/browse/Nope", headers=_hdr(full)).status_code == 404 def test_out_of_scope_endpoints_refused(self, scoped_client): admin = _login(scoped_client) key = _create(scoped_client, admin, vaults=["TestVault"], scope="files")["token"] h = _hdr(key) assert scoped_client.get("/api/config", headers=h).status_code == 403 assert scoped_client.get("/api/webhooks", headers=h).status_code == 403 assert scoped_client.get("/api/auth/tokens", headers=h).status_code == 403 assert scoped_client.post("/api/search/replace", json={"query": "a", "replacement": "b", "vault": "TestVault"}, headers=h).status_code == 403 assert scoped_client.post("/api/vaults/add", json={"name": "X", "path": "/tmp/x"}, headers=h).status_code == 403 # … mais la recherche et la liste des vaults restent ouvertes. assert scoped_client.get("/api/search", params={"q": "test"}, headers=h).status_code == 200 def test_full_token_unaffected(self, scoped_client): """Non-régression : une clé sans portée garde l'accès complet.""" admin = _login(scoped_client) key = _create(scoped_client, admin, name="legacy")["token"] assert scoped_client.get("/api/config", headers=_hdr(key)).status_code == 200 assert scoped_client.get("/api/webhooks", headers=_hdr(key)).status_code == 200 assert scoped_client.get("/api/auth/tokens", headers=_hdr(key)).status_code == 200 class TestScopedToolAccess: def _scoped_ctx(self): user = {"username": "admin", "role": "admin", "vaults": ["*"], "_token_vaults": ["TestVault"], "_token_scope": "files"} return _ctx_with_user(user) def test_file_tools_allowed(self, scoped_client): result = call_tool("list_all_files", self._scoped_ctx(), {"vault": "TestVault"}) assert result.ok and result.data["files"] def test_other_vault_denied_at_tool_level(self, scoped_client): with pytest.raises(ToolPermissionError): call_tool("list_all_files", self._scoped_ctx(), {"vault": "Autre"}) def test_non_file_tools_denied(self, scoped_client): for tool, args in (("web_search", {"query": "x"}), ("git_list_repos", {"provider": "gitea"}), ("notify_external", {"title": "t", "message": "m"})): with pytest.raises(ToolPermissionError) as exc: call_tool(tool, self._scoped_ctx(), args) assert exc.value.code == "token_scope_denied" def test_full_scope_unaffected(self, scoped_client): ctx = _ctx() # sans _token_scope → plein accès assert call_tool("list_all_files", ctx, {"vault": "TestVault"}).ok def _ctx_with_user(user): from backend.tools.api import ToolContext return ToolContext(user=user, audit_enabled=False)