Настоящий ли Claude отдаёт ваш API-прокси? Как это проверить

Коротко

Это руководство публикует APIVAI — шлюз к моделям Claude и GPT. Тесты работают с любым прокси, поддерживающим нативный формат Anthropic, в том числе с нашим; наши результаты — ближе к концу.

Почему подмены моделей опасаются

Проверки, которые ничего не доказывают

Как работает тест подписи thinking

400 invalid_request_error: messages.1.content.0: Invalid `signature` in `thinking` block

Однократная проверка

import os, requests

BASE = os.environ["RELAY_BASE_URL"]      # e.g. https://your-relay.example (no /v1)
KEY = os.environ["RELAY_API_KEY"]
MODEL = os.environ.get("MODEL", "claude-sonnet-4-6")

H = {"x-api-key": KEY, "anthropic-version": "2023-06-01", "content-type": "application/json"}
TOOLS = [{"name": "get_weather", "description": "Get the weather for a city",
          "input_schema": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}}]
THINK = {"type": "enabled", "budget_tokens": 1024}
Q = {"role": "user", "content": "What is the weather in Paris? Use the tool."}

def call(messages):
    r = requests.post(f"{BASE}/v1/messages", headers=H, timeout=120, json={
        "model": MODEL, "max_tokens": 1600, "thinking": THINK, "tools": TOOLS, "messages": messages})
    return r.status_code, r.json()

s, a = call([Q])
blocks = a.get("content", [])
print("1st:", s, a.get("id"), [b["type"] for b in blocks])
print("usage fields:", sorted(a.get("usage", {}).keys()))
tool = next((b for b in blocks if b["type"] == "tool_use"), None)
think = next((b for b in blocks if b["type"] == "thinking"), None)
if not (tool and think and think.get("signature")):
    raise SystemExit("No thinking block with a signature: cannot run the test (try another model or prompt).")

def second(content):
    return call([Q, {"role": "assistant", "content": content},
                 {"role": "user", "content": [{"type": "tool_result", "tool_use_id": tool["id"], "content": "Sunny, 18C"}]}])

s2, b2 = second(blocks)
print("2nd, signature unchanged:", s2, "(expect 200)")

bad = [dict(b, signature=b["signature"][:-8] + "AAAAAAAA") if b["type"] == "thinking" else b for b in blocks]
s3, b3 = second(bad)
print("3rd, signature tampered:", s3, b3.get("error", {}).get("message", ""), "(expect 400 Invalid `signature`)")

Наблюдение за прокси во времени

import json, os, re, time
import requests

BASE = os.environ["RELAY_BASE_URL"].rstrip("/")
KEY = os.environ["RELAY_API_KEY"]
MODELS = [m.strip() for m in os.environ.get("MODELS", "claude-haiku-4-5").split(",") if m.strip()]
LOG = os.environ.get("LOG_FILE", "relay_watch.jsonl")

HEADERS = {"x-api-key": KEY, "anthropic-version": "2023-06-01", "content-type": "application/json"}
TOOLS = [{"name": "get_weather", "description": "Get the weather for a city",
          "input_schema": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}}]
THINKING = {"type": "enabled", "budget_tokens": 1024}
QUESTION = {"role": "user", "content": "What is the weather in Paris? Use the tool."}


def post(model, messages):
    t = time.time()
    r = requests.post(f"{BASE}/v1/messages", headers=HEADERS, timeout=180, json={
        "model": model, "max_tokens": 1600, "thinking": THINKING, "tools": TOOLS, "messages": messages})
    try:
        body = r.json()
    except ValueError:
        body = {}
    return r.status_code, body, int((time.time() - t) * 1000)


