接入 OpenAI 或 Anthropic 兼容网关时，curl 成功一次只能证明当时有一条链路返回了结果。要做一个范围受控且可复现的验收，还需要记录成功率、延迟分布、返回模型名、Token 用量、固定回答匹配率和每次请求的错误。
利益关系说明：作者参与一个 API 服务项目的运营。本文不推荐具体服务，也不把单次结果写成长期承诺；所有待测网关都应使用同一套方法自行验证。
本文实现一个只依赖 Python 标准库的小规模接口验收脚本。OpenAI / Anthropic“兼容”只描述请求和响应格式，不代表相关品牌授权、合作或背书。脚本最多发送 20 个请求、最高 8 并发，是功能检查器，不是压力测试器。
1. 先定义安全边界
脚本遵守以下约束：
固定提示词和上限为：
FIXED_PROMPT = "Reply with exactly this text and nothing else: RELAY_AUDIT_OK"
MAX_REQUESTS = 20
MAX_CONCURRENCY = 8
MAX_RESPONSE_BYTES = 256 * 1024
固定回答匹配只能检查响应链路，不能证明模型身份。接口中的 model 字段同样只是“接口报告值”，不能独立作为底层模型证明。
2. 用统一数据结构保存每次请求
每条请求保存原始判断所需的最少字段：
from dataclasses import dataclass
@dataclass
class Result:
request_id: int
ok: bool
status: int | None
latency_ms: int
reported_model: str | None
reported_model_matches_requested: bool | None
reported_model_sha256_12: str | None
input_tokens: int | None
output_tokens: int | None
exact_match: bool
content_sha256_12: str | None
error: str | None
正文输出哈希而不是完整回答，既能比较固定回答是否变化，又减少把测试内容写进日志的风险。返回模型名只有与请求值完全一致时才明文保留；不一致时只记录短哈希并令总验收失败。哈希并不等于匿名化；真实业务日志仍应按数据分级和保留周期管理。
3. HTTP 200 不等于验收通过
OpenAI 兼容端点通常使用：
POST {base_url}/chat/completions
Authorization: Bearer <API_KEY>
Anthropic 兼容端点通常使用：
POST {base_url}/messages
x-api-key: <API_KEY>
anthropic-version: 2023-06-01
两种格式的文本与 Token 字段位置不同，因此解析器应分开实现。响应不是准确的 HTTP 200、没有唯一且受支持的 application/json 或 application/*+json MIME、不是 JSON 对象、包含重复对象键或非有限数值、文本为空、模型名缺失、usage 缺失，或者 Token 超出非负 signed 64-bit 整数范围时，都应判为失败。若 Content-Type 声明 charset，只接受 UTF-8。OpenAI 响应还必须是 assistant 文本消息并以 finish_reason=stop 结束；Anthropic 响应必须是 assistant message、以 stop_reason=end_turn 结束，并拒绝混入非文本块。
OpenAI 格式的严格解析核心如下：
def _token_count(value, field):
if (
isinstance(value, bool)
or not isinstance(value, int)
or not 0 <= value <= (1 << 63) – 1
):
raise ValueError(f"{field} must be a non-negative signed 64-bit integer")
return value
def _extract_openai(data):
choices = data.get("choices")
if (
not isinstance(choices, list)
or len(choices) != 1
or not isinstance(choices[0], dict)
):
raise ValueError("OpenAI response must contain exactly one valid choice")
message = choices[0].get("message")
if not isinstance(message, dict) or not isinstance(message.get("content"), str):
raise ValueError("OpenAI response message content is missing or invalid")
if message.get("role") != "assistant":
raise ValueError("OpenAI response message role must be assistant")
if choices[0].get("finish_reason") != "stop":
raise ValueError("OpenAI response finish_reason must be stop")
usage = data.get("usage")
if not isinstance(usage, dict):
raise ValueError("OpenAI response usage is missing or invalid")
model = data.get("model")
if (
not isinstance(model, str)
or not model.isascii()
or not 1 <= len(model) <= 200
or any(not 0x21 <= ord(character) <= 0x7E for character in model)
):
raise ValueError("OpenAI response model is missing or invalid")
return (
message["content"],
model,
_token_count(usage.get("prompt_tokens"), "prompt_tokens"),
_token_count(usage.get("completion_tokens"), "completion_tokens"),
)
Anthropic 格式需要拼接 content 中类型为 text 的文本块，并严格读取 input_tokens、output_tokens。完整实现见文末，前面的短代码块只是讲解片段，不应彼此直接拼接执行。
4. HTTPS 和拒绝重定向是密钥边界
Python 默认的 URL opener 会处理常见重定向。对携带 Authorization 或 x-api-key 的 POST 请求，验收工具更稳妥的做法是要求调用者直接提供最终 HTTPS 端点，并拒绝所有重定向：
class _RejectRedirects(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
return None
_NO_REDIRECT_OPENER = urllib.request.build_opener(_RejectRedirects())
with _NO_REDIRECT_OPENER.open(request, timeout=timeout) as response:
status = response.status
raw = response.read(MAX_RESPONSE_BYTES + 1)
if len(raw) > MAX_RESPONSE_BYTES:
raise ValueError("response body exceeds the 256 KiB safety limit")
Base URL 校验还应拒绝明文 HTTP、URL 中的用户名或密码、查询串、fragment、歧义端口、点路径、编码分隔符、旧式数字主机、伪括号主机、单标签主机、反斜杠和 RFC 3986 路径之外的字符，并输出规范化结果。不要把 API Key 放进 URL 路径；脚本会同时检查原始 URL、一次 URL 解码后的原始 URL、规范化 URL及其他待输出参数，防止规范化过程掩盖密钥。
5. 正确认识延迟和 timeout
使用单调时钟记录从发起请求到完整读取响应的耗时：
started = time.perf_counter()
with _NO_REDIRECT_OPENER.open(request, timeout=timeout) as response:
status = response.status
data = _safe_json_loads(_read_bounded(response))
latency_ms = round((time.perf_counter() – started) * 1000)
这里记录的是完成请求的耗时，不是流式接口的首 Token 延迟。标准库 urlopen 的 timeout 是 socket 无数据活动超时，不是强制的总墙钟截止时间：持续缓慢返回数据的端点可能让总耗时超过该值。因此不要对不可信端点运行；若业务需要严格总截止时间，应在独立进程或外层任务系统中另设强制时限。
6. 小并发要有硬限制
脚本使用 ThreadPoolExecutor，但参数校验会拒绝超过上限的输入：
if not 1 <= args.requests <= MAX_REQUESTS:
parser.error(f"–requests must be between 1 and {MAX_REQUESTS}")
if not 1 <= args.concurrency <= min(MAX_CONCURRENCY, args.requests):
parser.error(
f"–concurrency must be between 1 and min({MAX_CONCURRENCY}, requests)"
)
建议先单并发 3 次，再视服务条款、配额和测试窗口决定是否运行 2 并发 5 次。本文脚本不适合做生产压力测试。
7. 汇总时保留时间与环境
聚合结果至少包括：
- UTC 起止时间和非敏感环境标签；
- 成功数与成功率；
- 固定回答逐字匹配率；
- 返回模型与请求模型的匹配率，以及总体验收布尔值；
- 成功请求的最小、中位、最大延迟；
- 接口报告过的模型名集合；
- 每条请求的状态、耗时、Token、内容哈希和错误类型。
结构示例：
{
"run_started_at_utc": "2026-08-24T00:00:00Z",
"run_completed_at_utc": "2026-08-24T00:00:04Z",
"environment_label": "office-wifi",
"target": {
"base_url": "https://gateway.example/v1",
"mode": "openai",
"requested_model": "example-model"
},
"requests": 3,
"successful": 3,
"audit_passed": true,
"success_rate": 1.0,
"exact_match_rate": 1.0,
"model_match_rate": 1.0,
"latency_ms": {
"min": 812,
"median": 934,
"max": 1280
},
"reported_models": ["example-model"]
}
.example 是保留示例域名；以上数字只展示输出结构，不是任何服务的实测结果。success_rate 只表示 HTTP 200 与 schema 通过率，最终应看同时包含固定回答和模型匹配条件的 audit_passed。完整输出还会包含逐条 results 和安全提示。环境标签只写诸如“office-wifi”或“cloud-test”，不要写用户名、IP、客户名或内部项目名。
8. 运行方式：不要把真实密钥写进历史
在正常交互式终端中，下面通过 Python 隐藏输入，并只为本次命令临时注入环境变量。密钥不会以明文出现在 shell 历史中，但在命令运行期间仍存在于进程环境；不要在不可信共享主机上执行。
OpenAI 兼容格式：
RELAY_AUDIT_API_KEY="$(
python3 -c 'import getpass; print(getpass.getpass("Test API key: "))'
)" python3 relay_audit.py \\
–base-url 'https://gateway.example/v1' \\
–mode openai \\
–model '服务方文档中的准确模型名' \\
–requests 3 \\
–concurrency 1 \\
–timeout 30 \\
–environment-label 'office-wifi'
Anthropic 格式只需将 –mode openai 改为 –mode anthropic，并填写该服务文档中的最终 HTTPS Base URL 和准确模型名。脚本不会跟随重定向。
退出码含义：全部请求通过传输和 schema 检查、逐字返回 RELAY_AUDIT_OK，并且返回模型与请求模型一致时为 0；至少一条失败、回答不精确或模型不一致时为 1；参数或 API Key 绑定失败时为 2。
运行结束后不要只看本地 JSON，还要到服务方控制台核对同一 UTC 时间段的模型、Token、请求状态和扣费。存在无法解释的差异时，先停止扩大请求量。结果文件使用受限目录保存，完成复盘后按既定周期删除。
9. 完整脚本
下面代码与离线测试使用的版本保持一致。请使用这一完整代码块，不要把前文讲解片段再次拼接进去。
#!/usr/bin/env python3
"""Small, bounded audit client for OpenAI- or Anthropic-compatible relays.
The tool intentionally caps request count and concurrency. It is a functional
check, not a load-testing utility. API keys are read from an environment
variable and are never printed or written to disk.
"""
from __future__ import annotations
import argparse
import concurrent.futures
import datetime
import hashlib
import ipaddress
import json
import math
import os
import re
import statistics
import sys
import time
import unicodedata
import urllib.error
import urllib.parse
import urllib.request
from collections.abc import Callable
from dataclasses import asdict, dataclass
from typing import Any, NoReturn, TypeVar
FIXED_PROMPT = "Reply with exactly this text and nothing else: RELAY_AUDIT_OK"
MAX_REQUESTS = 20
MAX_CONCURRENCY = 8
MAX_RESPONSE_BYTES = 256 * 1024
MAX_TOKEN_COUNT = (1 << 63) – 1
T = TypeVar("T")
_JSON_CONTENT_TYPE_RE = re.compile(
r"^application/(?:json|[!#$%&'*+\\-.^_`|~0-9A-Za-z]+\\+json)"
r'(?:\\s*;\\s*charset\\s*=\\s*(?:"(?:utf-8|utf8)"|(?:utf-8|utf8)))?$',
re.IGNORECASE,
)
class _RejectRedirects(urllib.request.HTTPRedirectHandler):
"""Fail closed before credentials can be copied to a redirect target."""
def redirect_request(
self,
req: urllib.request.Request,
fp: Any,
code: int,
msg: str,
headers: Any,
newurl: str,
) –> None:
return None
_NO_REDIRECT_OPENER = urllib.request.build_opener(_RejectRedirects())
@dataclass
class Result:
request_id: int
ok: bool
status: int | None
latency_ms: int
reported_model: str | None
reported_model_matches_requested: bool | None
reported_model_sha256_12: str | None
input_tokens: int | None
output_tokens: int | None
exact_match: bool
content_sha256_12: str | None
error: str | None
@dataclass(frozen=True)
class _ValidatedBaseURL:
raw: str
canonical: str
def _reject_duplicate_keys(pairs: list[tuple[str, Any]]) –> dict[str, Any]:
value: dict[str, Any] = {}
for key, item in pairs:
if key in value:
raise ValueError("response JSON contains a duplicate object key")
value[key] = item
return value
def _reject_nonfinite_constant(_value: str) –> NoReturn:
raise ValueError("response JSON contains a non-finite numeric constant")
def _parse_finite_float(value: str) –> float:
parsed = float(value)
if not math.isfinite(parsed):
raise ValueError("response JSON contains an out-of-range floating-point value")
return parsed
def _validate_json_unicode(value: Any) –> None:
if isinstance(value, str):
try:
value.encode("utf-8")
except UnicodeEncodeError as exc:
raise ValueError("response JSON contains an invalid Unicode scalar") from exc
elif isinstance(value, list):
for item in value:
_validate_json_unicode(item)
elif isinstance(value, dict):
for key, item in value.items():
_validate_json_unicode(key)
_validate_json_unicode(item)
def _safe_json_loads(raw: bytes) –> dict[str, Any]:
value = json.loads(
raw.decode("utf-8"),
object_pairs_hook=_reject_duplicate_keys,
parse_constant=_reject_nonfinite_constant,
parse_float=_parse_finite_float,
)
if not isinstance(value, dict):
raise ValueError("response JSON is not an object")
_validate_json_unicode(value)
return value
def _validate_base_url(value: str) –> str:
"""Accept only a final HTTPS endpoint without loggable URL secrets."""
if (
not value
or len(value) > 2048
or not value.isascii()
or value != value.strip()
or "\\\\" in value
or any(not 0x21 <= ord(character) <= 0x7E for character in value)
):
raise ValueError("base URL must be 1-2048 visible ASCII characters")
try:
parsed = urllib.parse.urlsplit(value)
hostname = parsed.hostname
port = parsed.port
except ValueError as exc:
raise ValueError("base URL is malformed") from exc
if parsed.scheme != "https" or not hostname:
raise ValueError("base URL must use HTTPS and include a hostname")
if parsed.username is not None or parsed.password is not None:
raise ValueError("base URL must not contain user information")
if "?" in value or "#" in value:
raise ValueError("base URL must not contain a query string or fragment")
if parsed.netloc.endswith(":") or port is not None and not 1 <= port <= 65535:
raise ValueError("base URL port is malformed or out of range")
if hostname.endswith(".") or "%" in hostname:
raise ValueError("base URL hostname is ambiguous")
bracketed_host = "[" in parsed.netloc or "]" in parsed.netloc
port_text: str | None = None
if bracketed_host:
closing_bracket = parsed.netloc.rfind("]")
if closing_bracket >= 0 and parsed.netloc[closing_bracket + 1 :]:
port_text = parsed.netloc[closing_bracket + 2 :]
elif ":" in parsed.netloc:
port_text = parsed.netloc.rsplit(":", 1)[1]
if port_text is not None and (
not port_text.isdecimal() or str(int(port_text)) != port_text
):
raise ValueError("base URL port must use canonical decimal notation")
try:
address = ipaddress.ip_address(hostname)
except ValueError:
if bracketed_host:
raise ValueError("bracketed base URL host must be a valid IPv6 address")
labels = hostname.split(".")
legacy_numeric_label = r"(?:0[xX][0-9A-Fa-f]+|0[0-7]+|[0-9]+)"
if (
len(hostname) > 253
or len(labels) < 2
or all(re.fullmatch(legacy_numeric_label, label) for label in labels)
or any(
not label
or len(label) > 63
or not re.fullmatch(r"[A-Za-z0-9-]+", label)
or label.startswith("-")
or label.endswith("-")
for label in labels
)
):
raise ValueError("base URL hostname is malformed")
canonical_host = hostname.lower()
else:
if bracketed_host and address.version != 6:
raise ValueError("bracketed base URL host must be a valid IPv6 address")
canonical_host = address.compressed
if address.version == 6:
canonical_host = f"[{canonical_host}]"
path = parsed.path
if re.search(r"%(?![0-9A-Fa-f]{2})", path):
raise ValueError("base URL path contains malformed percent encoding")
if not re.fullmatch(r"[A-Za-z0-9._~!$&'()*+,;=:@/%-]*", path):
raise ValueError("base URL path contains a character outside RFC 3986 pchar")
try:
decoded_path = urllib.parse.unquote_to_bytes(path).decode("utf-8")
except UnicodeDecodeError as exc:
raise ValueError("base URL path contains invalid UTF-8 encoding") from exc
if (
"%2f" in path.lower()
or "%5c" in path.lower()
or "%2e" in path.lower()
or "%25" in path.lower()
or "//" in path
or "\\\\" in decoded_path
or (decoded_path and not decoded_path.isprintable())
or unicodedata.normalize("NFKC", decoded_path) != decoded_path
or any(segment in (".", "..") for segment in decoded_path.split("/"))
):
raise ValueError("base URL path contains ambiguous separators or dot segments")
canonical_path = re.sub(
r"%[0-9A-Fa-f]{2}", lambda match: match.group(0).upper(), path
).rstrip("/")
canonical_netloc = canonical_host
if port is not None and port != 443:
canonical_netloc = f"{canonical_host}:{port}"
return urllib.parse.urlunsplit(
("https", canonical_netloc, canonical_path, "", "")
)
def _base_url_argument(value: str) –> _ValidatedBaseURL:
return _ValidatedBaseURL(raw=value, canonical=_validate_base_url(value))
def _validate_api_key(value: str) –> str:
if (
not value
or not 8 <= len(value) <= 4096
or re.fullmatch(r"[+-]?[0-9]+", value)
or value != value.strip()
or any(not 0x21 <= ord(character) <= 0x7E for character in value)
):
raise ValueError(
"API key must be 8-4096 visible ASCII characters and not numeric-like"
)
return value
def _read_bounded(response: Any) –> bytes:
raw = response.read(MAX_RESPONSE_BYTES + 1)
if not isinstance(raw, bytes):
raise ValueError("response body is not bytes")
if len(raw) > MAX_RESPONSE_BYTES:
raise ValueError("response body exceeds the 256 KiB safety limit")
return raw
def _token_count(value: Any, field: str) –> int:
if (
isinstance(value, bool)
or not isinstance(value, int)
or not 0 <= value <= MAX_TOKEN_COUNT
):
raise ValueError(f"{field} must be a non-negative signed 64-bit integer")
return int(value)
def _environment_label(value: str) –> str:
if (
not value
or len(value) > 64
or value != value.strip()
or not value.isprintable()
or unicodedata.normalize("NFKC", value) != value
):
raise ValueError("environment label must be 1-64 printable characters")
return value
def _model_name(value: str) –> str:
if (
not value
or len(value) > 200
or not value.isascii()
or value != value.strip()
or any(not 0x21 <= ord(character) <= 0x7E for character in value)
):
raise ValueError("model identifier must be 1-200 visible ASCII characters")
return value
def _api_key_env_name(value: str) –> str:
if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]{0,63}", value):
raise ValueError("API key environment variable must be a valid shell name")
return value
def _mode_name(value: str) –> str:
if value not in ("openai", "anthropic"):
raise ValueError("mode must be openai or anthropic")
return value
def _integer_argument(value: str) –> int:
try:
return int(value)
except ValueError as exc:
raise ValueError("value must be an integer") from exc
def _float_argument(value: str) –> float:
try:
return float(value)
except ValueError as exc:
raise ValueError("value must be a number") from exc
def _safe_argument_type(validator: Callable[[str], T]) –> Callable[[str], T]:
def parse(value: str) –> T:
try:
return validator(value)
except ValueError as exc:
raise argparse.ArgumentTypeError(str(exc)) from None
return parse
class _SafeArgumentParser(argparse.ArgumentParser):
def error(self, message: str) –> NoReturn:
if "ignored explicit argument" in message:
message = "help option does not accept a value"
elif message.startswith(("unrecognized arguments:", "ambiguous option:")):
message = "unrecognized command-line arguments (values redacted)"
super().error(message)
def _utc_text() –> str:
return (
datetime.datetime.now(datetime.timezone.utc)
.isoformat()
.replace("+00:00", "Z")
)
def _redact_secret_values(value: Any, secret: str) –> Any:
if isinstance(value, str):
return value.replace(secret, "[REDACTED]")
if isinstance(value, list):
return [_redact_secret_values(item, secret) for item in value]
if isinstance(value, dict):
return {
key: _redact_secret_values(item, secret) for key, item in value.items()
}
if isinstance(value, (int, float)) and not isinstance(value, bool):
return "[REDACTED]" if secret in str(value) else value
return value
def _extract_openai(data: dict[str, Any]) –> tuple[str, str, int, int]:
choices = data.get("choices")
if (
not isinstance(choices, list)
or len(choices) != 1
or not isinstance(choices[0], dict)
):
raise ValueError("OpenAI response must contain exactly one valid choice")
message = choices[0].get("message")
if not isinstance(message, dict) or not isinstance(message.get("content"), str):
raise ValueError("OpenAI response message content is missing or invalid")
if message.get("role") != "assistant":
raise ValueError("OpenAI response message role must be assistant")
allowed_empty_payloads = {
"tool_calls": (None, []),
"function_call": (None, {}),
"refusal": (None, ""),
"audio": (None, {}),
}
for field, allowed in allowed_empty_payloads.items():
if field in message and message[field] not in allowed:
raise ValueError("OpenAI response contains a non-text message payload")
if choices[0].get("finish_reason") != "stop":
raise ValueError("OpenAI response finish_reason must be stop")
usage = data.get("usage")
if not isinstance(usage, dict):
raise ValueError("OpenAI response usage is missing or invalid")
model = data.get("model")
if not isinstance(model, str):
raise ValueError("OpenAI response model is missing or invalid")
return (
message["content"],
_model_name(model),
_token_count(usage.get("prompt_tokens"), "prompt_tokens"),
_token_count(usage.get("completion_tokens"), "completion_tokens"),
)
def _extract_anthropic(data: dict[str, Any]) –> tuple[str, str, int, int]:
if data.get("type") != "message" or data.get("role") != "assistant":
raise ValueError("Anthropic response must be an assistant message")
if data.get("stop_reason") != "end_turn":
raise ValueError("Anthropic response stop_reason must be end_turn")
blocks = data.get("content")
if not isinstance(blocks, list) or not blocks:
raise ValueError("Anthropic response content is missing or invalid")
texts = []
for block in blocks:
if (
not isinstance(block, dict)
or block.get("type") != "text"
or not isinstance(block.get("text"), str)
):
raise ValueError("Anthropic response contains an invalid or non-text block")
texts.append(block["text"])
usage = data.get("usage")
if not isinstance(usage, dict):
raise ValueError("Anthropic response usage is missing or invalid")
model = data.get("model")
if not isinstance(model, str):
raise ValueError("Anthropic response model is missing or invalid")
return (
"".join(texts),
_model_name(model),
_token_count(usage.get("input_tokens"), "input_tokens"),
_token_count(usage.get("output_tokens"), "output_tokens"),
)
def _request_once(
request_id: int,
*,
base_url: str,
api_key: str,
model: str,
mode: str,
timeout: float,
) –> Result:
started = time.perf_counter()
status: int | None = None
try:
raw_base_url = base_url
base_url = _validate_base_url(base_url)
api_key = _validate_api_key(api_key)
model = _model_name(model)
if (
api_key in raw_base_url
or api_key in urllib.parse.unquote(raw_base_url)
or api_key in base_url
or api_key in urllib.parse.unquote(base_url)
or api_key in model
):
raise ValueError("API key appears in an output-bound argument")
if mode == "openai":
url = f"{base_url}/chat/completions"
headers = {"Authorization": f"Bearer {api_key}"}
payload = {
"model": model,
"messages": [{"role": "user", "content": FIXED_PROMPT}],
"temperature": 0,
"max_tokens": 32,
"stream": False,
}
extractor = _extract_openai
elif mode == "anthropic":
url = f"{base_url}/messages"
headers = {
"x-api-key": api_key,
"anthropic-version": "2023-06-01",
}
payload = {
"model": model,
"messages": [{"role": "user", "content": FIXED_PROMPT}],
"temperature": 0,
"max_tokens": 32,
"stream": False,
}
extractor = _extract_anthropic
else:
raise ValueError("mode must be openai or anthropic")
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
headers.update(
{
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": "relay-audit/0.1.0-local",
}
)
request = urllib.request.Request(url, data=body, headers=headers, method="POST")
with _NO_REDIRECT_OPENER.open(request, timeout=timeout) as response:
status = response.status
if isinstance(status, bool) or not isinstance(status, int):
raise ValueError("response status is not an integer")
if status != 200:
raise ValueError("synchronous completion response must use HTTP 200")
response_headers = getattr(response, "headers", None)
get_all = getattr(response_headers, "get_all", None)
if callable(get_all):
content_type_values = get_all("Content-Type")
content_type = (
content_type_values[0]
if isinstance(content_type_values, list)
and len(content_type_values) == 1
else None
)
else:
content_type = (
response_headers.get("Content-Type")
if response_headers is not None
else None
)
if (
not isinstance(content_type, str)
or not content_type.isascii()
or any(not 0x20 <= ord(character) <= 0x7E for character in content_type)
):
raise ValueError("response Content-Type is missing or invalid")
if _JSON_CONTENT_TYPE_RE.fullmatch(content_type.strip()) is None:
raise ValueError("response Content-Type must be application JSON")
data = _safe_json_loads(_read_bounded(response))
content, reported_model, input_tokens, output_tokens = extractor(data)
if (
api_key in content
or api_key in reported_model
or api_key in str(input_tokens)
or api_key in str(output_tokens)
):
raise ValueError("response contained protected credential data and was redacted")
if not content.strip():
raise ValueError("response content is empty")
reported_model_matches_requested = reported_model == model
return Result(
request_id=request_id,
ok=True,
status=status,
latency_ms=round((time.perf_counter() – started) * 1000),
reported_model=(
reported_model if reported_model_matches_requested else None
),
reported_model_matches_requested=reported_model_matches_requested,
reported_model_sha256_12=hashlib.sha256(
reported_model.encode("utf-8")
).hexdigest()[:12],
input_tokens=input_tokens,
output_tokens=output_tokens,
exact_match=content == "RELAY_AUDIT_OK",
content_sha256_12=hashlib.sha256(content.encode("utf-8")).hexdigest()[:12],
error=None,
)
except urllib.error.HTTPError as exc:
try:
status = exc.code
suffix = " (redirects are refused)" if 300 <= exc.code < 400 else ""
error = f"HTTP {exc.code}{suffix}"
finally:
try:
exc.close()
except Exception:
pass
except urllib.error.URLError as exc:
error = f"network error: {type(exc.reason).__name__}"
except (TimeoutError, UnicodeDecodeError, ValueError) as exc:
safe_message = str(exc).replace(api_key, "[REDACTED]") if api_key else str(exc)
error = f"response error: {safe_message}"
except Exception as exc: # Keep the audit summary useful without leaking secrets.
error = f"unexpected error: {type(exc).__name__}"
return Result(
request_id=request_id,
ok=False,
status=status,
latency_ms=round((time.perf_counter() – started) * 1000),
reported_model=None,
reported_model_matches_requested=None,
reported_model_sha256_12=None,
input_tokens=None,
output_tokens=None,
exact_match=False,
content_sha256_12=None,
error=error,
)
def _is_hash12(value: Any) –> bool:
return isinstance(value, str) and re.fullmatch(r"[0-9a-f]{12}", value) is not None
def _result_has_valid_success_schema(result: Result, requested_model: str) –> bool:
if (
result.ok is not True
or isinstance(result.status, bool)
or not isinstance(result.status, int)
or result.status != 200
or isinstance(result.request_id, bool)
or not isinstance(result.request_id, int)
or result.request_id < 1
or isinstance(result.latency_ms, bool)
or not isinstance(result.latency_ms, int)
or result.latency_ms < 0
or result.error is not None
or not isinstance(result.exact_match, bool)
or not _is_hash12(result.content_sha256_12)
or not isinstance(result.reported_model_matches_requested, bool)
or not _is_hash12(result.reported_model_sha256_12)
):
return False
try:
_token_count(result.input_tokens, "input_tokens")
_token_count(result.output_tokens, "output_tokens")
except ValueError:
return False
if result.exact_match and result.content_sha256_12 != hashlib.sha256(
b"RELAY_AUDIT_OK"
).hexdigest()[:12]:
return False
if result.reported_model_matches_requested:
return (
result.reported_model == requested_model
and result.reported_model_sha256_12
== hashlib.sha256(requested_model.encode("utf-8")).hexdigest()[:12]
)
return result.reported_model is None
def _build_summary(
results: list[Result],
*,
base_url: str,
model: str,
mode: str,
environment_label: str | None = None,
run_started_at_utc: str | None = None,
run_completed_at_utc: str | None = None,
) –> dict[str, Any]:
successful = [
result
for result in results
if _result_has_valid_success_schema(result, model)
]
latencies = [result.latency_ms for result in successful]
models = sorted({result.reported_model for result in successful if result.reported_model})
request_ids_are_complete = [result.request_id for result in results] == list(
range(1, len(results) + 1)
)
denominator = len(results) or 1
return {
"run_started_at_utc": run_started_at_utc or _utc_text(),
"run_completed_at_utc": run_completed_at_utc or _utc_text(),
"environment_label": environment_label,
"target": {"base_url": base_url, "mode": mode, "requested_model": model},
"requests": len(results),
"successful": len(successful),
"audit_passed": all(
_result_has_valid_success_schema(result, model)
and result.exact_match
and result.reported_model_matches_requested is True
for result in results
)
and bool(results)
and request_ids_are_complete,
"success_rate": round(len(successful) / denominator, 4),
"exact_match_rate": round(
sum(result.exact_match for result in successful) / denominator, 4
),
"model_match_rate": round(
sum(
result.reported_model_matches_requested is True
for result in successful
)
/ denominator,
4,
),
"latency_ms": {
"min": min(latencies) if latencies else None,
"median": round(statistics.median(latencies)) if latencies else None,
"max": max(latencies) if latencies else None,
},
"reported_models": models,
"results": [asdict(result) for result in results],
"notes": [
"This is a small functional check, not a model-identity proof.",
"Compare the provider dashboard's usage and billing records separately.",
"Do not test with secrets, personal data, or production workloads.",
"The timeout is a socket inactivity limit, not a hard wall-clock deadline.",
"The result contains endpoint and model identifiers; restrict access and retention.",
"Mismatched models are stored only as short hashes; credential echoes are redacted.",
],
}
def parse_args(argv: list[str] | None = None) –> argparse.Namespace:
parser = _SafeArgumentParser(
prog="relay_audit.py",
description="Run a small, reproducible API relay check.",
allow_abbrev=False,
)
parser.add_argument(
"–base-url",
required=True,
type=_safe_argument_type(_base_url_argument),
help="Final HTTPS API base URL, usually ending in /v1; redirects are refused.",
)
parser.add_argument(
"–model",
required=True,
type=_safe_argument_type(_model_name),
help="Exact model identifier shown by the provider (max 200 chars).",
)
parser.add_argument(
"–mode", type=_safe_argument_type(_mode_name), default="openai"
)
parser.add_argument(
"–requests", type=_safe_argument_type(_integer_argument), default=3
)
parser.add_argument(
"–concurrency", type=_safe_argument_type(_integer_argument), default=1
)
parser.add_argument(
"–timeout",
type=_safe_argument_type(_float_argument),
default=30.0,
help="Socket inactivity timeout in seconds; not a hard wall-clock deadline.",
)
parser.add_argument(
"–api-key-env",
type=_safe_argument_type(_api_key_env_name),
default="RELAY_AUDIT_API_KEY",
help="Environment variable containing the test key.",
)
parser.add_argument(
"–environment-label",
type=_safe_argument_type(_environment_label),
help="Optional non-sensitive network/test environment label (max 64 chars).",
)
args = parser.parse_args(argv)
validated_base_url = args.base_url
args.base_url_raw = validated_base_url.raw
args.base_url = validated_base_url.canonical
if not 1 <= args.requests <= MAX_REQUESTS:
parser.error(f"–requests must be between 1 and {MAX_REQUESTS}")
if not 1 <= args.concurrency <= min(MAX_CONCURRENCY, args.requests):
parser.error(f"–concurrency must be between 1 and min({MAX_CONCURRENCY}, requests)")
if not 1 <= args.timeout <= 120:
parser.error("–timeout must be between 1 and 120 seconds")
return args
def main(argv: list[str] | None = None) –> int:
args = parse_args(argv)
api_key = os.environ.get(args.api_key_env)
if not api_key:
print("Missing API key environment variable.", file=sys.stderr)
return 2
try:
api_key = _validate_api_key(api_key)
except ValueError as exc:
print(f"Invalid API key: {exc}", file=sys.stderr)
return 2
output_bound_values = (
args.base_url_raw,
urllib.parse.unquote(args.base_url_raw),
args.base_url,
urllib.parse.unquote(args.base_url),
args.model,
args.environment_label or "",
)
if any(api_key in value for value in output_bound_values):
print(
"Refusing to run: the API key also appears in an output-bound argument.",
file=sys.stderr,
)
return 2
run_started_at_utc = _utc_text()
with concurrent.futures.ThreadPoolExecutor(max_workers=args.concurrency) as executor:
futures = [
executor.submit(
_request_once,
request_id,
base_url=args.base_url,
api_key=api_key,
model=args.model,
mode=args.mode,
timeout=args.timeout,
)
for request_id in range(1, args.requests + 1)
]
results = [future.result() for future in futures]
results.sort(key=lambda result: result.request_id)
summary = _build_summary(
results,
base_url=args.base_url,
model=args.model,
mode=args.mode,
environment_label=args.environment_label,
run_started_at_utc=run_started_at_utc,
run_completed_at_utc=_utc_text(),
)
exit_code = 0 if summary["audit_passed"] else 1
summary = _redact_secret_values(summary, api_key)
serialized = json.dumps(summary, ensure_ascii=False, indent=2)
escaped_api_key = json.dumps(api_key, ensure_ascii=False)[1:–1]
if api_key in serialized or escaped_api_key in serialized:
print("null")
return 1
print(serialized)
return exit_code
if __name__ == "__main__":
raise SystemExit(main())
把脚本放进自己的 tools/ 或验收目录，先运行离线单元测试，再使用非敏感测试账户做小规模检查。是否接入应以自己的验收记录为准。
