From 55990745a8bac8b1689afe9c0d2285766204cfef Mon Sep 17 00:00:00 2001 From: "Zhenjiang.Zhang" Date: Mon, 9 Nov 2026 21:50:03 +0800 Subject: [PATCH] =?UTF-8?q?feat(vpc):=20=E6=94=AF=E6=8C=81=E5=9B=BD?= =?UTF-8?q?=E5=86=85=E7=89=88=E4=BC=81=E4=B8=9A=20VPC=20=E4=B8=93=E5=B1=9E?= =?UTF-8?q?=E5=9F=9F=EF=BC=88QD=5FCN=5FVPC=5FENDPOINT=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docker-compose.yml | 2 + docs/accounts.md | 23 +++++ docs/configuration.md | 3 +- qoder_accounts.py | 193 ++++++++++++++++++++++++++++++++---- qoder_proxy.py | 38 ++++++- tests/_test_vpc_endpoint.py | 182 ++++++++++++++++++++++++++++++++++ 6 files changed, 419 insertions(+), 22 deletions(-) create mode 100644 tests/_test_vpc_endpoint.py diff --git a/docker-compose.yml b/docker-compose.yml index 77b34e2..a95393a 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -27,6 +27,8 @@ services: # - API_KEY=your_api_key_here # ↑ 留空：容器以 --lan 启动，首次自动生成随机 Key 写入 ./accounts/settings.json # 并打印在日志（docker compose logs qoder-proxy | grep -i "api key"） + # 企业 VPC（仅 CN）：实例名或 acme.vpc.qoder.com.cn + # - QD_CN_VPC_ENDPOINT=acme - TZ=Asia/Shanghai volumes: # 持久化挂载账号凭证与用量流水 diff --git a/docs/accounts.md b/docs/accounts.md index 9327c7c..90b0e0d 100644 --- a/docs/accounts.md +++ b/docs/accounts.md @@ -131,6 +131,29 @@ > #90 之前只认第 2、3 条，而 userinfo **根本不返回 `user_type`**，所以企业号会被误判成个人号：团队闸门不生效、共享额度不计入有效余额。现在组织字段是第一判据。 +### 企业 VPC 专属域（CN） + +如果企业购买的是 Qoder CN（全家桶）企业 VPC 版，服务入口不是公网 `gateway.qoder.com.cn`，而是实例专属域。启动网关前设置： + +```bash +export QD_CN_VPC_ENDPOINT=acme # 也可以填 acme.vpc.qoder.com.cn +./start-qoder-proxy.sh +``` + +网关按官方 CLI 的固定映射推导三个端点： + +| 用途 | 地址 | +|---|---| +| 登录 / 官网 | `https://acme.vpc.qoder.com.cn` | +| OpenAPI | `https://acme-openapi.vpc.qoder.com.cn` | +| 推理网关 | `https://acme-gateway.vpc.qoder.com.cn` | + +- 只对国内版生效；国际版账号仍走公网 `api3.qoder.sh` +- VPC 模式**不会回退**到公网 openapi / gateway，避免把企业凭证发到错误入口 +- 网关所在机器必须能访问企业 VPC（专线、VPN 或内网 DNS）。专属域解析到内网地址是预期行为，SSRF 防护只放行这三个推导出来的主机 +- 「添加账号（OAuth）」「导入 PAT」会走 VPC 的 OpenAPI；「扫描本机凭证」会同时检查 `~/.qoder-cn/.auth` 和 `~/.qoder-cn/.vpc/<实例名>/.auth` +- 启动日志会打印 `vpc : acme (openapi=... gateway=...)`；如果当前出口仍是国际版，会额外打印 WARN + ### 「企业号抓不到」的常见原因 | 现象 | 原因 | 处理 | diff --git a/docs/configuration.md b/docs/configuration.md index 78e6edc..f2be6a3 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -43,7 +43,7 @@ python qoder_proxy.py [--host HOST] [--port PORT] [--lan] [--api-key KEY] - 每个 Key 有名称，可以单独启用 / 停用 / 删除，删除后立即失效 - 每个 Key 可以绑定出口： - **国际版**：走 `api3.qoder.sh`，连不上时依次切换到 `api2` / `api1` - - **国内版**：走 `gateway.qoder.com.cn` + - **国内版**：走 `gateway.qoder.com.cn`；设置 `QD_CN_VPC_ENDPOINT` 后走企业实例专属网关 - **不绑定**：跟随看板顶部的全局出口 - 请求的模型和 Key 绑定的出口不匹配时，直接返回 400，给出说明 @@ -147,6 +147,7 @@ Redis 仅开放 Compose 内部网络，数据由专用命名卷保存。 | `QD_AFFINITY_MAX_MSGS` | `0` | 对话超过 N 条消息后不再固定账号，`0` 表示不限制 | | `QD_AFFINITY_DEBUG` | `0` | 打印会话亲和的调试日志 | | `QD_INTL_GATEWAY` | `https://api3.qoder.sh` | 国际版推理主机。连接失败时依次切换到 `api2` / `api1` | +| `QD_CN_VPC_ENDPOINT` | — | 国内版企业 VPC 实例名或专属域（如 `acme` / `acme.vpc.qoder.com.cn`）。设置后自动推导 openapi / gateway / website 三个专属域，且不回退公网。仅 CN 生效，见[企业 VPC 专属域](accounts.md#企业-vpc-专属域cn) | ### 会话缓存保温（可选） diff --git a/qoder_accounts.py b/qoder_accounts.py index b9ae5ef..7e03c24 100644 --- a/qoder_accounts.py +++ b/qoder_accounts.py @@ -252,15 +252,111 @@ def normalize_quota_payload(payload, realm=""): return out +VPC_DOMAIN_SUFFIX = "vpc.qoder.com.cn" + + +def parse_vpc_endpoint(value): + """把 QD_CN_VPC_ENDPOINT 解析成企业 VPC 的三个官方主机。 + + 接受实例名（acme）或三个按服务拆分的域名（acme.vpc.qoder.com.cn / + acme-openapi.vpc.qoder.com.cn / acme-gateway.vpc.qoder.com.cn）。 + 空值返回 None；路径、端口、凭据、query、fragment 或空白一律拒绝， + 避免一个值只改掉流程里的一部分端点。 + """ + raw = str(value or "") + if not raw.strip(): + return None + if raw != raw.strip() or any(ch in raw for ch in "\r\n\t"): + raise ValueError("invalid VPC endpoint: surrounding whitespace or " + "control characters") + target = raw if "://" in raw else "https://" + raw + try: + parsed = urllib.parse.urlsplit(target) + port = parsed.port + except ValueError: + raise ValueError("invalid VPC endpoint: malformed host or port") + if (parsed.scheme not in ("http", "https") or not parsed.hostname + or port is not None or parsed.username is not None + or parsed.password is not None or parsed.query + or parsed.fragment or "#" in raw + or parsed.path not in ("", "/")): + raise ValueError("invalid VPC endpoint: expected an instance name or a " + "domain under %s" % VPC_DOMAIN_SUFFIX) + host = parsed.hostname.lower() + instance = host + if host.endswith("." + VPC_DOMAIN_SUFFIX): + instance = host[:-len("." + VPC_DOMAIN_SUFFIX)] + if instance.endswith("-openapi"): + instance = instance[:-len("-openapi")] + elif instance.endswith("-gateway"): + instance = instance[:-len("-gateway")] + elif "." in host: + raise ValueError("invalid VPC endpoint: expected an instance name or a " + "domain under %s" % VPC_DOMAIN_SUFFIX) + if not re.fullmatch(r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?", instance): + raise ValueError("invalid VPC endpoint: bad instance name %r" % instance) + domain = instance + "." + VPC_DOMAIN_SUFFIX + return { + "instance": instance, + "domain": domain, + "website": "https://" + domain, + "openapi": "https://" + instance + "-openapi." + VPC_DOMAIN_SUFFIX, + "gateway": "https://" + instance + "-gateway." + VPC_DOMAIN_SUFFIX, + } + + +_VPC_CN = None +VPC_CONFIG_ERROR = "" +try: + _VPC_CN = parse_vpc_endpoint(os.environ.get("QD_CN_VPC_ENDPOINT")) +except ValueError as exc: + VPC_CONFIG_ERROR = str(exc) + +_VPC_HOSTS = frozenset( + urllib.parse.urlsplit(_VPC_CN[key]).hostname.lower() + for key in ("website", "openapi", "gateway") + if _VPC_CN and urllib.parse.urlsplit(_VPC_CN[key]).hostname +) + + +def vpc_configured(): + return _VPC_CN is not None + + +def vpc_instance(): + return (_VPC_CN or {}).get("instance", "") + + +def vpc_config_error(): + return VPC_CONFIG_ERROR + + +def vpc_endpoint_hosts(): + return _VPC_HOSTS + + +def is_configured_vpc_host(host): + """仅对 QD_CN_VPC_ENDPOINT 推导出的三个主机放行解析级 SSRF 校验。 + + 企业 VPC 域名通常解析到公司内网地址，通用规则会把它当内网目标拒绝； + 但这是用户显式配置的固定主机，不是请求里可注入的 URL。 + """ + name = str(host or "").strip().strip("[]").lower() + return bool(name) and name in vpc_endpoint_hosts() + + REALM_CONFIGS = { "cn": { "name": "国内版 (China)", - "openapi": "https://openapi.qoder.com.cn", - "gateway": "https://gateway.qoder.com.cn", - "website": "https://qoder.com.cn", + "openapi": (_VPC_CN or {}).get("openapi") + or "https://openapi.qoder.com.cn", + "gateway": (_VPC_CN or {}).get("gateway") + or "https://gateway.qoder.com.cn", + "website": (_VPC_CN or {}).get("website") or "https://qoder.com.cn", + "domain": (_VPC_CN or {}).get("domain") or "qoder.com.cn", + "vpc_instance": (_VPC_CN or {}).get("instance", ""), "client_id": "1c5e33e1-364d-4ce6-b02c-acaa81274a5c", "redirect_uri": "qoder-work-cn://", - "domain": "qoder.com.cn", "ua": "QoderWork/1.1.75", # has_checkin 只是"历史上该区域曾开放 sash 签到"的提示位，**不再作为 # 门控**：能力改为运行时探测（见 Account.checkin_capability）。官方把 @@ -1188,6 +1284,10 @@ def openapi_candidates(realm): base = str(cfg.get("openapi") or "").rstrip("/") if base: out.append(base) + # 企业 VPC：只能访问实例专属域，绝不回退到公网 openapi（那会把企业 + # 凭证发给另一个入口，且通常直接 401/404）。 + if cfg.get("vpc_instance"): + return out domain = str(cfg.get("domain") or "").strip() # 国际版活动平台另有一个可用入口 center.qoder.sh（0.4.3 客户端仍引用；实测 # intl=200 / cn=401），放在官网域名之前作为回退。 @@ -1210,12 +1310,32 @@ def gateway_candidates(realm): """ cfg = get_realm_config(realm) out = [cfg["gateway"]] + # 企业 VPC 只有实例专属推理域，公网 api1/api2/api3 不能作为故障回退。 + if cfg.get("vpc_instance"): + return out for host in cfg.get("gateway_fallbacks") or (): if host and host not in out: out.append(host) return out +def cli_auth_dirs(realm): + """官方 CLI 的凭据目录：公网 ``.auth`` +（VPC 模式）实例专属 ``.auth``。 + + 企业 VPC 账号可能由官方 CLI 登录在 ``~/.qoder-cn/.auth``，也可能按 + 实例隔离在 ``~/.qoder-cn/.vpc//.auth``；扫描两者，导入时 + 的 realm 仍是 cn。 + """ + cfg = get_realm_config(realm) + root = os.path.join(os.path.expanduser("~"), cfg["home_dir"]) + out = [] + instance = str(cfg.get("vpc_instance") or "") + if instance: + out.append(os.path.join(root, ".vpc", instance, ".auth")) + out.append(os.path.join(root, ".auth")) + return out + + def detect_realm_from_domain(domain): d = str(domain or "").lower() if "qoder.sh" in d or (d.endswith("qoder.com") and "qoder.com.cn" not in d) \ @@ -1470,6 +1590,19 @@ def fold_daily_usage(log_path=None, day=None, max_lines=200000): ipaddress.ip_network("2001:2::/48"), ipaddress.ip_network("fdfe:dcba:9876::/48"), ] +# macOS 的 getaddrinfo 会把 fake-IP 以 IPv4-translated 形式（::ffff:0:c612:2d） +# 一并返回；只认纯 IPv4 会让 all(...) 失败，于是所有公网域名都被误判成内网。 +_FAKEIP_V4_TRANSLATED = ipaddress.ip_network("::ffff:0:0:0/96") + + +def _is_fake_ip(ip): + candidates = [ip] + if isinstance(ip, ipaddress.IPv6Address): + if ip.ipv4_mapped is not None: + candidates.append(ip.ipv4_mapped) + elif ip in _FAKEIP_V4_TRANSLATED: + candidates.append(ipaddress.IPv4Address(int(ip) & 0xFFFFFFFF)) + return any(addr in net for addr in candidates for net in _FAKEIP_NETS) def _host_boundary_violation(ip, allow_local): @@ -1496,6 +1629,10 @@ def validate_public_http_url(url, allow_local=False): host = (parsed.hostname or "").strip().strip("[]").lower() if not host: raise ValueError("URL host is required") + # 企业 VPC 专属域通常解析到公司内网，不能按公网目标拒绝；只放行 + # QD_CN_VPC_ENDPOINT 推导出的固定主机。 + if is_configured_vpc_host(host): + return url if host == "localhost" or host.endswith(".localhost") or host.endswith(".local"): if not allow_local: raise ValueError("requests to localhost are not allowed") @@ -1521,7 +1658,7 @@ def validate_public_http_url(url, allow_local=False): addrs.append(ipaddress.ip_address(addr)) except ValueError: raise ValueError("URL host resolved to a non-IP address: %r" % addr) - if addrs and all(any(ip in net for net in _FAKEIP_NETS) for ip in addrs): + if addrs and all(_is_fake_ip(ip) for ip in addrs): return url # fake-IP DNS：真实 IP 不可见，名称级校验已通过 for ip in addrs: if _host_boundary_violation(ip, allow_local): @@ -1602,7 +1739,8 @@ def upstream_clock_skew(timeout=4.0): """ from email.utils import parsedate_to_datetime try: - req = urllib.request.Request("https://openapi.qoder.com.cn/", method="HEAD") + base = openapi_candidates("cn")[0] + req = urllib.request.Request(base.rstrip("/") + "/", method="HEAD") with _opener_for_request().open(req, timeout=timeout) as resp: date_hdr = resp.headers.get("Date") if not date_hdr: @@ -4495,16 +4633,15 @@ def import_rows(self, rows, realm=None, overwrite=False, group=None): @staticmethod def _local_machine_id(realm): """读取本机官方客户端的 machine_id（优先，保持设备一致），缺则生成。""" - cfg = get_realm_config(realm) - home = os.path.join(os.path.expanduser("~"), cfg["home_dir"], ".auth") - p = os.path.join(home, "machine_id") - try: - with open(p, encoding="utf-8") as fh: - mid = fh.read().strip() - if mid: - return mid - except Exception: - pass + for auth_dir in cli_auth_dirs(realm): + try: + with open(os.path.join(auth_dir, "machine_id"), + encoding="utf-8") as fh: + mid = fh.read().strip() + if mid: + return mid + except Exception: + pass return str(uuid.uuid4()) def start_login(self, realm="cn", platform="CLI"): @@ -4820,9 +4957,10 @@ def scan_desktop_credentials(): item["error"] = str(exc) found.append(item) - # 2) CLI 端 user / user.{profile} - auth_dir = os.path.join(os.path.expanduser("~"), cfg["home_dir"], ".auth") - if os.path.isdir(auth_dir): + # 2) CLI 端 user / user.{profile}（含 VPC 实例专属目录） + for auth_dir in cli_auth_dirs(realm): + if not os.path.isdir(auth_dir): + continue machine_key = "" try: with open(os.path.join(auth_dir, "machine_id"), @@ -4899,6 +5037,8 @@ def import_desktop_credential(path=None, realm=None): token = refresh = "" uid = nickname = "" + user_type = DEFAULT_USER_TYPE + org_id = org_name = "" exp = 0 base = os.path.basename(path) if base == "auth.v1.dat": @@ -4908,6 +5048,12 @@ def import_desktop_credential(path=None, realm=None): user = data.get("user") or {} uid = str(user.get("id") or "") nickname = str(user.get("name") or "") + user_type = (str(user.get("user_type") or user.get("userType") or "") + or DEFAULT_USER_TYPE) + org_id = str(user.get("organization_id") + or user.get("organizationId") or "") + org_name = str(user.get("organization_name") + or user.get("organizationName") or "") try: import datetime exp = int(datetime.datetime.strptime( @@ -4929,6 +5075,12 @@ def import_desktop_credential(path=None, realm=None): refresh = str(data.get("refresh_token") or "") uid = str(data.get("uid") or "") nickname = str(data.get("name") or "") + user_type = (str(data.get("user_type") or data.get("userType") or "") + or DEFAULT_USER_TYPE) + org_id = str(data.get("organization_id") + or data.get("organizationId") or "") + org_name = str(data.get("organization_name") + or data.get("organizationName") or "") exp = normalize_epoch(data.get("expire_time")) if not token: @@ -4946,6 +5098,9 @@ def import_desktop_credential(path=None, realm=None): "expiresAt": exp or (int(time.time()) + 30 * 86400), "source": "desktop-app", "enabled": True, + "userType": user_type, + "organizationId": org_id, + "organizationName": org_name, }) return add_to_pool(account) diff --git a/qoder_proxy.py b/qoder_proxy.py index 1da657f..4a2ea65 100644 --- a/qoder_proxy.py +++ b/qoder_proxy.py @@ -119,6 +119,19 @@ def key_model_message(entry, model): ipaddress.ip_network("2001:2::/48"), ipaddress.ip_network("fdfe:dcba:9876::/48"), ] +# 与 qoder_accounts 同款：macOS 的 getaddrinfo 会同时返回 IPv4 与 +# IPv4-translated IPv6（::ffff:0:c612:2d），需要按内嵌 IPv4 识别 fake-IP。 +_FAKEIP_V4_TRANSLATED = ipaddress.ip_network("::ffff:0:0:0/96") + + +def _is_fake_ip(ip): + candidates = [ip] + if isinstance(ip, ipaddress.IPv6Address): + if ip.ipv4_mapped is not None: + candidates.append(ip.ipv4_mapped) + elif ip in _FAKEIP_V4_TRANSLATED: + candidates.append(ipaddress.IPv4Address(int(ip) & 0xFFFFFFFF)) + return any(addr in net for addr in candidates for net in _FAKEIP_NETS) def _host_boundary_violation(ip, allow_local): @@ -145,6 +158,10 @@ def validate_public_http_url(url, allow_local=False): host = (parsed.hostname or "").strip().strip("[]").lower() if not host: raise ValueError("URL host is required") + # 企业 VPC 专属域通常解析到公司内网；只放行 QD_CN_VPC_ENDPOINT + # 推导出的固定主机，普通公网 URL 的校验逻辑保持不变。 + if qoder_accounts.is_configured_vpc_host(host): + return url if host == "localhost" or host.endswith(".localhost") or host.endswith(".local"): if not allow_local: raise ValueError("requests to localhost are not allowed") @@ -170,7 +187,7 @@ def validate_public_http_url(url, allow_local=False): addrs.append(ipaddress.ip_address(addr)) except ValueError: raise ValueError("URL host resolved to a non-IP address: %r" % addr) - if addrs and all(any(ip in net for net in _FAKEIP_NETS) for ip in addrs): + if addrs and all(_is_fake_ip(ip) for ip in addrs): return url # fake-IP DNS：真实 IP 不可见，名称级校验已通过 for ip in addrs: if _host_boundary_violation(ip, allow_local): @@ -14403,6 +14420,14 @@ def main(): help="write the server PID to this file (stop.bat uses it; " "removed on exit)") args = ap.parse_args() + if qoder_accounts.vpc_config_error(): + sys.stderr.write( + "[ERROR] invalid QD_CN_VPC_ENDPOINT: %s\n" + " expected an instance name (e.g. acme) or one of\n" + " acme.vpc.qoder.com.cn / acme-openapi.vpc.qoder.com.cn /\n" + " acme-gateway.vpc.qoder.com.cn\n" + % qoder_accounts.vpc_config_error()) + raise SystemExit(1) qoder_logpaths.install_stream_logging(os.path.dirname(os.path.abspath(__file__))) if not args.usage_dir and not os.environ.get("QD_PROXY_USAGE_DIR"): _migration = qoder_logpaths.migrate_legacy_usage( @@ -14615,7 +14640,16 @@ def _probe_health(timeout=2.0): rep.domain if rep else "-")) log("realm : %s (%s)" % ( CURRENT_REALM, - "qoder.com" if CURRENT_REALM == "intl" else "qoder.com.cn")) + get_realm_config(CURRENT_REALM).get("domain") or "qoder.com.cn")) + if qoder_accounts.vpc_configured(): + _cn = qoder_accounts.get_realm_config("cn") + log("vpc : %s (openapi=%s gateway=%s)" + % (qoder_accounts.vpc_instance(), _cn["openapi"], _cn["gateway"]), + tag="realm") + if CURRENT_REALM != "cn": + log("VPC endpoint is CN-only, but the current realm is %s - " + "switch to CN in the dashboard or set QD_PROXY_DEFAULT_REALM=cn" + % CURRENT_REALM, level="WARN", tag="realm") log("catalog : %s" % (BASEPROMPT_PATH if BASEPROMPT else "baseprompt.json MISSING")) diff --git a/tests/_test_vpc_endpoint.py b/tests/_test_vpc_endpoint.py new file mode 100644 index 0000000..9d334e2 --- /dev/null +++ b/tests/_test_vpc_endpoint.py @@ -0,0 +1,182 @@ +"""QD_CN_VPC_ENDPOINT 企业 VPC 专属域套件。 + +判据： + 1. 解析只接受实例名或 vpc.qoder.com.cn 下的三种官方域名，拒绝路径 / + 端口 / 凭据 / query / fragment，防止只改掉流程里的一部分端点； + 2. 配置后 CN 的 openapi / gateway / website 全部切换到专属域，且 + openapi / gateway 候选**不允许**回退到公网； + 3. 专属域解析到内网地址时，两处 SSRF 校验只放行这三个推导主机； + 未配置的 vpc.qoder.com.cn 域名仍按公网规则拒绝； + 4. 未设置环境变量时，现有 CN 行为逐字节不变。 + + python tests/_test_vpc_endpoint.py +""" +import json +import os +import subprocess +import sys + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, HERE) +from _suite_head import ROOT, bootstrap, check, declare_suite, finish # noqa: E402 + +os.environ.pop("QD_CN_VPC_ENDPOINT", None) +os.environ.pop("QD_CN_OPENAPI", None) +P, A = bootstrap() +declare_suite(__file__, ["qoder_accounts.py", "qoder_proxy.py"]) + +print("\n[vpc] 1) 解析：实例名与三种官方域名") +CASES = { + "acme": ("acme", "https://acme.vpc.qoder.com.cn", + "https://acme-openapi.vpc.qoder.com.cn", + "https://acme-gateway.vpc.qoder.com.cn"), + "ACME.VPC.QODER.COM.CN": ("acme", "https://acme.vpc.qoder.com.cn", + "https://acme-openapi.vpc.qoder.com.cn", + "https://acme-gateway.vpc.qoder.com.cn"), + "https://acme-openapi.vpc.qoder.com.cn/": ( + "acme", "https://acme.vpc.qoder.com.cn", + "https://acme-openapi.vpc.qoder.com.cn", + "https://acme-gateway.vpc.qoder.com.cn"), + "acme-gateway.vpc.qoder.com.cn": ( + "acme", "https://acme.vpc.qoder.com.cn", + "https://acme-openapi.vpc.qoder.com.cn", + "https://acme-gateway.vpc.qoder.com.cn"), +} +for value, expected in CASES.items(): + got = A.parse_vpc_endpoint(value) + actual = ((got or {}).get("instance"), (got or {}).get("website"), + (got or {}).get("openapi"), (got or {}).get("gateway")) + check("解析 %r" % value, actual == expected, actual) + +check("空值 = 未配置（返回 None）", A.parse_vpc_endpoint("") is None) + +print("\n[vpc] 2) 解析：拒绝部分重定向输入") +BAD = ( + " acme", "acme ", "acme\n", "acme:8443", + "acme/path", "acme?x=1", "acme#frag", + "other.example.com", "acme.vpc.qoder.com.cn.evil.test", + "ftp://acme", "u:p@acme.vpc.qoder.com.cn", + "acme..vpc.qoder.com.cn", "-acme", "acme-", +) +for value in BAD: + try: + A.parse_vpc_endpoint(value) + ok = False + except ValueError: + ok = True + check("拒绝 %r" % value, ok) + +print("\n[vpc] 3) 未配置：现有 CN 行为不变") +check("CN openapi 仍是公网", + A.get_realm_config("cn")["openapi"] == "https://openapi.qoder.com.cn") +check("CN gateway 仍是单一官方主机", + A.gateway_candidates("cn") == ["https://gateway.qoder.com.cn"]) +check("CN openapi 候选不含 VPC 域", + not any("vpc.qoder.com.cn" in c for c in A.openapi_candidates("cn"))) +check("未配置时 is_configured_vpc_host 为 False", + not A.is_configured_vpc_host("acme-openapi.vpc.qoder.com.cn")) + + +def run_py(code, env_extra): + env = dict(os.environ) + env.pop("QD_CN_OPENAPI", None) + env["PYTHONPATH"] = os.pathsep.join( + [ROOT] + ([env["PYTHONPATH"]] if env.get("PYTHONPATH") else [])) + env.update(env_extra) + out = subprocess.check_output([sys.executable, "-c", code], + cwd=ROOT, env=env, text=True, + stderr=subprocess.STDOUT, timeout=60) + return out.strip().splitlines()[-1] + + +print("\n[vpc] 4) 子进程：QD_CN_VPC_ENDPOINT=acme 的模块状态") +CODE = r''' +import json +import qoder_accounts as A +cn = A.get_realm_config("cn") +print(json.dumps({ + "error": A.vpc_config_error(), + "instance": A.vpc_instance(), + "openapi": cn["openapi"], + "gateway": cn["gateway"], + "website": cn["website"], + "domain": cn["domain"], + "openapi_candidates": A.openapi_candidates("cn"), + "gateway_candidates": A.gateway_candidates("cn"), + "hosts": sorted(A.vpc_endpoint_hosts()), +}, ensure_ascii=False)) +''' +data = json.loads(run_py(CODE, {"QD_CN_VPC_ENDPOINT": "acme"})) +check("配置读取无错误", data["error"] == "", data["error"]) +check("实例名 = acme", data["instance"] == "acme", data["instance"]) +check("openapi = 专属域", + data["openapi"] == "https://acme-openapi.vpc.qoder.com.cn", data["openapi"]) +check("gateway = 专属域", + data["gateway"] == "https://acme-gateway.vpc.qoder.com.cn", data["gateway"]) +check("website = 专属域", + data["website"] == "https://acme.vpc.qoder.com.cn", data["website"]) +check("domain = 专属域", + data["domain"] == "acme.vpc.qoder.com.cn", data["domain"]) +check("openapi 候选只有专属域（不回退公网）", + data["openapi_candidates"] == ["https://acme-openapi.vpc.qoder.com.cn"], + data["openapi_candidates"]) +check("gateway 候选只有专属域（不回退 api1/2/3）", + data["gateway_candidates"] == ["https://acme-gateway.vpc.qoder.com.cn"], + data["gateway_candidates"]) +check("放行主机集合 = 三个专属域", data["hosts"] == [ + "acme-gateway.vpc.qoder.com.cn", + "acme-openapi.vpc.qoder.com.cn", + "acme.vpc.qoder.com.cn", +], data["hosts"]) + +print("\n[vpc] 5) 子进程：非法值会留下启动错误") +err = run_py("import qoder_accounts as A; print(A.vpc_config_error())", + {"QD_CN_VPC_ENDPOINT": "bad host"}) +check("非法值产生非空错误", bool(err) and "invalid VPC endpoint" in err, err) + +print("\n[vpc] 6) 子进程：内网解析时只放行配置的专属域") +CODE_SSRF = r''' +import json +import socket + +_orig = socket.getaddrinfo + + +def fake(host, port, *args, **kwargs): + if str(host).lower().endswith(".vpc.qoder.com.cn"): + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", + ("10.20.30.40", port or 443))] + return _orig(host, port, *args, **kwargs) + + +socket.getaddrinfo = fake +import qoder_accounts as A +import qoder_proxy as P + +res = {} +for name, fn in (("accounts", A.validate_public_http_url), + ("proxy", P.validate_public_http_url)): + def probe(url): + try: + fn(url) + return "ok" + except Exception as exc: + return "blocked:%s" % exc + res[name + "_vpc"] = probe( + "https://acme-openapi.vpc.qoder.com.cn/api/v1/userinfo") + res[name + "_other"] = probe("https://other.vpc.qoder.com.cn/api/v1/userinfo") +print(json.dumps(res, ensure_ascii=False)) +''' +ssrf = json.loads(run_py(CODE_SSRF, {"QD_CN_VPC_ENDPOINT": "acme"})) +check("qoder_accounts 放行配置的专属域（即使解析到内网）", + ssrf.get("accounts_vpc") == "ok", ssrf.get("accounts_vpc")) +check("qoder_proxy 放行配置的专属域（即使解析到内网）", + ssrf.get("proxy_vpc") == "ok", ssrf.get("proxy_vpc")) +check("未配置的 vpc.qoder.com.cn 仍按内网目标拒绝", + str(ssrf.get("accounts_other", "")).startswith("blocked:"), + ssrf.get("accounts_other")) +check("qoder_proxy 同样拒绝未配置的同后缀域名", + str(ssrf.get("proxy_other", "")).startswith("blocked:"), + ssrf.get("proxy_other")) + +finish()