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