379 lines
15 KiB
Python
379 lines
15 KiB
Python
"""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()
|