"""Fail-closed ContextForge production token bootstrap tests.""" from __future__ import annotations import json import os import tempfile import threading import unittest from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from unittest.mock import patch from scripts.contextforge_bootstrap import ( RUNTIME_PERMISSIONS, BootstrapError, bootstrap, ) CATALOG_NAME = "runtime-prod" REPLACEMENT_NAME = "runtime-prod-0a93ae1" class ContextForgeHandler(BaseHTTPRequestHandler): calls: list[tuple[str, str]] = [] created_scope: dict[str, object] = {} old_token_revoked = False new_token_revoked = False denied_runtime_paths: set[str] = set() created_name = "" def log_message(self, *_: object) -> None: return def _json(self) -> dict[str, object]: length = int(self.headers.get("Content-Length", "0")) return json.loads(self.rfile.read(length)) if length else {} def _send(self, status: int, body: object) -> None: payload = json.dumps(body).encode() self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload) def do_POST(self) -> None: # noqa: N802 self.calls.append(("POST", self.path)) body = self._json() if self.path == "/auth/email/login" and body.get("password") == "admin-password": self._send(200, {"access_token": "session-token"}) return if self.path == "/tokens" and self.headers.get("Authorization") == "Bearer session-token": self.created_scope = body.get("scope", {}) type(self).created_scope = self.created_scope type(self).created_name = str(body.get("name", "")) self._send( 201, { "access_token": "runtime-api-token", "token": {"id": "new-token-id"}, }, ) return self._send(401, {"detail": "denied"}) def do_GET(self) -> None: # noqa: N802 self.calls.append(("GET", self.path)) if ( self.path.split("?", 1)[0] in {"/gateways", "/tools", "/prompts", "/servers"} and self.headers.get("Authorization") == "Bearer runtime-api-token" ): if self.path.split("?", 1)[0] in self.denied_runtime_paths: self._send(403, {"detail": "insufficient scope"}) return self._send(200, []) return if ( self.path.startswith("/gateways") and self.headers.get("Authorization") == "Bearer runtime-api-token-old" ): self._send(200, []) return if ( self.path.startswith("/tokens?") and self.headers.get("Authorization") == "Bearer session-token" ): self._send( 200, { "tokens": [ { "id": "old-token-id", "name": CATALOG_NAME, "is_active": True, "is_revoked": self.old_token_revoked, } ], "total": 1, }, ) return if self.headers.get("Authorization") == "Bearer runtime-api-token-old": self._send(403, {"detail": "insufficient scope"}) return self._send(401, {"detail": "denied"}) def do_DELETE(self) -> None: # noqa: N802 self.calls.append(("DELETE", self.path)) if ( self.path in {"/tokens/old-token-id", "/tokens/new-token-id"} and self.headers.get("Authorization") == "Bearer session-token" ): if self.path.endswith("old-token-id"): type(self).old_token_revoked = True else: type(self).new_token_revoked = True self.send_response(204) self.end_headers() return self._send(401, {"detail": "denied"}) class ContextForgeBootstrapTests(unittest.TestCase): def setUp(self) -> None: ContextForgeHandler.calls = [] ContextForgeHandler.created_scope = {} ContextForgeHandler.old_token_revoked = False ContextForgeHandler.new_token_revoked = False ContextForgeHandler.denied_runtime_paths = set() ContextForgeHandler.created_name = "" self.server = ThreadingHTTPServer(("127.0.0.1", 0), ContextForgeHandler) self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) self.thread.start() self.base_url = f"http://127.0.0.1:{self.server.server_port}" def tearDown(self) -> None: self.server.shutdown() self.server.server_close() self.thread.join(timeout=2) def test_creates_scoped_token_without_returning_value(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) password = root / "password" token = root / "contextforge_api_token" password.write_text("admin-password", encoding="utf-8") os.chmod(root, 0o700) os.chmod(password, 0o600) with patch( "scripts.contextforge_bootstrap._validate_base_url", return_value=self.base_url ): result = bootstrap( base_url=self.base_url, admin_email="admin@example.com", admin_password_file=password, output_token_file=token, source_network="203.0.113.0/24", token_name=CATALOG_NAME, expires_days=365, ) self.assertEqual(result["action"], "created") self.assertNotIn("runtime-api-token", repr(result)) self.assertEqual(result["permissions"], list(RUNTIME_PERMISSIONS)) self.assertEqual( set(result["permissions"]), { "gateways.read", "gateways.create", "gateways.update", "gateways.delete", "tools.read", "tools.update", "tools.execute", "prompts.read", "servers.read", "servers.create", "servers.update", "servers.use", }, ) self.assertNotIn("*", result["permissions"]) self.assertEqual( ContextForgeHandler.created_scope["permissions"], list(RUNTIME_PERMISSIONS) ) self.assertEqual(token.read_text(encoding="utf-8"), "runtime-api-token") if os.name != "nt": self.assertEqual(token.stat().st_mode & 0o777, 0o600) def test_reuses_only_a_working_existing_token(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) password = root / "password" token = root / "contextforge_api_token" password.write_text("unused", encoding="utf-8") token.write_text("runtime-api-token", encoding="utf-8") os.chmod(root, 0o700) os.chmod(password, 0o600) os.chmod(token, 0o600) with patch( "scripts.contextforge_bootstrap._validate_base_url", return_value=self.base_url ): result = bootstrap( base_url=self.base_url, admin_email="admin@example.com", admin_password_file=password, output_token_file=token, source_network="203.0.113.0/24", token_name=CATALOG_NAME, expires_days=365, ) self.assertEqual( result, {"status": "pass", "action": "reused", "tokenName": "runtime-prod"} ) self.assertFalse(any(path == "/tokens" for _, path in ContextForgeHandler.calls)) def test_invalid_existing_token_fails_without_overwrite(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) password = root / "password" token = root / "contextforge_api_token" password.write_text("unused", encoding="utf-8") token.write_text("invalid-token", encoding="utf-8") os.chmod(root, 0o700) os.chmod(password, 0o600) os.chmod(token, 0o600) with ( patch( "scripts.contextforge_bootstrap._validate_base_url", return_value=self.base_url, ), self.assertRaisesRegex(BootstrapError, "HTTP 401"), ): bootstrap( base_url=self.base_url, admin_email="admin@example.com", admin_password_file=password, output_token_file=token, source_network="203.0.113.0/24", token_name=CATALOG_NAME, expires_days=365, ) self.assertEqual(token.read_text(encoding="utf-8"), "invalid-token") def test_rotates_one_named_under_scoped_token_without_exposing_values(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) password = root / "password" token = root / "contextforge_api_token" password.write_text("admin-password", encoding="utf-8") token.write_text("runtime-api-token-old", encoding="utf-8") os.chmod(root, 0o700) os.chmod(password, 0o600) os.chmod(token, 0o600) with patch( "scripts.contextforge_bootstrap._validate_base_url", return_value=self.base_url ): result = bootstrap( base_url=self.base_url, admin_email="admin@example.com", admin_password_file=password, output_token_file=token, source_network="203.0.113.0/24", token_name=CATALOG_NAME, expires_days=365, rotate_existing=True, replacement_token_name=REPLACEMENT_NAME, ) self.assertEqual(result["action"], "created") self.assertEqual(result["tokenName"], REPLACEMENT_NAME) self.assertEqual(ContextForgeHandler.created_name, REPLACEMENT_NAME) self.assertTrue(ContextForgeHandler.old_token_revoked) self.assertEqual(token.read_text(encoding="utf-8"), "runtime-api-token") self.assertNotIn("runtime-api-token", repr(result)) self.assertIn(("DELETE", "/tokens/old-token-id"), ContextForgeHandler.calls) def test_explicit_rotation_replaces_an_existing_token_that_still_passes_read_checks( self, ) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) password = root / "password" token = root / "contextforge_api_token" password.write_text("admin-password", encoding="utf-8") token.write_text("runtime-api-token", encoding="utf-8") os.chmod(root, 0o700) os.chmod(password, 0o600) os.chmod(token, 0o600) with patch( "scripts.contextforge_bootstrap._validate_base_url", return_value=self.base_url ): result = bootstrap( base_url=self.base_url, admin_email="admin@example.com", admin_password_file=password, output_token_file=token, source_network="203.0.113.0/24", token_name=CATALOG_NAME, expires_days=365, rotate_existing=True, replacement_token_name=REPLACEMENT_NAME, ) self.assertEqual(result["action"], "created") self.assertTrue(ContextForgeHandler.old_token_revoked) self.assertEqual(ContextForgeHandler.created_name, REPLACEMENT_NAME) self.assertIn(("DELETE", "/tokens/old-token-id"), ContextForgeHandler.calls) def test_rejects_rotation_without_a_distinct_replacement_name(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) password = root / "password" token = root / "contextforge_api_token" password.write_text("admin-password", encoding="utf-8") token.write_text("runtime-api-token-old", encoding="utf-8") os.chmod(root, 0o700) os.chmod(password, 0o600) os.chmod(token, 0o600) for replacement in (None, CATALOG_NAME): with self.assertRaisesRegex(BootstrapError, "distinct replacement"): bootstrap( base_url="https://contextforge.example.test", admin_email="admin@example.com", admin_password_file=password, output_token_file=token, source_network="203.0.113.0/24", token_name=CATALOG_NAME, expires_days=365, rotate_existing=True, replacement_token_name=replacement, ) self.assertEqual(token.read_text(encoding="utf-8"), "runtime-api-token-old") self.assertEqual(ContextForgeHandler.calls, []) def test_revokes_new_token_when_scope_verification_fails(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) password = root / "password" token = root / "contextforge_api_token" password.write_text("admin-password", encoding="utf-8") os.chmod(root, 0o700) os.chmod(password, 0o600) ContextForgeHandler.denied_runtime_paths = {"/servers"} with ( patch( "scripts.contextforge_bootstrap._validate_base_url", return_value=self.base_url, ), self.assertRaisesRegex(BootstrapError, "servers returned HTTP 403"), ): bootstrap( base_url=self.base_url, admin_email="admin@example.com", admin_password_file=password, output_token_file=token, source_network="203.0.113.0/24", token_name=CATALOG_NAME, expires_days=365, ) self.assertTrue(ContextForgeHandler.new_token_revoked) self.assertFalse(token.exists()) self.assertIn(("DELETE", "/tokens/new-token-id"), ContextForgeHandler.calls) def test_rejects_unapproved_plain_http_endpoint(self) -> None: with self.assertRaisesRegex(BootstrapError, "not approved"): bootstrap( base_url=self.base_url, admin_email="admin@example.com", admin_password_file=Path("missing"), output_token_file=Path("missing-output"), source_network="203.0.113.0/24", token_name=CATALOG_NAME, expires_days=365, ) if __name__ == "__main__": unittest.main()