import http.cookiejar import base64 import hashlib import json import os import re import socket import subprocess import tempfile import time import unittest import urllib.error import urllib.request from pathlib import Path ROOT = Path(__file__).resolve().parents[1] def free_port(): with socket.socket() as sock: sock.bind(("127.0.0.1", 0)) return sock.getsockname()[1] class ZhulanCellTest(unittest.TestCase): @classmethod def setUpClass(cls): cls.temp = tempfile.TemporaryDirectory() temp_root = Path(cls.temp.name) seed = temp_root / "seed" forge_root = temp_root / "forge" bare = forge_root / "bingshuo" / "guanghu-ice-heart.git" allowed = seed / "server-tools" / "zhulan-remote-cell" allowed.mkdir(parents=True) (allowed / "hello.txt").write_text("before\n", encoding="utf-8") subprocess.run(["git", "init", "-b", "main", str(seed)], check=True, capture_output=True) subprocess.run(["git", "-C", str(seed), "add", "."], check=True, capture_output=True) subprocess.run( [ "git", "-C", str(seed), "-c", "user.name=Zhulan Test", "-c", "user.email=zhulan-test@invalid", "commit", "-m", "test source", ], check=True, capture_output=True, ) cls.source_sha = subprocess.run( ["git", "-C", str(seed), "rev-parse", "HEAD"], check=True, capture_output=True, text=True, ).stdout.strip() bare.parent.mkdir(parents=True) subprocess.run(["git", "clone", "--bare", str(seed), str(bare)], check=True, capture_output=True) policy = json.loads((ROOT / "policy.example.json").read_text(encoding="utf-8")) policy["repository_base_url"] = forge_root.as_uri() policy["repositories"]["bingshuo/guanghu-ice-heart"]["validation_commands"] = [ ["git", "diff", "--check"] ] cls.policy_file = temp_root / "policy.json" cls.policy_file.write_text(json.dumps(policy), encoding="utf-8") cls.port = free_port() env = os.environ.copy() env.update( { "ZHULAN_BIND": "127.0.0.1", "ZHULAN_PORT": str(cls.port), "ZHULAN_DB": str(Path(cls.temp.name) / "state.sqlite3"), "ZHULAN_SECRET_FILE": str(Path(cls.temp.name) / "missing-test-secret"), "ZHULAN_POLICY": str(cls.policy_file), "ZHULAN_UI_DIR": str(ROOT / "ui"), "ZHULAN_WORKSPACE_ROOT": str(Path(cls.temp.name) / "workspaces"), "ZHULAN_CANDIDATE_ROOT": str(Path(cls.temp.name) / "candidates"), "ZHULAN_COOKIE_SECURE": "0", "ZHULAN_COOKIE_PATH": "/", "ZHULAN_TEST_MODE": "1", "ZHULAN_TEST_OWNER_PASSWORD": "correct-horse-test-only", } ) cls.process = subprocess.Popen( ["python3", str(ROOT / "runtime" / "zhulan_cell.py")], env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) cls.base = f"http://127.0.0.1:{cls.port}" deadline = time.time() + 8 while time.time() < deadline: try: with urllib.request.urlopen(cls.base + "/health", timeout=0.5) as response: if response.status == 200: break except Exception: time.sleep(0.08) else: out, err = cls.process.communicate(timeout=2) raise RuntimeError(f"test server did not start\n{out}\n{err}") @classmethod def tearDownClass(cls): cls.process.terminate() cls.process.wait(timeout=5) if cls.process.stdout: cls.process.stdout.close() if cls.process.stderr: cls.process.stderr.close() cls.temp.cleanup() def setUp(self): self.jar = http.cookiejar.CookieJar() self.opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(self.jar)) def request(self, path, body=None, headers=None, expected=200, form=False): if form and body is not None: from urllib.parse import urlencode raw = urlencode(body).encode() else: raw = json.dumps(body).encode() if body is not None else None request = urllib.request.Request( self.base + path, data=raw, headers={"Content-Type": "application/x-www-form-urlencoded" if form else "application/json", **(headers or {})}, ) try: with self.opener.open(request, timeout=3) as response: self.assertEqual(response.status, expected) return json.loads(response.read()) except urllib.error.HTTPError as exc: with exc: self.assertEqual(exc.code, expected) return json.loads(exc.read()) def create( self, dev="DEV-20260813-005", slug="approval-ui", base_sha="5f9e83e1b716fd50ee2880b24ea47f83b75d1f8f", ): return self.request( "/api/v1/public/requests", { "persona_id": "ICE-GL-ZL-001", "development_id": dev, "repository": "bingshuo/guanghu-ice-heart", "base_sha": base_sha, "branch": f"zhulan/{dev}/{slug}", "paths": ["server-tools/zhulan-remote-cell"], "actions": ["read", "edit", "test", "commit", "push_candidate"], "description": "完善铸澜手机审批入口与代码门禁", }, expected=201, ) def login(self): result = self.request( "/api/v1/owner/login", {"username": "bingshuo", "password": "correct-horse-test-only"}, ) return result["csrf"] def approved_oauth_token(self, created): csrf = self.login() self.request( f"/api/v1/owner/requests/{created['request_id']}/approve", {"ttl_seconds": 10800}, headers={"X-CSRF-Token": csrf}, ) registered = self.request( "/oauth/register", { "client_name": "ChatGPT test connector", "redirect_uris": ["https://chatgpt.com/connector/oauth/test-callback"], "token_endpoint_auth_method": "none", }, expected=201, ) verifier = "v" * 64 challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).decode().rstrip("=") oauth = { "client_id": registered["client_id"], "redirect_uri": registered["redirect_uris"][0], "state": "state-test-123", "code_challenge": challenge, "resource": "https://guanghulab.com/zhulan/mcp", "scope": "zhulan.develop", "csrf": self.request("/api/v1/owner/session")["csrf"], "request_id": created["request_id"], } class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None request = urllib.request.Request( self.base + "/oauth/authorize", data=__import__("urllib.parse").parse.urlencode(oauth).encode(), headers={"Content-Type": "application/x-www-form-urlencoded"}, ) no_redirect = urllib.request.build_opener( urllib.request.HTTPCookieProcessor(self.jar), NoRedirect() ) try: no_redirect.open(request, timeout=3) self.fail("authorize should redirect") except urllib.error.HTTPError as exc: with exc: self.assertEqual(exc.code, 302) location = exc.headers["Location"] from urllib.parse import parse_qs, urlparse code = parse_qs(urlparse(location).query)["code"][0] token = self.request( "/oauth/token", { "grant_type": "authorization_code", "code": code, "client_id": registered["client_id"], "redirect_uri": registered["redirect_uris"][0], "code_verifier": verifier, "resource": "https://guanghulab.com/zhulan/mcp", }, form=True, ) return token["access_token"] def mcp_tool(self, token, name, arguments=None): response = self.request( "/mcp", { "jsonrpc": "2.0", "id": 91, "method": "tools/call", "params": {"name": name, "arguments": arguments or {}}, }, headers={"Authorization": f"Bearer {token}"}, ) return response["result"] def test_public_config_declares_front_and_runtime(self): config = self.request("/api/v1/public/config") self.assertEqual(config["runtime_node"], "BS-SG-003") self.assertIn("BS-GZ-006", config["topology"]["front"]) self.assertIn("GHS-012", [x["id"] for x in config["ui_brains"]]) def test_wrong_persona_is_denied(self): result = self.request( "/api/v1/public/requests", { "persona_id": "ICE-GL-ZY001", "development_id": "DEV-20260813-005", "repository": "bingshuo/guanghu-ice-heart", "base_sha": "5f9e83e1b716fd50ee2880b24ea47f83b75d1f8f", "branch": "zhulan/DEV-20260813-005/nope", "paths": ["server-tools"], "actions": ["read"], "description": "错误人格不得进入铸澜车道", }, expected=400, ) self.assertEqual(result["error"], "persona_not_allowed") def test_approval_pickup_and_capability_verification(self): created = self.create(slug=f"flow-{int(time.time())}") self.assertRegex(created["request_id"], r"^ZLR-[0-9]{8}-[A-F0-9]{10}$") status = self.request( f"/api/v1/public/requests/{created['request_id']}/status", {"claim_token": created["claim_token"]}, ) self.assertEqual(status["state"], "PENDING") csrf = self.login() listing = self.request("/api/v1/owner/requests") self.assertTrue(any(item["id"] == created["request_id"] for item in listing["requests"])) approved = self.request( f"/api/v1/owner/requests/{created['request_id']}/approve", {"ttl_seconds": 10800}, headers={"X-CSRF-Token": csrf}, ) self.assertEqual(approved["state"], "APPROVED") pickup = self.request( f"/api/v1/public/requests/{created['request_id']}/pickup", {"claim_token": created["claim_token"]}, ) verified = self.request( "/api/v1/runtime/verify", {"capability": pickup["capability"], "action": "push_candidate"}, ) self.assertEqual(verified["claims"]["branch"], status["branch"]) self.assertEqual(verified["claims"]["node_id"], "BS-SG-003") second = self.request( f"/api/v1/public/requests/{created['request_id']}/pickup", {"claim_token": created["claim_token"]}, expected=409, ) self.assertEqual(second["error"], "capability_already_picked_up") filtered = self.request(f"/api/v1/owner/requests/{created['request_id']}/receipts") self.assertGreaterEqual(len(filtered["receipts"]), 4) receipts = self.request("/api/v1/owner/receipts") previous = "0" * 64 for receipt in receipts["receipts"]: self.assertEqual(receipt["previous_hash"], previous) previous = receipt["receipt_hash"] def test_csrf_is_required_for_decision(self): created = self.create(slug=f"csrf-{int(time.time())}") self.login() result = self.request( f"/api/v1/owner/requests/{created['request_id']}/approve", {"ttl_seconds": 10800}, expected=403, ) self.assertEqual(result["error"], "csrf_invalid") def test_owner_can_revoke_active_oauth_and_capability(self): created = self.create(slug=f"revoke-{int(time.time())}") token = self.approved_oauth_token(created) restored = self.mcp_tool(token, "zhulan_restore_lane") self.assertFalse(restored["isError"]) session = self.request("/api/v1/owner/session") revoked = self.request( f"/api/v1/owner/requests/{created['request_id']}/revoke", {"reason": "test owner stop"}, headers={"X-CSRF-Token": session["csrf"]}, ) self.assertEqual(revoked["state"], "REVOKED") denied = self.mcp_tool(token, "zhulan_restore_lane") self.assertTrue(denied["isError"]) self.assertIn("mcp/www_authenticate", denied["_meta"]) def test_mcp_requires_oauth_for_development_tools(self): response = self.request( "/mcp", { "jsonrpc": "2.0", "id": 4, "method": "tools/call", "params": {"name": "zhulan_restore_lane", "arguments": {}}, }, ) result = response["result"] self.assertTrue(result["isError"]) self.assertIn("mcp/www_authenticate", result["_meta"]) tools = self.request( "/mcp", {"jsonrpc": "2.0", "id": 5, "method": "tools/list", "params": {}} )["result"]["tools"] protected = next(tool for tool in tools if tool["name"] == "zhulan_restore_lane") public = next(tool for tool in tools if tool["name"] == "zhulan_request_development") self.assertEqual(protected["securitySchemes"][0]["type"], "oauth2") self.assertEqual(public["securitySchemes"][0]["type"], "noauth") self.assertNotIn("capability", protected["inputSchema"]["properties"]) def test_oauth_pkce_binds_to_approved_request_and_code_is_single_use(self): created = self.create(slug=f"oauth-{int(time.time())}") csrf = self.login() self.request( f"/api/v1/owner/requests/{created['request_id']}/approve", {"ttl_seconds": 10800}, headers={"X-CSRF-Token": csrf}, ) registered = self.request( "/oauth/register", { "client_name": "ChatGPT test connector", "redirect_uris": ["https://chatgpt.com/connector/oauth/test-callback"], "token_endpoint_auth_method": "none", }, expected=201, ) verifier = "v" * 64 challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).decode().rstrip("=") oauth = { "client_id": registered["client_id"], "redirect_uri": registered["redirect_uris"][0], "state": "state-test-123", "code_challenge": challenge, "resource": "https://guanghulab.com/zhulan/mcp", "scope": "zhulan.develop", } session = self.request("/api/v1/owner/session") oauth["csrf"] = session["csrf"] oauth["request_id"] = created["request_id"] request = urllib.request.Request( self.base + "/oauth/authorize", data=__import__("urllib.parse").parse.urlencode(oauth).encode(), headers={"Content-Type": "application/x-www-form-urlencoded"}, ) no_redirect = urllib.request.build_opener( urllib.request.HTTPCookieProcessor(self.jar), urllib.request.HTTPHandler(), ) class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None no_redirect = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(self.jar), NoRedirect()) try: no_redirect.open(request, timeout=3) self.fail("authorize should redirect") except urllib.error.HTTPError as exc: with exc: self.assertEqual(exc.code, 302) location = exc.headers["Location"] from urllib.parse import parse_qs, urlparse code = parse_qs(urlparse(location).query)["code"][0] token_request = { "grant_type": "authorization_code", "code": code, "client_id": registered["client_id"], "redirect_uri": registered["redirect_uris"][0], "code_verifier": verifier, "resource": "https://guanghulab.com/zhulan/mcp", } token = self.request("/oauth/token", token_request, form=True) self.assertEqual(token["token_type"], "Bearer") replay = self.request("/oauth/token", token_request, expected=403, form=True) self.assertEqual(replay["error"], "invalid_grant") restored = self.request( "/mcp", { "jsonrpc": "2.0", "id": 7, "method": "tools/call", "params": {"name": "zhulan_restore_lane", "arguments": {}}, }, headers={"Authorization": f"Bearer {token['access_token']}"}, ) content = restored["result"]["structuredContent"] self.assertEqual(content["development_id"], "DEV-20260813-005") self.assertEqual(content["topology"]["front"], "BS-GZ-006_PROXY_ONLY") def test_oauth_registration_rejects_untrusted_callback(self): result = self.request( "/oauth/register", { "client_name": "evil", "redirect_uris": ["https://example.com/steal"], "token_endpoint_auth_method": "none", }, expected=400, ) self.assertEqual(result["error"], "redirect_uri_not_allowed") def test_existing_workspace_binding_mismatch_fails_before_git_changes(self): created = self.create( dev="DEV-20260813-098", slug="binding", base_sha=self.source_sha ) token = self.approved_oauth_token(created) prepared = self.mcp_tool(token, "zhulan_prepare_workspace")["structuredContent"] workspace = ( Path(self.temp.name) / "workspaces" / "DEV-20260813-098" / "bingshuo__guanghu-ice-heart" ) before = subprocess.run( ["git", "-C", str(workspace), "rev-parse", "HEAD"], check=True, capture_output=True, text=True, ).stdout.strip() binding = workspace / ".git" / "zhulan-binding.json" altered = json.loads(binding.read_text(encoding="utf-8")) altered["branch"] = "zhulan/DEV-20260813-098/tampered" binding.write_text(json.dumps(altered), encoding="utf-8") denied = self.mcp_tool(token, "zhulan_prepare_workspace") self.assertTrue(denied["isError"]) self.assertEqual(denied["structuredContent"]["error"], "workspace_binding_mismatch") after = subprocess.run( ["git", "-C", str(workspace), "rev-parse", "HEAD"], check=True, capture_output=True, text=True, ).stdout.strip() self.assertEqual(before, prepared["head"]) self.assertEqual(after, before) def test_exact_reapproval_resumes_same_workspace(self): dev = "DEV-20260813-097" first = self.create(dev=dev, slug="renew", base_sha=self.source_sha) first_token = self.approved_oauth_token(first) first_prepared = self.mcp_tool(first_token, "zhulan_prepare_workspace")[ "structuredContent" ] second = self.create(dev=dev, slug="renew", base_sha=self.source_sha) second_token = self.approved_oauth_token(second) resumed = self.mcp_tool(second_token, "zhulan_prepare_workspace")[ "structuredContent" ] self.assertEqual(resumed["head"], first_prepared["head"]) binding = ( Path(self.temp.name) / "workspaces" / dev / "bingshuo__guanghu-ice-heart" / ".git" / "zhulan-binding.json" ) rebound = json.loads(binding.read_text(encoding="utf-8")) self.assertEqual(rebound["request_id"], second["request_id"]) self.assertEqual(rebound["renewed_from_request_id"], first["request_id"]) def test_full_workspace_candidate_flow_is_scoped_and_repeatable(self): dev = "DEV-20260813-099" created = self.request( "/api/v1/public/requests", { "persona_id": "ICE-GL-ZL-001", "development_id": dev, "repository": "bingshuo/guanghu-ice-heart", "base_sha": self.source_sha, "branch": f"zhulan/{dev}/full-flow", "paths": ["server-tools/zhulan-remote-cell"], "actions": ["read", "edit", "test", "commit", "push_candidate"], "description": "端到端验证受限工作区和内部候选复核库", }, expected=201, ) token = self.approved_oauth_token(created) prepared = self.mcp_tool(token, "zhulan_prepare_workspace")["structuredContent"] self.assertEqual(prepared["head"], self.source_sha) self.assertEqual(prepared["candidate_store"], "BS-SG-003_INTERNAL_REVIEW_ONLY") read = self.mcp_tool( token, "zhulan_read_file", {"path": "server-tools/zhulan-remote-cell/hello.txt"}, )["structuredContent"] self.assertEqual(read["content"], "before\n") written = self.mcp_tool( token, "zhulan_write_file", { "path": "server-tools/zhulan-remote-cell/hello.txt", "content": "after\n", "expected_sha256": read["sha256"], }, )["structuredContent"] self.assertEqual(written["sha256"], hashlib.sha256(b"after\n").hexdigest()) escaped = self.mcp_tool( token, "zhulan_write_file", {"path": "../escape", "content": "no", "expected_sha256": ""}, ) self.assertTrue(escaped["isError"]) self.assertEqual(escaped["structuredContent"]["error"], "path_traversal") initial_validation = self.mcp_tool(token, "zhulan_run_validation")["structuredContent"] self.assertEqual(initial_validation["decision"], "PASS") self.assertFalse(initial_validation["clean"]) committed = self.mcp_tool( token, "zhulan_commit_candidate", {"message": "test: update bounded file"} )["structuredContent"] premature = self.mcp_tool(token, "zhulan_push_candidate") self.assertTrue(premature["isError"]) final_validation = self.mcp_tool(token, "zhulan_run_validation")["structuredContent"] self.assertTrue(final_validation["clean"]) self.assertEqual(final_validation["head"], committed["commit_sha"]) pushed = self.mcp_tool(token, "zhulan_push_candidate")["structuredContent"] self.assertEqual(pushed["commit_sha"], committed["commit_sha"]) self.assertEqual(pushed["central_publication"], "NOT_PERFORMED_REQUIRES_ZHUYUAN_REVIEW") prepared_again = self.mcp_tool(token, "zhulan_prepare_workspace")["structuredContent"] self.assertEqual(prepared_again["head"], committed["commit_sha"]) candidate = Path(self.temp.name) / "candidates" / "bingshuo__guanghu-ice-heart.git" remote_sha = subprocess.run( ["git", "--git-dir", str(candidate), "rev-parse", f"refs/heads/zhulan/{dev}/full-flow"], check=True, capture_output=True, text=True, ).stdout.strip() self.assertEqual(remote_sha, committed["commit_sha"]) if __name__ == "__main__": unittest.main()