"""P1 控制面/控制台单测（server.py / console.html / sub2api.py / push.py）。
只覆盖本批 P1 控制面改动，不跑真实模型回合、不联网、不碰生产：
server 侧走本地临时端口 + 临时 SQLite；网关侧用假 request /
假 admin（key、token 全是测试假值）；推送 DNS 用注入的
resolver。绝不改 worker.py / coordinator.py / cost.py / host.py。
"""
import http.client
import json
import os
import sqlite3
import stat
import sys
import tempfile
import threading
import unittest
from pathlib import Path
from unittest import mock
try:
from cloud.auth import AuthStore
from cloud.push import dispatch_event, validate_push_target
from cloud.push import create_subscription
from cloud.server import create_server
from cloud import server as server_mod
from cloud.sub2api import Sub2apiError
from cloud.sub2api import ensure_gateway_keys_table
from cloud.sub2api import get_user_gateway_key
from cloud.sub2api import rotate_user_gateway_key
from cloud.sub2api import validate_or_rotate_gateway_key
except ImportError:
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cloud.auth import AuthStore
from cloud.push import dispatch_event, validate_push_target
from cloud.push import create_subscription
from cloud.server import create_server
from cloud import server as server_mod
from cloud.sub2api import Sub2apiError
from cloud.sub2api import ensure_gateway_keys_table
from cloud.sub2api import get_user_gateway_key
from cloud.sub2api import rotate_user_gateway_key
from cloud.sub2api import validate_or_rotate_gateway_key
def _req(port, method, path, body=None, key=None, raw_body=None):
conn = http.client.HTTPConnection("127.0.0.1", port, timeout=10)
hdrs = {}
if raw_body is not None:
payload = raw_body
elif body is not None:
payload = json.dumps(body).encode("utf-8")
hdrs["Content-Type"] = "application/json"
else:
payload = None
if key is not None:
hdrs["X-API-Key"] = key
try:
conn.request(method, path, body=payload, headers=hdrs)
resp = conn.getresponse()
data = resp.read()
finally:
conn.close()
try:
parsed = json.loads(data.decode("utf-8")) if data else None
except ValueError:
parsed = data.decode("utf-8", "replace")
return resp.status, parsed
class _FakeGatewayAdmin:
"""假网关管理员：建号/授余额只记调用，不联网。"""
def __init__(self):
self.created = []
self.deleted = []
self.balances = []
def create_user(self, email, password, username):
sub_id = 1000 + len(self.created)
self.created.append((email, username))
return sub_id
def delete_user(self, sub_user_id):
self.deleted.append(sub_user_id)
return True
def set_balance(self, sub_user_id, amount):
self.balances.append((sub_user_id, amount))
def resolve_group_id(self):
return 2
class _FakeGatewayRequest:
"""假网关 HTTP：models 状态可调，其余走固定成功路径。"""
def __init__(self, models_status=200):
self.models_status = models_status
self.calls = []
def __call__(self, method, url, body=None, token=None):
self.calls.append((method, url))
if url.endswith("/v1/models"):
return self.models_status, {"data": []}
if url.endswith("/api/v1/auth/login"):
return 200, {"data": {"access_token": "fake-user-token",
"refresh_token": "fake-refresh",
"expires_in": 3600}}
if url.endswith("/api/v1/keys"):
return 201, {"data": {"id": "fake-key-id-1",
"key": "fake-gateway-key-1"}}
if "/balance" in url:
return 200, {"data": {}}
return 200, {"data": {}}
class Sub2apiP1TestCase(unittest.TestCase):
def setUp(self):
self.conn = sqlite3.connect(":memory:")
ensure_gateway_keys_table(self.conn)
self.admin = _FakeGatewayAdmin()
# 假 origin：请求本身全是 fake，不产生真实网络调用。
self._old_origin = os.environ.get("SUB2API_BASE_URL")
os.environ["SUB2API_BASE_URL"] = "http://127.0.0.1:9"
def tearDown(self):
self.conn.close()
if self._old_origin is None:
os.environ.pop("SUB2API_BASE_URL", None)
else:
os.environ["SUB2API_BASE_URL"] = self._old_origin
def _seed_row(self, key="fake-old-key"):
self.conn.execute(
"INSERT INTO user_gateway_keys(user_id, sub2api_user_id,"
" api_key, refresh_token, created_at, updated_at)"
" VALUES(?, ?, ?, ?, ?, ?)",
("u1", 42, key, "fake-refresh", 1.0, 1.0))
self.conn.commit()
def test_validate_no_row_provisions(self):
req = _FakeGatewayRequest()
status, key, sub_id, key_id = validate_or_rotate_gateway_key(
self.conn, "u1", admin=self.admin, request=req)
self.assertEqual(status, "provisioned")
self.assertEqual(key, "fake-gateway-key-1")
self.assertEqual(key_id, "fake-key-id-1")
self.assertIsNotNone(get_user_gateway_key(self.conn, "u1"))
def test_validate_healthy_key_not_rotated(self):
self._seed_row()
req = _FakeGatewayRequest(models_status=200)
status, key, _sub, _kid = validate_or_rotate_gateway_key(
self.conn, "u1", admin=self.admin, request=req)
self.assertEqual(status, "ok")
self.assertEqual(key, "fake-old-key")
self.assertEqual(self.admin.created, [])
def test_validate_dead_key_rotates_not_pending(self):
self._seed_row()
req = _FakeGatewayRequest(models_status=403)
status, key, _sub, _kid = validate_or_rotate_gateway_key(
self.conn, "u1", admin=self.admin, request=req)
self.assertEqual(status, "rotated")
self.assertEqual(key, "fake-gateway-key-1")
row = get_user_gateway_key(self.conn, "u1")
self.assertEqual(row["api_key"], "fake-gateway-key-1")
self.assertNotEqual(row["api_key"], "fake-old-key")
def test_validate_zero_balance_403_rotates(self):
# 余额为 0 时网关鉴权层回 403：与 401 同属死 key，必须轮换。
self._seed_row()
req = _FakeGatewayRequest(models_status=403)
status, _key, _sub, _kid = validate_or_rotate_gateway_key(
self.conn, "u1", admin=self.admin, request=req)
self.assertEqual(status, "rotated")
def test_validate_gateway_down_keeps_old_row(self):
self._seed_row()
req = _FakeGatewayRequest(models_status=500)
with self.assertRaises(Sub2apiError):
validate_or_rotate_gateway_key(
self.conn, "u1", admin=self.admin, request=req)
row = get_user_gateway_key(self.conn, "u1")
self.assertEqual(row["api_key"], "fake-old-key")
def test_rotate_replaces_row(self):
self._seed_row()
req = _FakeGatewayRequest()
status, key, sub_id, key_id = rotate_user_gateway_key(
self.conn, "u1", admin=self.admin, request=req)
self.assertEqual(status, "rotated")
self.assertEqual(key_id, "fake-key-id-1")
self.assertEqual(
get_user_gateway_key(self.conn, "u1")["api_key"], key)
class PushP1TestCase(unittest.TestCase):
def setUp(self):
self.conn = sqlite3.connect(":memory:")
def tearDown(self):
self.conn.close()
def test_literal_private_and_loopback_blocked(self):
for target in (
"http://127.0.0.1:9/hook",
"http://10.0.0.5/hook",
"http://172.16.4.4/hook",
"http://192.168.1.10/hook",
"http://169.254.169.254/latest/meta-data/",
"http://[::1]/hook"):
self.assertIsNotNone(
validate_push_target(target), target)
def test_metadata_hostname_blocked(self):
self.assertIsNotNone(validate_push_target(
"http://metadata.google.internal/hook"))
def test_hostname_resolved_to_private_blocked(self):
err = validate_push_target(
"https://hooks.example.com/hook",
resolver=lambda host: ["10.1.2.3"])
self.assertIsNotNone(err)
def test_hostname_public_resolves_ok(self):
err = validate_push_target(
"https://hooks.example.com/hook",
resolver=lambda host: ["93.184.216.34"])
self.assertIsNone(err)
def test_dns_failure_rejected(self):
def _boom(host):
raise OSError("no dns in test")
self.assertIsNotNone(validate_push_target(
"https://hooks.example.com/hook", resolver=_boom))
def test_dispatch_loopback_never_posts(self):
# 真实发送路径（sender=None，生产服务端走的路径）：
# loopback 目标在预检即被挡，不产生任何 POST。
create_subscription(self.conn, "proj-a", "user-1", "webhook",
"http://127.0.0.1:9/hook", "instant")
result = dispatch_event(self.conn, "failed", "proj-a",
{"task_id": "t1"}, sender=None)
self.assertEqual(result["sent"], [])
self.assertEqual(len(result["errors"]), 1)
self.assertIn("blocked", result["errors"][0]["reason"])
class ServerP1TestCase(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.db_path = os.path.join(self.tmp.name, "control.db")
self.auth = AuthStore(self.db_path)
self.owner_id, self.owner_key = self.auth.create_user("owner")
self.member_id, self.member_key = self.auth.create_user("member")
self.outsider_id, self.outsider_key = self.auth.create_user(
"outsider")
self._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS")
self._old_gateway = os.environ.get("CLOUD_GATEWAY")
self._old_backup_root = os.environ.get("CLOUD_BACKUP_ROOT")
self._old_backup_dir = os.environ.get("CLOUD_BACKUP_DIR")
os.environ["CODEX_GLOBAL_ADMINS"] = self.owner_id
os.environ.pop("CLOUD_GATEWAY", None)
os.environ.pop("CLOUD_BACKUP_ROOT", None)
os.environ.pop("CLOUD_BACKUP_DIR", None)
self._old_max_projects = server_mod.MAX_PROJECTS_PER_USER
self.server = create_server(self.db_path, "127.0.0.1", 0)
self.port = self.server.server_address[1]
self.thread = threading.Thread(
target=self.server.serve_forever, daemon=True)
self.thread.start()
s, proj = _req(self.port, "POST", "/projects",
{"name": "owner/repo"}, self.owner_key)
self.assertEqual(s, 201, proj)
self.project_id = proj["id"]
self.auth.set_member(
self.project_id, self.owner_id, self.member_id, "maintainer")
def tearDown(self):
self.server.shutdown()
self.server.server_close()
self.thread.join(timeout=5)
try:
self.auth.conn.close()
except Exception:
pass
self.tmp.cleanup()
server_mod.MAX_PROJECTS_PER_USER = self._old_max_projects
if self._old_admins is None:
os.environ.pop("CODEX_GLOBAL_ADMINS", None)
else:
os.environ["CODEX_GLOBAL_ADMINS"] = self._old_admins
if self._old_gateway is None:
os.environ.pop("CLOUD_GATEWAY", None)
else:
os.environ["CLOUD_GATEWAY"] = self._old_gateway
if self._old_backup_root is None:
os.environ.pop("CLOUD_BACKUP_ROOT", None)
else:
os.environ["CLOUD_BACKUP_ROOT"] = self._old_backup_root
if self._old_backup_dir is None:
os.environ.pop("CLOUD_BACKUP_DIR", None)
else:
os.environ["CLOUD_BACKUP_DIR"] = self._old_backup_dir
# 恢复猴子补丁过的 server 模块函数（relay/rotate 相关测试用）。
for name in ("_validate_sub2api_key", "_list_sub2api_models",
"_rotate_sub2api_key"):
try:
mock.patch.object(server_mod, name).stop()
except Exception:
pass
# -- helpers ---------------------------------------------------
def _audit_actions(self, user_id=None):
conn = sqlite3.connect(self.db_path)
try:
rows = conn.execute(
"SELECT action, detail FROM audit_log"
+ (" WHERE user_id = ?" if user_id else ""),
(user_id,) if user_id else ()).fetchall()
finally:
conn.close()
return rows
def _set_quota(self, target, quota):
return _req(self.port, "POST", "/admin/quotas",
{"user_id": target, "quota_usd_micros": quota},
self.owner_key)
# -- 备份 -------------------------------------------------------
def test_backup_whitelist_and_mode(self):
root = os.path.join(self.tmp.name, "backup-root")
os.environ["CLOUD_BACKUP_ROOT"] = root
out_dir = os.path.join(root, "daily")
s, body = _req(self.port, "POST", "/admin/backup",
{"out_dir": out_dir}, self.owner_key)
self.assertEqual(s, 200, body)
self.assertTrue(os.path.exists(body["path"]))
self.assertEqual(stat.S_IMODE(os.stat(body["path"]).st_mode),
0o600)
details = [d for _a, d in self._audit_actions(self.owner_id)]
self.assertTrue(any("admin_backup" in a
for a, _d in self._audit_actions(
self.owner_id)))
self.assertTrue(any(out_dir in (d or "")
for d in details))
# 白名单外：系统目录与 .. 逃逸一律 400。
for bad in ("/etc", os.path.join(root, "..", "escape")):
s, body = _req(self.port, "POST", "/admin/backup",
{"out_dir": bad}, self.owner_key)
self.assertEqual(s, 400, (bad, body))
# -- workers 聚合 ------------------------------------------------
def test_workers_admin_full_member_aggregate_only(self):
s, _b = _req(self.port, "POST", "/workers/heartbeat",
{"worker_id": "w-p1", "executor": "docker",
"capacity": 2, "running": 0}, self.owner_key)
self.assertEqual(s, 200, _b)
s, body = _req(self.port, "GET", "/workers", key=self.owner_key)
self.assertEqual(s, 200)
self.assertIn("workers", body)
self.assertTrue(any(w.get("worker_id") == "w-p1"
for w in body["workers"]))
s, body = _req(self.port, "GET", "/workers", key=self.member_key)
self.assertEqual(s, 200)
self.assertNotIn("workers", body)
self.assertNotIn("w-p1", json.dumps(body))
self.assertEqual(body["workers_total"], 1)
self.assertEqual(body["workers_online"], 1)
self.assertEqual(body["workers_offline"], 0)
# -- github repos 过滤 -------------------------------------------
def test_github_repos_filtered_by_member_projects(self):
s, proj = _req(self.port, "POST", "/projects",
{"name": "with-repo",
"repo_url": "https://github.com/acme/alpha.git"},
self.owner_key)
self.assertEqual(s, 201, proj)
self.auth.set_member(proj["id"], self.owner_id, self.member_id,
"maintainer")
conn = sqlite3.connect(self.db_path)
try:
conn.execute(
"CREATE TABLE IF NOT EXISTS github_app_installations("
"installation_id TEXT, account_login TEXT,"
" repo_full_name TEXT, created_at REAL,"
" UNIQUE(installation_id, repo_full_name))")
conn.execute(
"INSERT OR IGNORE INTO github_app_installations"
" VALUES(?, ?, ?, ?)", ("7", "acme", "acme/alpha", 1.0))
conn.execute(
"INSERT OR IGNORE INTO github_app_installations"
" VALUES(?, ?, ?, ?)", ("7", "acme", "acme/beta", 1.0))
conn.commit()
finally:
conn.close()
s, body = _req(self.port, "GET", "/github/repos",
key=self.member_key)
self.assertEqual(s, 200, body)
names = [r["repo_full_name"] for r in body["repos"]]
self.assertEqual(names, ["acme/alpha"])
s, body = _req(self.port, "GET", "/github/repos",
key=self.outsider_key)
self.assertEqual(s, 200, body)
self.assertEqual(body["repos"], [])
# -- key 审计 -----------------------------------------------------
def test_api_key_audit_only_id_and_label(self):
s, body = _req(self.port, "POST", "/users/me/keys",
{"label": "p1-test"}, self.owner_key)
self.assertEqual(s, 201, body)
key_id = body["id"]
plain = body["api_key"]
s, _b = _req(self.port, "DELETE",
"/users/me/keys/" + key_id, key=self.owner_key)
self.assertEqual(s, 200, _b)
s, body = _req(self.port, "POST",
"/users/me/keys/revoke-others", {},
self.owner_key)
self.assertEqual(s, 200, body)
actions = {a for a, _d in self._audit_actions(self.owner_id)}
self.assertIn("api_key_created", actions)
self.assertIn("api_key_revoked", actions)
self.assertIn("api_key_revoke_others", actions)
blob = json.dumps(self._audit_actions(self.owner_id))
self.assertNotIn(plain, blob)
# -- 项目上限 ------------------------------------------------------
def test_project_limits(self):
s, body = _req(self.port, "POST", "/projects",
{"name": "x" * 101}, self.owner_key)
self.assertEqual(s, 400, body)
self.assertIn("100", body["error"])
s, body = _req(self.port, "POST", "/projects",
{"name": "ok", "id": "y" * 65}, self.owner_key)
self.assertEqual(s, 400, body)
self.assertIn("64", body["error"])
server_mod.MAX_PROJECTS_PER_USER = 1
# setUp 已建 1 个 owner 项目，限额 1 时再建即超限。
s, body = _req(self.port, "POST", "/projects",
{"name": "second"}, self.owner_key)
self.assertEqual(s, 400, body)
self.assertIn("max 1", body["error"])
# -- 协调员预留 ----------------------------------------------------
def test_coordinator_reserve_blocks_parent(self):
s, _b = self._set_quota(self.owner_id, 1000000)
self.assertEqual(s, 200, _b)
parts = [{"name": "a", "prompt": "做 a"},
{"name": "b", "prompt": "做 b"},
{"name": "c", "prompt": "做 c"}]
s, body = _req(self.port, "POST", "/tasks",
{"project_id": self.project_id, "prompt": "协调",
"coordinator_parts": parts}, self.owner_key)
self.assertEqual(s, 402, body)
self.assertEqual(body["error"],
"coordinator_quota_reserve_exceeded")
self.assertEqual(body["coordinator_parts"], 3)
s, _b = self._set_quota(self.owner_id, 10000000)
self.assertEqual(s, 200, _b)
s, body = _req(self.port, "POST", "/tasks",
{"project_id": self.project_id, "prompt": "协调",
"coordinator_parts": parts}, self.owner_key)
self.assertEqual(s, 201, body)
self.assertIn("parent task creator",
body.get("quota_attribution", ""))
self.assertEqual(body.get("quota_reserved_usd_micros"), 3000000)
# -- 自动化配额门 ----------------------------------------------------
def test_automation_create_quota_gate(self):
s, _b = self._set_quota(self.owner_id, 0)
self.assertEqual(s, 200, _b)
s, body = _req(self.port, "POST",
"/projects/%s/automations" % self.project_id,
{"name": "nightly", "prompt": "巡检"},
self.owner_key)
self.assertEqual(s, 402, body)
s, _b = self._set_quota(self.owner_id, 10000000)
self.assertEqual(s, 200, _b)
s, body = _req(self.port, "POST",
"/projects/%s/automations" % self.project_id,
{"name": "nightly", "prompt": "巡检"},
self.owner_key)
self.assertEqual(s, 201, body)
# -- sub2api 懒开通 / 轮换端点 --------------------------------------
def test_relay_models_lazy_provision_not_just_pending(self):
os.environ["CLOUD_GATEWAY"] = "sub2api"
orig_validate = server_mod._validate_sub2api_key
orig_list = server_mod._list_sub2api_models
try:
server_mod._validate_sub2api_key = (
lambda conn, uid: ("provisioned", "fake-key", 7, 9))
server_mod._list_sub2api_models = (
lambda origin, key: ["model-a"])
s, body = _req(self.port, "GET", "/relay/models",
key=self.owner_key)
self.assertEqual(s, 200, body)
self.assertEqual(body["models"], ["model-a"])
self.assertEqual(body["provisioning"], "provisioned")
server_mod._validate_sub2api_key = (
lambda conn, uid: (_ for _ in ()).throw(
Sub2apiError("网关暂不可用")))
s, body = _req(self.port, "GET", "/relay/models",
key=self.owner_key)
self.assertEqual(s, 502, body)
finally:
server_mod._validate_sub2api_key = orig_validate
server_mod._list_sub2api_models = orig_list
def test_gateway_rotate_endpoint(self):
s, body = _req(self.port, "POST", "/users/me/gateway/rotate",
{}, self.owner_key)
self.assertEqual(s, 400, body)
os.environ["CLOUD_GATEWAY"] = "sub2api"
orig_rotate = server_mod._rotate_sub2api_key
try:
server_mod._rotate_sub2api_key = (
lambda conn, uid: ("rotated", "fake-key", 7, 9))
s, body = _req(self.port, "POST", "/users/me/gateway/rotate",
{}, self.owner_key)
self.assertEqual(s, 200, body)
self.assertTrue(body["rotated"])
self.assertNotIn("api_key", body)
actions = {a for a, _d in self._audit_actions(self.owner_id)}
self.assertIn("gateway_key_rotated", actions)
finally:
server_mod._rotate_sub2api_key = orig_rotate
# -- 推送审计与分发 SSRF ----------------------------------------------
def test_push_audit_and_dispatch_ssrf(self):
s, sub = _req(self.port, "POST",
"/projects/%s/push/subscriptions" % self.project_id,
{"channel": "webhook",
"target": "https://example.com/hook",
"preference": "instant"}, self.owner_key)
self.assertEqual(s, 201, sub)
actions = {a for a, _d in self._audit_actions(self.owner_id)}
self.assertIn("push_subscription_created", actions)
s, _b = _req(self.port, "DELETE",
"/projects/%s/push/subscriptions/%s"
% (self.project_id, sub["id"]), key=self.owner_key)
self.assertEqual(s, 200, _b)
actions = {a for a, _d in self._audit_actions(self.owner_id)}
self.assertIn("push_subscription_deleted", actions)
# 内网目标：分发不 POST、sent 为空、错误含 delivery failed。
s, sub = _req(self.port, "POST",
"/projects/%s/push/subscriptions" % self.project_id,
{"channel": "webhook",
"target": "http://127.0.0.1:9/hook",
"preference": "instant"}, self.owner_key)
self.assertEqual(s, 201, sub)
s, body = _req(self.port, "POST", "/admin/push/dispatch",
{"event_type": "failed",
"project_id": self.project_id,
"payload": {"task_id": "t1"}}, self.owner_key)
self.assertEqual(s, 200, body)
self.assertEqual(body["sent"], [])
self.assertEqual(len(body["errors"]), 1)
self.assertIn("delivery failed", body["errors"][0]["reason"])
actions = {a for a, _d in self._audit_actions(self.owner_id)}
self.assertIn("admin_push_dispatch", actions)
# -- 配额双字段与控制台文案 --------------------------------------------
def test_quota_payload_has_unpriced_fields(self):
s, body = _req(self.port, "GET", "/users/me/quota",
key=self.owner_key)
self.assertEqual(s, 200, body)
self.assertIn("unpriced_tokens_used", body)
self.assertIn("unpriced_token_cap", body)
def test_console_p1_markers(self):
conn = http.client.HTTPConnection(
"127.0.0.1", self.port, timeout=10)
try:
conn.request("GET", "/console")
resp = conn.getresponse()
text = resp.read().decode("utf-8")
finally:
conn.close()
self.assertEqual(resp.status, 200)
for marker in (
"登记评审需求",
"记录允许",
"记录拒绝",
"当前版本只留回执，不拦截实时行为",
"模型回合结束、改动在 worktree 里尚未合并",
"控制面已标记 merged，不代表 git 主分支已合入",
"协调员与 goal/计划互斥",
"仅 failed/interrupted/merge-failed 可重试",
"仅 maintainer 可合并"):
self.assertIn(marker, text)
if __name__ == "__main__":
unittest.main()