修复: 支持受控 Claude 自定义网关认证

.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)

.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，也不得把它当业务对象。"""