From 3d0b35018fadd7d59448e729cb53f94290d723ce Mon Sep 17 00:00:00 2001 From: zizi Date: Wed, 22 Jul 2026 21:52:19 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D:=20=E6=94=AF=E6=8C=81?= =?UTF-8?q?=E5=8F=97=E6=8E=A7=20Claude=20=E8=87=AA=E5=AE=9A=E4=B9=89?= =?UTF-8?q?=E7=BD=91=E5=85=B3=E8=AE=A4=E8=AF=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../replay-eval/scripts/claude_runtime.py | 48 ++++++- .../scripts/test_claude_runtime.py | 122 +++++++++++++++++- 2 files changed, 167 insertions(+), 3 deletions(-) diff --git a/.claude/skills/replay-eval/scripts/claude_runtime.py b/.claude/skills/replay-eval/scripts/claude_runtime.py index ba5daaf..07cd8da 100644 --- a/.claude/skills/replay-eval/scripts/claude_runtime.py +++ b/.claude/skills/replay-eval/scripts/claude_runtime.py @@ -21,6 +21,7 @@ import time from dataclasses import dataclass from decimal import Decimal, InvalidOperation, ROUND_HALF_UP from typing import Any, Callable, Mapping, Sequence +from urllib.parse import urlsplit SANDBOX_EXECUTABLE = "/usr/bin/sandbox-exec" @@ -30,6 +31,8 @@ ENVIRONMENT_ALLOWLIST = frozenset( { "ANTHROPIC_API_KEY", "CLAUDE_CODE_OAUTH_TOKEN", + "ANTHROPIC_AUTH_TOKEN", + "ANTHROPIC_BASE_URL", "HTTPS_PROXY", "HTTP_PROXY", "NO_PROXY", @@ -43,7 +46,9 @@ ENVIRONMENT_ALLOWLIST = frozenset( "NODE_EXTRA_CA_CERTS", } ) -AUTHENTICATION_FIELDS = frozenset({"ANTHROPIC_API_KEY", "CLAUDE_CODE_OAUTH_TOKEN"}) +AUTHENTICATION_FIELDS = frozenset( + {"ANTHROPIC_API_KEY", "CLAUDE_CODE_OAUTH_TOKEN", "ANTHROPIC_AUTH_TOKEN"} +) HASH_PATTERN = re.compile(r"^(?:sha256:)?[0-9a-f]{64}$") MODEL_ID_PATTERN = re.compile( r"^(?=.{6,128}$)[A-Za-z0-9][A-Za-z0-9._:-]{5,127}(?:\[[A-Za-z0-9._:-]+\])?$" @@ -436,6 +441,39 @@ def build_sandbox_command( ] +def _validate_base_url(value: str) -> None: + """拒绝会把认证信息送往非 HTTP(S) 或带隐藏凭据的地址。""" + + invalid_message = "ANTHROPIC_BASE_URL 必须是合法且不含凭据、查询或片段的 HTTP(S) 地址" + if ( + not isinstance(value, str) + or not value + or value != value.strip() + or any(char.isspace() or ord(char) < 0x20 or ord(char) == 0x7F for char in value) + or "?" in value + or "#" in value + ): + raise ValueError(invalid_message) + try: + parsed = urlsplit(value) + hostname = parsed.hostname + port = parsed.port + username = parsed.username + password = parsed.password + except ValueError: + # 解析器的底层异常可能携带原始 URL，不能把它继续传播到安全错误边界。 + raise ValueError(invalid_message) from None + if ( + parsed.scheme not in {"http", "https"} + or not parsed.netloc + or not hostname + or username is not None + or password is not None + or (parsed.netloc.endswith(":") and port is None) + ): + raise ValueError(invalid_message) + + def _minimal_environment( source: Mapping[str, str], isolation_directory: pathlib.Path, *, require_authentication: bool ) -> dict[str, str]: @@ -449,6 +487,14 @@ def _minimal_environment( authentication = AUTHENTICATION_FIELDS.intersection(environment) if len(authentication) > 1: raise ValueError("一次调用只能使用一种授权认证") + auth_token = environment.get("ANTHROPIC_AUTH_TOKEN") + base_url = environment.get("ANTHROPIC_BASE_URL") + if auth_token is not None and not auth_token.strip(): + raise ValueError("ANTHROPIC_AUTH_TOKEN 不能为空") + if auth_token is not None and (base_url is None or not base_url.strip()): + raise ValueError("使用 ANTHROPIC_AUTH_TOKEN 时必须同时配置 ANTHROPIC_BASE_URL") + if base_url is not None: + _validate_base_url(base_url) if require_authentication and not authentication: raise ValueError("真实调用缺少经授权的 Claude 认证") environment["HOME"] = str(isolation_directory) diff --git a/.claude/skills/replay-eval/scripts/test_claude_runtime.py b/.claude/skills/replay-eval/scripts/test_claude_runtime.py index 384cc28..5c214f0 100644 --- a/.claude/skills/replay-eval/scripts/test_claude_runtime.py +++ b/.claude/skills/replay-eval/scripts/test_claude_runtime.py @@ -19,6 +19,7 @@ sys.path.insert(0, str(SCRIPT_DIR)) from claude_runtime import ( # noqa: E402 ClaudeRuntimeError, ExecutionProfile, + _minimal_environment, build_sandbox_command, canonical_json, run_claude, @@ -168,12 +169,20 @@ class ClaudeRuntimeTest(unittest.TestCase): profile = execution_profile() runner = runner_for(success_envelope()) + auth_token = "auth-token-must-not-leak" + base_url = "https://gateway.example.invalid/anthropic/v1" result = run_claude( profile, {"request": "secret-body"}, runner=runner, binding_verifier=lambda _profile: None, - source_environment={"LANG": "zh_CN.UTF-8", "DATABASE_URL": "secret"}, + source_environment={ + "LANG": "zh_CN.UTF-8", + "ANTHROPIC_AUTH_TOKEN": auth_token, + "ANTHROPIC_BASE_URL": base_url, + "DATABASE_URL": "secret", + "UNRELATED_SECRET": "must-be-stripped", + }, ) self.assertEqual(result.structured_output, {"value": "ok"}) @@ -183,20 +192,114 @@ class ClaudeRuntimeTest(unittest.TestCase): self.assertEqual(receipt["totalCostUsd"], "0.120000") self.assertEqual(receipt["structuredOutputSha256"], sha256_json({"value": "ok"})) self.assertNotIn("secret-body", canonical_json(receipt)) + self.assertNotIn(auth_token, canonical_json(receipt)) + self.assertNotIn(base_url, canonical_json(receipt)) command, kwargs = runner.calls[0] self.assertEqual(command[0], "/usr/bin/sandbox-exec") self.assertEqual(command[-1], SYSTEM_PROMPT) self.assertEqual(kwargs["input"], '{"request":"secret-body"}') self.assertEqual(kwargs["env"].get("LANG"), "zh_CN.UTF-8") + self.assertEqual(kwargs["env"].get("ANTHROPIC_AUTH_TOKEN"), auth_token) + self.assertEqual(kwargs["env"].get("ANTHROPIC_BASE_URL"), base_url) self.assertNotIn("DATABASE_URL", kwargs["env"]) + self.assertNotIn("UNRELATED_SECRET", kwargs["env"]) self.assertNotIn("PATH", kwargs["env"]) + self.assertNotIn(auth_token, canonical_json(command)) + self.assertNotIn(base_url, canonical_json(command)) self.assertEqual(pathlib.Path(str(kwargs["cwd"])), pathlib.Path(kwargs["env"]["HOME"])) self.assertFalse(pathlib.Path(str(kwargs["cwd"])).exists()) + def test_auth_token_requires_a_safe_base_url(self): + """AUTH_TOKEN 必须绑定无凭据、无查询和无片段的 HTTP(S) 网关地址。""" + + auth_token = "auth-token-never-echoed" + invalid_base_urls = ( + None, + "", + " ", + "ftp://gateway.example.invalid/v1", + "https:///v1", + "https://user:password@gateway.example.invalid/v1", + "https://gateway.example.invalid/v1?token=leak", + "https://gateway.example.invalid/v1#fragment", + ) + for base_url in invalid_base_urls: + source_environment = {"ANTHROPIC_AUTH_TOKEN": auth_token} + if base_url is not None: + source_environment["ANTHROPIC_BASE_URL"] = base_url + with self.subTest(base_url=base_url), self.assertRaises(ClaudeRuntimeError) as raised: + run_claude( + execution_profile(), + {"request": "x"}, + binding_verifier=lambda _profile: None, + source_environment=source_environment, + ) + self.assertEqual(raised.exception.primary_code, "WRITER_RECEIPT_INVALID") + self.assertNotIn(auth_token, str(raised.exception)) + self.assertNotIn(auth_token, canonical_json(raised.exception.details)) + if base_url: + self.assertNotIn(base_url, str(raised.exception)) + self.assertNotIn(base_url, canonical_json(raised.exception.details)) + + def test_api_key_and_oauth_remain_supported_and_exclusive(self): + """API key/OAuth 仍可进入最小环境，多个认证字段仍然失败关闭。""" + + for auth_field in ("ANTHROPIC_API_KEY", "CLAUDE_CODE_OAUTH_TOKEN"): + runner = runner_for(success_envelope()) + run_claude( + execution_profile(), + {"request": "x"}, + runner=runner, + binding_verifier=lambda _profile: None, + source_environment={auth_field: "legacy-auth-secret"}, + ) + self.assertEqual(runner.calls[0][1]["env"].get(auth_field), "legacy-auth-secret") + + with self.assertRaises(ClaudeRuntimeError) as raised: + run_claude( + execution_profile(), + {"request": "x"}, + binding_verifier=lambda _profile: None, + source_environment={ + "ANTHROPIC_API_KEY": "api-secret", + "CLAUDE_CODE_OAUTH_TOKEN": "oauth-secret", + }, + ) + self.assertEqual(raised.exception.primary_code, "WRITER_RECEIPT_INVALID") + self.assertNotIn("api-secret", str(raised.exception)) + self.assertNotIn("oauth-secret", str(raised.exception)) + + def test_real_runner_requires_an_allowlisted_authentication_field(self): + """真实 subprocess 没有 API key、OAuth 或 AUTH_TOKEN 时必须阻断。""" + + with self.assertRaises(ClaudeRuntimeError) as raised: + run_claude( + execution_profile(), + {"request": "x"}, + binding_verifier=lambda _profile: None, + source_environment={"ANTHROPIC_BASE_URL": "https://gateway.example.invalid/v1"}, + ) + + self.assertEqual(raised.exception.primary_code, "WRITER_RECEIPT_INVALID") + self.assertEqual(str(raised.exception), "真实调用缺少经授权的 Claude 认证") + + def test_minimal_environment_rejects_invalid_base_url_even_without_auth_token(self): + """任何被转发的 BASE_URL 都必须是安全的 HTTP(S) 地址。""" + + with tempfile.TemporaryDirectory(dir="/private/tmp") as directory: + with self.assertRaises(ValueError): + _minimal_environment( + {"ANTHROPIC_BASE_URL": "https://user:password@gateway.example.invalid/v1"}, + pathlib.Path(directory), + require_authentication=False, + ) + def test_503_success_subtype_is_still_api_error(self): """subtype=success 不能覆盖 is_error、503、空模型 usage 或非零退出。""" + auth_token = "api-error-token-must-not-leak" + base_url = "https://gateway.example.invalid/anthropic/v1" envelope = success_envelope() envelope.update( { @@ -208,18 +311,33 @@ class ClaudeRuntimeTest(unittest.TestCase): "modelUsage": {}, } ) + runner = runner_for( + envelope, + returncode=1, + stderr=f"gateway={base_url} token={auth_token}", + ) with self.assertRaises(ClaudeRuntimeError) as raised: run_claude( execution_profile(), {"request": "x"}, - runner=runner_for(envelope, returncode=1, stderr="gateway secret"), + runner=runner, binding_verifier=lambda _profile: None, + source_environment={ + "ANTHROPIC_AUTH_TOKEN": auth_token, + "ANTHROPIC_BASE_URL": base_url, + }, ) self.assertEqual(raised.exception.primary_code, "WRITER_API_ERROR") self.assertIn("WRITER_NONZERO_EXIT", raised.exception.causes) self.assertNotIn("gateway secret", str(raised.exception)) self.assertNotIn("gateway secret", canonical_json(raised.exception.details)) + self.assertNotIn(auth_token, canonical_json(raised.exception.details)) + self.assertNotIn(base_url, canonical_json(raised.exception.details)) + self.assertNotIn(auth_token, canonical_json(raised.exception.receipt.as_dict())) + self.assertNotIn(base_url, canonical_json(raised.exception.receipt.as_dict())) + self.assertNotIn(auth_token, canonical_json(runner.calls[0][0])) + self.assertNotIn(base_url, canonical_json(runner.calls[0][0])) def test_result_never_falls_back_when_structured_output_is_missing(self): """即使 result 恰好是合法 JSON，也不得把它当业务对象。"""