def check(model):
    res = {"ts": time.strftime("%Y-%m-%dT%H:%M:%S"), "model": model}
    status, a, ms = post(model, [QUESTION])
    res.update(status=status, latency_ms=ms)
    if status != 200:
        res["verdict"] = f"error {status}: {a.get('error', {}).get('message', '')[:120]}"
        return res
    blocks = a.get("content", [])
    usage = a.get("usage", {})
    tool = next((b for b in blocks if b.get("type") == "tool_use"), None)
    think = next((b for b in blocks if b.get("type") == "thinking"), None)
    res.update(
        id_format=bool(re.match(r"^msg_[0-9A-Za-z]{20,}$", a.get("id", ""))),
        tool_id_format=bool(tool and tool.get("id", "").startswith("toolu_")),
        usage_fields=all(k in usage for k in ("service_tier", "cache_creation")),
        # Один и тот же запрос к одной и той же модели всегда даёт одно и то же число входных токенов.
        # Другое число — другой токенизатор (другая модель) или добавленный системный промпт.
        input_tokens=usage.get("input_tokens"),
    )
    if not (tool and think and think.get("signature")):
        res["verdict"] = "no signed thinking block: signature test not possible"
        return res

    def second(content):
        return post(model, [QUESTION, {"role": "assistant", "content": content},
                            {"role": "user", "content": [{"type": "tool_result", "tool_use_id": tool["id"], "content": "Sunny, 18C"}]}])

    s2, _, _ = second(blocks)
    bad = [dict(b, signature=b["signature"][:-8] + "AAAAAAAA") if b.get("type") == "thinking" else b for b in blocks]
    s3, b3, _ = second(bad)
    rejected = s3 == 400 and "signature" in json.dumps(b3).lower()
    res.update(roundtrip=s2, tampered=s3, tamper_rejected=rejected)
    if s2 == 200 and rejected:
        res["verdict"] = "ok"
    elif s2 == 200 and s3 == 200:
        res["verdict"] = "SIGNATURES NOT CHECKED"
    else:
        res["verdict"] = f"thinking blocks broken ({s2}/{s3})"
    return res


def last_runs():
    prev = {}
    if os.path.exists(LOG):
        with open(LOG, encoding="utf-8") as f:
            for line in f:
                try:
                    r = json.loads(line)
                    prev[r["model"]] = r
                except (ValueError, KeyError):
                    pass
    return prev


def differences(old, new):
    if not old:
        return []
    out = []
    for k in ("verdict", "input_tokens", "id_format", "tool_id_format", "usage_fields"):
        if old.get(k) != new.get(k):
            out.append(f"{k}: {old.get(k)} -> {new.get(k)}")
    if old.get("latency_ms") and new.get("latency_ms") and new["latency_ms"] > 3 * old["latency_ms"]:
        out.append(f"latency_ms: {old['latency_ms']} -> {new['latency_ms']}")
    return out


def notify(text):
    token, chat = os.environ.get("TG_TOKEN"), os.environ.get("TG_CHAT")
    if token and chat:
        requests.post(f"https://api.telegram.org/bot{token}/sendMessage", timeout=30, data={"chat_id": chat, "text": text})


if __name__ == "__main__":
    prev = last_runs()
    with open(LOG, "a", encoding="utf-8") as f:
        for m in MODELS:
            r = check(m)
            f.write(json.dumps(r, ensure_ascii=False) + "\n")
            diff = differences(prev.get(m), r)
            print(m, r["verdict"], r.get("input_tokens"), f"{r['latency_ms']} ms", "; ".join(diff))
            if diff or r["verdict"] != "ok":
                notify(f"relay_watch {m}: {r['verdict']}\n" + "\n".join(diff))

0 */6 * * * . /opt/relay-watch/env && python3 /opt/relay-watch/relay_watch.py >> /opt/relay-watch/cron.log 2>&1

jq -r '[.ts, .model, .verdict, .input_tokens, .latency_ms] | @tsv' relay_watch.jsonl | tail -20

Как читать изменения

Чего не покажет никакой внешний тест

Наши результаты

FAQ

Можно ли запустить тест на эндпоинте в формате OpenAI?

Как часто запускать?

Сколько стоит одна проверка?

Заметит ли тест замену Opus на Sonnet?

Если прокси прошёл тест, можно не беспокоиться?

Связанные материалы