171 lines
8.3 KiB
Python
171 lines
8.3 KiB
Python
"""Client administration: service layer, audit log, admin API and CLI."""
|
|
|
|
import contextlib
|
|
import io
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from mcp_calculator import admin_cli
|
|
from mcp_calculator.admin import AuditLog, ClientAdmin, Conflict, NotFound, build_admin_app
|
|
from mcp_calculator.clients import ClientConfigError, ClientFile, hash_token
|
|
from tests.support import asgi_request
|
|
|
|
ADMIN_TOKEN = "admin-token-0123456789"
|
|
|
|
|
|
class AdminTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.dir = Path(self.tmp.name)
|
|
self.file = ClientFile(self.dir / "clients.json")
|
|
self.audit_path = self.dir / "audit.log"
|
|
self.changes = 0
|
|
self.admin = ClientAdmin(self.file, AuditLog(self.audit_path), on_change=self._on_change)
|
|
|
|
def _on_change(self):
|
|
self.changes += 1
|
|
|
|
def audit_entries(self) -> list[dict]:
|
|
return [json.loads(line) for line in self.audit_path.read_text().splitlines()]
|
|
|
|
|
|
class ClientAdminTest(AdminTestCase):
|
|
def test_create_returns_token_once_and_stores_hash(self):
|
|
client, token = self.admin.create("team-a", actor="test", max_concurrent=4, note="Team A")
|
|
self.assertTrue(token.startswith("mcpc_"))
|
|
self.assertEqual(client["token_hint"], token[:9])
|
|
stored = self.file.read()["team-a"]
|
|
self.assertEqual(stored.token_sha256, hash_token(token))
|
|
self.assertNotIn(token, (self.dir / "clients.json").read_text())
|
|
self.assertEqual(self.changes, 1)
|
|
|
|
def test_network_client_has_no_token(self):
|
|
client, token = self.admin.create("batch", actor="test", auth="network", allowed_networks=["10.0.0.1"])
|
|
self.assertIsNone(token)
|
|
self.assertEqual(client["auth"], "network")
|
|
|
|
def test_lifecycle(self):
|
|
_, first = self.admin.create("team-a", actor="test")
|
|
self.assertEqual(self.admin.update("team-a", {"enabled": False}, actor="test")["status"], "disabled")
|
|
_, second = self.admin.rotate("team-a", actor="test")
|
|
self.assertNotEqual(first, second)
|
|
self.assertEqual(self.file.read()["team-a"].token_sha256, hash_token(second))
|
|
self.assertFalse(self.file.read()["team-a"].enabled, "rotate keeps the other settings")
|
|
self.admin.delete("team-a", actor="test")
|
|
self.assertEqual(self.admin.list(), [])
|
|
|
|
def test_errors(self):
|
|
self.admin.create("team-a", actor="test")
|
|
with self.assertRaises(Conflict):
|
|
self.admin.create("team-a", actor="test")
|
|
with self.assertRaises(NotFound):
|
|
self.admin.update("nobody", {"enabled": False}, actor="test")
|
|
with self.assertRaises(ClientConfigError):
|
|
self.admin.create("Bad Name", actor="test")
|
|
with self.assertRaises(ClientConfigError):
|
|
self.admin.update("team-a", {"token_sha256": "0" * 64}, actor="test")
|
|
|
|
def test_audit_log_has_no_secrets(self):
|
|
_, token = self.admin.create("team-a", actor="admin:ops", max_concurrent=2)
|
|
self.admin.update("team-a", {"enabled": False}, actor="admin:ops")
|
|
self.admin.rotate("team-a", actor="admin:ops")
|
|
entries = self.audit_entries()
|
|
self.assertEqual([e["action"] for e in entries], ["create", "update", "rotate"])
|
|
self.assertTrue(all(e["actor"] == "admin:ops" and e["client"] == "team-a" for e in entries))
|
|
text = self.audit_path.read_text()
|
|
self.assertNotIn(token, text)
|
|
self.assertNotIn(hash_token(token), text)
|
|
|
|
|
|
class AdminApiTest(AdminTestCase, unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.app = build_admin_app(self.admin, [f"ops:{ADMIN_TOKEN}"])
|
|
self.auth = {"Authorization": f"Bearer {ADMIN_TOKEN}"}
|
|
|
|
async def test_requires_admin_token(self):
|
|
self.assertEqual((await asgi_request(self.app, "GET", "/admin/clients"))[0], 401)
|
|
self.assertEqual((await asgi_request(self.app, "GET", "/admin/clients", headers={"Authorization": "Bearer wrong"}))[0], 401)
|
|
self.assertEqual((await asgi_request(self.app, "GET", "/healthz"))[0], 200)
|
|
|
|
async def test_crud(self):
|
|
status, body = await asgi_request(self.app, "POST", "/admin/clients", headers=self.auth, body={"name": "team-a", "max_concurrent": 3})
|
|
self.assertEqual(status, 201)
|
|
self.assertTrue(body["token"].startswith("mcpc_"))
|
|
|
|
status, body = await asgi_request(self.app, "GET", "/admin/clients", headers=self.auth)
|
|
self.assertEqual([c["name"] for c in body["clients"]], ["team-a"])
|
|
self.assertNotIn("token_sha256", body["clients"][0])
|
|
|
|
status, body = await asgi_request(self.app, "PATCH", "/admin/clients/team-a", headers=self.auth, body={"enabled": False})
|
|
self.assertEqual((status, body["status"]), (200, "disabled"))
|
|
|
|
status, body = await asgi_request(self.app, "POST", "/admin/clients/team-a/rotate", headers=self.auth)
|
|
self.assertEqual(status, 200)
|
|
self.assertIn("token", body)
|
|
|
|
self.assertEqual((await asgi_request(self.app, "DELETE", "/admin/clients/team-a", headers=self.auth))[0], 200)
|
|
self.assertEqual((await asgi_request(self.app, "GET", "/admin/clients/team-a", headers=self.auth))[0], 404)
|
|
self.assertEqual(self.audit_entries()[0]["actor"], "admin:ops")
|
|
|
|
async def test_validation_errors(self):
|
|
cases = [
|
|
("POST", "/admin/clients", {"max_concurrent": 1}, 400), # no name
|
|
("POST", "/admin/clients", {"name": "a", "role": "root"}, 400), # unknown field
|
|
("POST", "/admin/clients", {"name": "a", "expires": "someday"}, 400),
|
|
("PATCH", "/admin/clients/missing", {"enabled": False}, 404),
|
|
]
|
|
for method, path, body, expected in cases:
|
|
with self.subTest(body=body):
|
|
self.assertEqual((await asgi_request(self.app, method, path, headers=self.auth, body=body))[0], expected)
|
|
await asgi_request(self.app, "POST", "/admin/clients", headers=self.auth, body={"name": "dup"})
|
|
self.assertEqual((await asgi_request(self.app, "POST", "/admin/clients", headers=self.auth, body={"name": "dup"}))[0], 409)
|
|
|
|
|
|
class AdminCliTest(AdminTestCase):
|
|
def cli(self, *args: str) -> tuple[int, str]:
|
|
out = io.StringIO()
|
|
with contextlib.redirect_stdout(out), contextlib.redirect_stderr(out):
|
|
code = admin_cli.main(["--file", str(self.file.path), *args])
|
|
return code, out.getvalue()
|
|
|
|
def test_create_list_and_manage(self):
|
|
code, out = self.cli("create", "team-a", "--max-concurrent", "4", "--expires", "2030-12-31")
|
|
self.assertEqual(code, 0)
|
|
self.assertIn("TOKEN (shown only once", out)
|
|
code, out = self.cli("list")
|
|
self.assertIn("team-a", out)
|
|
self.assertIn("active", out)
|
|
|
|
self.assertEqual(self.cli("disable", "team-a")[0], 0)
|
|
self.assertFalse(self.file.read()["team-a"].enabled)
|
|
self.assertEqual(self.cli("enable", "team-a")[0], 0)
|
|
self.assertTrue(self.file.read()["team-a"].enabled)
|
|
|
|
self.assertEqual(self.cli("set", "team-a", "--max-concurrent", "default", "--networks", "10.0.0.0/8", "--expires", "never")[0], 0)
|
|
client = self.file.read()["team-a"]
|
|
self.assertIsNone(client.max_concurrent)
|
|
self.assertIsNone(client.expires)
|
|
self.assertEqual([str(n) for n in client.allowed_networks], ["10.0.0.0/8"])
|
|
|
|
code, out = self.cli("--json", "rotate", "team-a")
|
|
self.assertTrue(json.loads(out)["token"].startswith("mcpc_"))
|
|
self.assertEqual(self.cli("delete", "team-a", "--yes")[0], 0)
|
|
self.assertEqual(self.file.read(), {})
|
|
self.assertTrue(all(e["actor"].startswith("cli:") for e in self.audit_entries()))
|
|
|
|
def test_network_only_client(self):
|
|
self.assertEqual(self.cli("create", "batch", "--network-only", "--networks", "10.131.2.15")[0], 0)
|
|
self.assertEqual(self.file.read()["batch"].auth, "network")
|
|
self.assertEqual(self.cli("create", "nonet", "--network-only")[0], 1)
|
|
|
|
def test_errors_exit_non_zero(self):
|
|
self.assertEqual(self.cli("disable", "missing")[0], 1)
|
|
self.assertEqual(self.cli("set", "missing", "--note", "x")[0], 1)
|
|
self.cli("create", "team-a")
|
|
code, out = self.cli("set", "team-a")
|
|
self.assertEqual(code, 1)
|
|
self.assertIn("nothing to change", out)
|