feat(fingerprint): 引入 per-key 请求指纹系统,替代全局 TLS 指纹开关

为每个 ProviderAPIKey 生成并持久化独立的请求指纹配置,涵盖 TLS impersonate
profile、浏览器 UA、Stainless SDK 头部、Node/Chrome/Electron 版本等维度。
指纹基于 key ID 确定性生成,支持手动编辑和批量重新生成。

- 新增 fingerprint 模块:生成、加载、校验、懒持久化
- 数据库迁移:provider_api_keys 新增 fingerprint JSON 列
- 请求链路注入:handler 基类设置上下文指纹,request_builder 和 envelope 消费
- HTTP Client 支持动态 impersonate profile 选择
- Antigravity 适配器使用指纹覆盖 UA/session/Node 版本
- 前端移除手动 TLS 指纹开关,新增批量 regenerate_fingerprint 操作
- 号池管理 UI 优化:token 缩写格式、blocked 行样式、时间显示改为日期格式
This commit is contained in:
fawney19
2026-03-05 02:07:34 +08:00
parent 1d04c41ae7
commit 32ccf61baa
24 changed files with 782 additions and 84 deletions

View File

@@ -0,0 +1,41 @@
"""add fingerprint column to provider_api_keys
Revision ID: 6a9b8c7d5e4f
Revises: 5f1d2e3c4b5a
Create Date: 2026-03-04 23:50:00.000000
"""
from __future__ import annotations
from collections.abc import Sequence
import sqlalchemy as sa
from sqlalchemy import inspect
from alembic import op
# revision identifiers, used by Alembic.
revision: str = "6a9b8c7d5e4f"
down_revision: str | None = "5f1d2e3c4b5a"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def column_exists(table_name: str, column_name: str) -> bool:
bind = op.get_bind()
inspector = inspect(bind)
columns = [c["name"] for c in inspector.get_columns(table_name)]
return column_name in columns
def upgrade() -> None:
if not column_exists("provider_api_keys", "fingerprint"):
op.add_column(
"provider_api_keys",
sa.Column("fingerprint", sa.JSON(), nullable=True),
)
def downgrade() -> None:
if column_exists("provider_api_keys", "fingerprint"):
op.drop_column("provider_api_keys", "fingerprint")

View File

@@ -169,7 +169,13 @@ export interface PoolKeysQuery {
export interface PoolBatchAction {
key_ids: string[]
action: 'enable' | 'disable' | 'delete' | 'clear_cooldown' | 'reset_cost'
action:
| 'enable'
| 'disable'
| 'delete'
| 'clear_cooldown'
| 'reset_cost'
| 'regenerate_fingerprint'
}
export async function getPoolOverview(): Promise<PoolOverviewResponse> {

View File

@@ -442,8 +442,6 @@ export interface ClaudeCodeAdvancedConfig {
// 会话数量控制null/undefined 表示不限制
max_sessions?: number | null
session_idle_timeout_minutes?: number | null
// TLS 指纹模拟(模拟 Node.js/Claude Code 客户端指纹)
enable_tls_fingerprint?: boolean
// 会话 ID 伪装(固定 metadata.user_id 中 session 片段)
session_id_masking_enabled?: boolean
// Cache TTL 统一(强制所有 cache_control 使用相同 TTL 类型)

View File

@@ -203,18 +203,6 @@
<h3 class="text-sm font-medium border-b pb-2">
Claude Code
</h3>
<div class="flex items-center justify-between p-3 border rounded-lg bg-muted/50">
<div class="space-y-0.5">
<span class="text-sm font-medium">TLS 指纹模拟</span>
<p class="text-xs text-muted-foreground">
模拟 Node.js / Claude Code 客户端指纹
</p>
</div>
<Switch
:model-value="claudeForm.enable_tls_fingerprint"
@update:model-value="(v: boolean) => claudeForm.enable_tls_fingerprint = v"
/>
</div>
<div class="flex items-center justify-between p-3 border rounded-lg bg-muted/50">
<div class="space-y-0.5">
<span class="text-sm font-medium">Session ID 伪装</span>
@@ -507,7 +495,6 @@ interface ClaudeFormState {
session_control_enabled: boolean
max_sessions: number | undefined
session_idle_timeout_minutes: number
enable_tls_fingerprint: boolean
session_id_masking_enabled: boolean
cache_ttl_override_enabled: boolean
cache_ttl_override_target: string
@@ -518,7 +505,6 @@ const claudeForm = ref<ClaudeFormState>({
session_control_enabled: true,
max_sessions: undefined,
session_idle_timeout_minutes: 5,
enable_tls_fingerprint: true,
session_id_masking_enabled: true,
cache_ttl_override_enabled: false,
cache_ttl_override_target: 'ephemeral',
@@ -823,7 +809,6 @@ watch(() => props.modelValue, async (open) => {
session_control_enabled: cc?.max_sessions !== null,
max_sessions: cc?.max_sessions ?? undefined,
session_idle_timeout_minutes: cc?.session_idle_timeout_minutes ?? 5,
enable_tls_fingerprint: cc?.enable_tls_fingerprint !== false,
session_id_masking_enabled: cc?.session_id_masking_enabled !== false,
cache_ttl_override_enabled: cc?.cache_ttl_override_enabled ?? false,
cache_ttl_override_target: cc?.cache_ttl_override_target ?? 'ephemeral',
@@ -866,7 +851,6 @@ async function handleSave() {
payload.claude_code_advanced = {
max_sessions: cf.session_control_enabled ? (cf.max_sessions ?? null) : null,
session_idle_timeout_minutes: cf.session_control_enabled ? cf.session_idle_timeout_minutes : null,
enable_tls_fingerprint: cf.enable_tls_fingerprint,
session_id_masking_enabled: cf.session_id_masking_enabled,
cache_ttl_override_enabled: cf.cache_ttl_override_enabled,
cache_ttl_override_target: cf.cache_ttl_override_enabled ? cf.cache_ttl_override_target : undefined,

View File

@@ -310,7 +310,7 @@
v-for="key in keyPage.keys"
:key="key.key_id"
class="border-b border-border/40 last:border-b-0 hover:bg-muted/30 transition-colors"
:class="{ 'opacity-50': !key.is_active }"
:class="getRowClass(key)"
>
<TableCell class="py-3">
<div class="max-w-[260px] min-w-0">
@@ -461,7 +461,7 @@
<div class="flex items-center justify-between gap-2">
<span class="text-muted-foreground">Token</span>
<span class="tabular-nums text-foreground/90">
{{ formatStatInteger(key.total_tokens) }}
{{ formatTokenCount(key.total_tokens) }}
</span>
</div>
<div class="flex items-center justify-between gap-2">
@@ -599,7 +599,7 @@
v-for="key in keyPage.keys"
:key="key.key_id"
class="p-4 sm:p-5 hover:bg-muted/30 transition-colors"
:class="{ 'opacity-50': !key.is_active }"
:class="getRowClass(key)"
>
<div class="flex items-center gap-3">
<div class="flex-1 min-w-0">
@@ -825,7 +825,7 @@
</div>
<div class="flex items-center justify-between gap-2">
<span class="text-muted-foreground">Token</span>
<span class="tabular-nums">{{ formatStatInteger(key.total_tokens) }}</span>
<span class="tabular-nums">{{ formatTokenCount(key.total_tokens) }}</span>
</div>
<div class="flex items-center justify-between gap-2">
<span class="text-muted-foreground">费用</span>
@@ -1837,7 +1837,7 @@ function getSchedulingBadgeVariant(key: PoolKeyDetail): PoolStatusVariant {
if (getAccountAlertLabel(key)) return 'destructive'
const reason = key.scheduling_reason
if (reason === 'manual_disabled') return 'dark'
if (reason === 'manual_disabled') return 'secondary'
if (reason === 'cooldown' || reason === 'circuit_open' || reason === 'cost_exhausted') return 'destructive'
if (reason === 'cost_soft' || reason === 'cost') return 'warning'
if (reason === 'health_low' || reason === 'health_degraded' || reason === 'health') return 'warning'
@@ -1876,6 +1876,12 @@ function formatTTL(seconds: number): string {
return m > 0 ? `${m}m ${s}s` : `${s}s`
}
function getRowClass(key: PoolKeyDetail): string {
const status = getSchedulingStatus(key)
if (!key.is_active || status === 'blocked') return 'bg-muted/50 opacity-60'
return ''
}
function getHealthScoreColor(score: number): string {
if (score >= 0.8) return 'text-green-600 dark:text-green-400'
if (score >= 0.5) return 'text-yellow-600 dark:text-yellow-400'
@@ -2116,6 +2122,14 @@ function formatStatInteger(value: number | null | undefined): string {
return Math.round(n).toLocaleString('en-US')
}
function formatTokenCount(value: number | null | undefined): string {
const n = Number(value ?? 0)
if (!Number.isFinite(n) || n <= 0) return '0'
if (n >= 1_000_000) return `${(n / 1_000_000).toFixed(1)}M`
if (n >= 1_000) return `${(n / 1_000).toFixed(1)}K`
return String(Math.round(n))
}
function formatStatUsd(value: number | null | undefined): string {
const n = Number(value ?? 0)
if (!Number.isFinite(n) || n <= 0) return '$0.00'
@@ -2126,11 +2140,13 @@ function formatStatUsd(value: number | null | undefined): string {
}
function formatRelativeTime(isoStr: string): string {
const diff = (Date.now() - new Date(isoStr).getTime()) / 1000
if (diff < 60) return '刚刚'
if (diff < 3600) return `${Math.floor(diff / 60)}m 前`
if (diff < 86400) return `${Math.floor(diff / 3600)}h 前`
return `${Math.floor(diff / 86400)}d 前`
const date = new Date(isoStr)
const pad = (n: number) => String(n).padStart(2, '0')
const M = pad(date.getMonth() + 1)
const D = pad(date.getDate())
const h = pad(date.getHours())
const m = pad(date.getMinutes())
return `${M}-${D} ${h}:${m}`
}
// --- Init ---

View File

@@ -27,6 +27,7 @@ from src.core.exceptions import NotFoundException
from src.core.logger import logger
from src.database import get_db
from src.models.database import Provider, ProviderAPIKey, Usage
from src.services.provider.fingerprint import generate_fingerprint
from src.services.provider.pool import redis_ops as pool_redis
from src.services.provider.pool.account_state import resolve_pool_account_state
from src.services.provider.pool.config import parse_pool_config
@@ -143,7 +144,14 @@ async def batch_import_keys(
# POST /api/admin/pool/{provider_id}/keys/batch-action
# ---------------------------------------------------------------------------
ALLOWED_ACTIONS = {"enable", "disable", "delete", "clear_cooldown", "reset_cost"}
ALLOWED_ACTIONS = {
"enable",
"disable",
"delete",
"clear_cooldown",
"reset_cost",
"regenerate_fingerprint",
}
_COOLDOWN_REASON_LABELS: dict[str, str] = {
"rate_limited_429": "429 限流",
@@ -568,7 +576,7 @@ async def batch_action_keys(
request: Request,
db: Session = Depends(get_db),
) -> BatchActionResponse:
"""Batch enable/disable/delete/clear_cooldown/reset_cost on pool keys."""
"""Batch enable/disable/delete/clear_cooldown/reset_cost/regenerate_fingerprint on pool keys."""
adapter = AdminBatchActionKeysAdapter(provider_id=provider_id, body=body)
return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
@@ -977,6 +985,11 @@ class AdminListPoolKeysAdapter(AdminApiAdapter):
model_include_patterns=include_patterns,
model_exclude_patterns=exclude_patterns,
proxy=_mask_proxy_password(getattr(k, "proxy", None)),
fingerprint=(
getattr(k, "fingerprint", None)
if isinstance(getattr(k, "fingerprint", None), dict)
else None
),
account_quota=_build_account_quota(
provider_type,
getattr(k, "upstream_metadata", None),
@@ -1035,13 +1048,15 @@ class AdminBatchImportKeysAdapter(AdminApiAdapter):
try:
encrypted_key = crypto_service.encrypt(item.api_key)
new_key_id = str(uuid.uuid4())
new_key = ProviderAPIKey(
id=str(uuid.uuid4()),
id=new_key_id,
provider_id=self.provider_id,
name=item.name or f"imported-{idx}",
api_key=encrypted_key,
auth_type=item.auth_type or "api_key",
proxy=key_proxy,
fingerprint=generate_fingerprint(seed=new_key_id),
is_active=True,
created_at=now,
updated_at=now,
@@ -1136,7 +1151,11 @@ class AdminBatchActionKeysAdapter(AdminApiAdapter):
await pool_redis.clear_cost(pid, kid)
affected += 1
if self.body.action in {"enable", "disable", "delete"}:
elif self.body.action == "regenerate_fingerprint":
key.fingerprint = generate_fingerprint(seed=None)
affected += 1
if self.body.action in {"enable", "disable", "delete", "regenerate_fingerprint"}:
try:
db.commit()
except Exception as exc:
@@ -1150,6 +1169,7 @@ class AdminBatchActionKeysAdapter(AdminApiAdapter):
"delete": "deleted",
"clear_cooldown": "cooldown cleared",
"reset_cost": "cost reset",
"regenerate_fingerprint": "fingerprint regenerated",
}
admin_name = context.user.username if context.user else "admin"

View File

@@ -64,7 +64,6 @@ class PoolSchedulingReason(BaseModel):
detail: str | None = None
class PoolKeyDetail(BaseModel):
"""Detailed status of a single pool key."""
@@ -95,6 +94,7 @@ class PoolKeyDetail(BaseModel):
model_include_patterns: list[str] | None = None
model_exclude_patterns: list[str] | None = None
proxy: dict[str, Any] | None = None
fingerprint: dict[str, Any] | None = None
account_quota: str | None = None
cooldown_reason: str | None = None
cooldown_ttl_seconds: int | None = None
@@ -163,7 +163,7 @@ class BatchImportResponse(BaseModel):
class BatchActionRequest(BaseModel):
key_ids: list[str] = Field(..., max_length=500)
action: str # enable / disable / delete / clear_cooldown / reset_cost
action: str # enable / disable / delete / clear_cooldown / reset_cost / regenerate_fingerprint
class BatchActionResponse(BaseModel):

View File

@@ -1491,8 +1491,11 @@ class AdminImportConfigAdapter(AdminApiAdapter):
)
encrypted_auth_config = crypto_service.encrypt(auth_config_str)
from src.services.provider.fingerprint import generate_fingerprint
new_key_id = str(uuid.uuid4())
new_key = ProviderAPIKey(
id=str(uuid.uuid4()),
id=new_key_id,
provider_id=provider_id,
api_formats=normalized_formats,
auth_type=key_data.get("auth_type", "api_key"),
@@ -1514,6 +1517,7 @@ class AdminImportConfigAdapter(AdminApiAdapter):
model_exclude_patterns=key_data.get("model_exclude_patterns"),
is_active=key_data.get("is_active", True),
proxy=key_data.get("proxy"),
fingerprint=generate_fingerprint(seed=new_key_id),
health_by_format={},
circuit_breaker_by_format={},
)

View File

@@ -77,6 +77,8 @@ from src.models.database import (
User,
)
from src.services.provider.behavior import get_provider_behavior
from src.services.provider.fingerprint import ensure_key_fingerprint
from src.services.provider.request_context import set_current_fingerprint
from src.services.provider.stream_policy import (
enforce_stream_mode_for_upstream,
get_upstream_stream_policy,
@@ -719,6 +721,8 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
else:
request_body = dict(original_request_body)
set_current_fingerprint(ensure_key_fingerprint(key, persist_if_missing=True))
provider_type = str(getattr(provider, "provider_type", "") or "").lower()
behavior = get_provider_behavior(
provider_type=provider_type,

View File

@@ -40,6 +40,8 @@ from src.core.exceptions import (
)
from src.core.logger import logger
from src.services.provider.behavior import get_provider_behavior
from src.services.provider.fingerprint import ensure_key_fingerprint
from src.services.provider.request_context import set_current_fingerprint
from src.services.provider.stream_policy import (
enforce_stream_mode_for_upstream,
get_upstream_stream_policy,
@@ -325,6 +327,8 @@ class CliStreamMixin:
)
ctx.needs_conversion = needs_conversion
set_current_fingerprint(ensure_key_fingerprint(key, persist_if_missing=True))
provider_type = str(getattr(provider, "provider_type", "") or "").lower()
behavior = get_provider_behavior(
provider_type=provider_type,

View File

@@ -33,6 +33,8 @@ from src.core.exceptions import (
)
from src.core.logger import logger
from src.services.provider.behavior import get_provider_behavior
from src.services.provider.fingerprint import ensure_key_fingerprint
from src.services.provider.request_context import set_current_fingerprint
from src.services.provider.stream_policy import (
enforce_stream_mode_for_upstream,
get_upstream_stream_policy,
@@ -141,6 +143,8 @@ class CliSyncMixin:
)
needs_conversion = bool(getattr(candidate, "needs_conversion", False))
set_current_fingerprint(ensure_key_fingerprint(key, persist_if_missing=True))
provider_type = str(getattr(provider, "provider_type", "") or "").lower()
behavior = get_provider_behavior(
provider_type=provider_type,

View File

@@ -22,6 +22,8 @@ from typing import Any
from src.core.api_format import (
UPSTREAM_DROP_HEADERS,
HeaderBuilder,
build_anthropic_extra_headers,
build_browser_fingerprint_headers,
get_auth_config_for_endpoint,
make_signature_key,
)
@@ -29,6 +31,7 @@ from src.core.crypto import crypto_service
from src.models.endpoint_models import _CONDITION_OPS, _TYPE_IS_VALUES, parse_re_flags
from src.services.provider.auth import get_provider_auth # noqa: F401
from src.services.provider.envelope import ProviderEnvelope
from src.services.provider.request_context import get_current_fingerprint
# ==============================================================================
# 统一的头部配置常量
@@ -1202,6 +1205,17 @@ class PassthroughRequestBuilder(RequestBuilder):
pre_computed_auth: 预先计算的认证信息 (auth_header, auth_value)
用于 Service Account 等异步获取 token 的场景
"""
raw_family = getattr(endpoint, "api_family", None)
raw_kind = getattr(endpoint, "endpoint_kind", None)
endpoint_sig: str | None = None
if isinstance(raw_family, str) and isinstance(raw_kind, str) and raw_family and raw_kind:
endpoint_sig = make_signature_key(raw_family, raw_kind)
else:
# 兜底:允许 endpoint.api_format 已经是 signature key 的情况
raw_format = getattr(endpoint, "api_format", None)
if isinstance(raw_format, str) and ":" in raw_format:
endpoint_sig = raw_format
# 1. 根据 API 格式自动设置认证头
if pre_computed_auth:
# 使用预先计算的认证信息Service Account 等场景)
@@ -1209,21 +1223,6 @@ class PassthroughRequestBuilder(RequestBuilder):
else:
# 标准 API Key 认证
decrypted_key = crypto_service.decrypt(key.api_key)
raw_family = getattr(endpoint, "api_family", None)
raw_kind = getattr(endpoint, "endpoint_kind", None)
endpoint_sig: str | None = None
if (
isinstance(raw_family, str)
and isinstance(raw_kind, str)
and raw_family
and raw_kind
):
endpoint_sig = make_signature_key(raw_family, raw_kind)
else:
# 兜底:允许 endpoint.api_format 已经是 signature key 的情况
raw_format = getattr(endpoint, "api_format", None)
if isinstance(raw_format, str) and ":" in raw_format:
endpoint_sig = raw_format
auth_header, auth_type = get_auth_config_for_endpoint(endpoint_sig or "openai:chat")
auth_value = f"Bearer {decrypted_key}" if auth_type == "bearer" else decrypted_key
@@ -1244,7 +1243,13 @@ class PassthroughRequestBuilder(RequestBuilder):
if header_rules:
builder.apply_rules(header_rules, protected_keys)
# 4. 添加额外头部
# 4. 注入 per-key 指纹头(仅 Claude 格式需要浏览器指纹绕过 Cloudflare 检测)。
if str(endpoint_sig or "").strip().lower().startswith("claude:"):
fp = get_current_fingerprint()
builder.add_many(build_browser_fingerprint_headers(fp))
builder.add_many(build_anthropic_extra_headers(fp))
# 5. 添加额外头部
effective_extra_headers = self._merge_extra_headers_with_original(
original_headers,
extra_headers,
@@ -1253,10 +1258,10 @@ class PassthroughRequestBuilder(RequestBuilder):
if effective_extra_headers:
builder.add_many(effective_extra_headers)
# 5. 设置认证头(最高优先级,上游始终使用 header 认证)
# 6. 设置认证头(最高优先级,上游始终使用 header 认证)
builder.add(auth_header, auth_value)
# 6. 确保有 Content-Type
# 7. 确保有 Content-Type
headers = builder.build()
if not any(k.lower() == "content-type" for k in headers):
headers["Content-Type"] = "application/json"

View File

@@ -19,6 +19,7 @@ import httpx
from src.config import config
from src.core.logger import logger
from src.services.provider.fingerprint import KNOWN_IMPERSONATE_PROFILES
from src.services.proxy_node.resolver import (
build_proxy_url,
compute_proxy_cache_key,
@@ -246,17 +247,28 @@ class HTTPClientPool:
proxy_url = build_proxy_url(proxy_config) if proxy_config else None
# curl_cffi Transport: real TLS fingerprint impersonation.
# When tls_profile requires fingerprint impersonation and curl_cffi
# is available, use CurlCffiTransport instead of the default httpx
# transport. This gives us a genuine browser/Node.js TLS handshake.
if tls_profile_key == "claude_code_nodejs":
# Supports:
# - "claude_code_nodejs" (legacy alias, uses default chrome120 impersonate)
# - direct chrome impersonate profile names (e.g. "chrome124")
use_curl_cffi_tls = False
if tls_profile_key:
use_curl_cffi_tls = (
tls_profile_key == "claude_code_nodejs"
or tls_profile_key in KNOWN_IMPERSONATE_PROFILES
)
if use_curl_cffi_tls:
from src.clients.curl_cffi_transport import (
CURL_CFFI_AVAILABLE,
CurlCffiTransport,
)
if CURL_CFFI_AVAILABLE:
transport = CurlCffiTransport(proxy=proxy_url)
transport_kwargs: dict[str, Any] = {"proxy": proxy_url}
if tls_profile_key != "claude_code_nodejs":
transport_kwargs["impersonate"] = tls_profile_key
transport = CurlCffiTransport(**transport_kwargs)
client = httpx.AsyncClient(
transport=transport,
follow_redirects=True,

View File

@@ -32,6 +32,8 @@ from src.core.api_format.headers import (
HeaderBuilder,
build_adapter_base_headers_for_endpoint,
build_adapter_headers_for_endpoint,
build_anthropic_extra_headers,
build_browser_fingerprint_headers,
build_upstream_headers_for_endpoint,
detect_capabilities_for_endpoint,
extract_client_api_key_for_endpoint,
@@ -119,6 +121,8 @@ __all__ = [
"detect_capabilities_for_endpoint",
"HeaderBuilder",
"build_upstream_headers_for_endpoint",
"build_browser_fingerprint_headers",
"build_anthropic_extra_headers",
"merge_headers_with_protection",
"filter_response_headers",
"redact_headers_for_log",

View File

@@ -14,7 +14,7 @@ from __future__ import annotations
import json
from collections.abc import Set as AbstractSet
from typing import Any
from typing import TYPE_CHECKING, Any
from src.core.api_format.enums import ApiFamily
from src.core.api_format.metadata import (
@@ -26,6 +26,9 @@ from src.core.api_format.metadata import (
from src.core.api_format.signature import EndpointSignature, parse_signature_key
from src.core.logger import logger
if TYPE_CHECKING:
from src.services.provider.fingerprint import FingerprintProfile
# =============================================================================
# 头部常量定义
# =============================================================================
@@ -125,6 +128,59 @@ RESPONSE_DROP_HEADERS: frozenset[str] = (
)
_DEFAULT_CHROME_MAJOR = "140"
def _extract_chrome_major(version: str) -> str:
value = str(version or "").strip()
if not value:
return _DEFAULT_CHROME_MAJOR
return value.split(".", 1)[0] or _DEFAULT_CHROME_MAJOR
def _resolve_sec_ch_platform(fp: FingerprintProfile) -> str:
os_name = (fp.stainless_os or "").strip().lower()
platform_info = (fp.platform_info or "").strip().lower()
if os_name.startswith("win") or "windows" in platform_info:
return '"Windows"'
if os_name in {"mac", "macos", "darwin"} or "darwin" in platform_info:
return '"macOS"'
return '"Linux"'
def build_browser_fingerprint_headers(fp: FingerprintProfile | None = None) -> dict[str, str]:
"""Build browser fingerprint headers, optionally overridden by per-key fingerprint."""
headers = {**BROWSER_FINGERPRINT_HEADERS}
if fp is None:
return headers
chrome_major = _extract_chrome_major(fp.chrome_version)
headers["User-Agent"] = fp.user_agent or headers["User-Agent"]
headers["sec-ch-ua"] = f'"Not=A?Brand";v="24", "Chromium";v="{chrome_major}"'
headers["sec-ch-ua-platform"] = _resolve_sec_ch_platform(fp)
return headers
def build_anthropic_extra_headers(fp: FingerprintProfile | None = None) -> dict[str, str]:
"""Build Anthropic extra headers, optionally overridden by per-key fingerprint."""
headers = {**_ANTHROPIC_EXTRA_HEADERS}
if fp is None:
return headers
headers["x-stainless-os"] = fp.stainless_os or headers["x-stainless-os"]
headers["x-stainless-arch"] = fp.stainless_arch or headers["x-stainless-arch"]
headers["x-stainless-package-version"] = (
fp.stainless_package_version or headers["x-stainless-package-version"]
)
# browser:chrome runtime-version = Chrome 版本;
# CLI 路径(ClaudeCodeEnvelope)使用 Node.js其 extra_headers 会以 fp.stainless_runtime_version 覆盖。
headers["x-stainless-runtime-version"] = (
fp.chrome_version or headers["x-stainless-runtime-version"]
)
return headers
# =============================================================================
# 请求头规范化
# =============================================================================
@@ -568,12 +624,12 @@ def build_adapter_base_headers_for_endpoint(
auth_value = f"Bearer {api_key}" if auth_type == "bearer" else api_key
# 以浏览器指纹为底层默认值,绕过 Cloudflare 等反爬防护
headers: dict[str, str] = {**BROWSER_FINGERPRINT_HEADERS}
headers: dict[str, str] = build_browser_fingerprint_headers()
# Claude API family 额外注入 Anthropic 专属 header
definition = resolve_endpoint_definition(endpoint)
if definition and definition.api_family == ApiFamily.CLAUDE:
headers.update(_ANTHROPIC_EXTRA_HEADERS)
headers.update(build_anthropic_extra_headers())
headers[auth_header] = auth_value
headers["Content-Type"] = "application/json"

View File

@@ -1660,6 +1660,9 @@ class ProviderAPIKey(ExportMixin, Base):
# null 表示使用 Provider 级别代理(默认行为)
proxy = Column(JSON, nullable=True, default=None)
# 每个账号持久化的请求指纹配置TLS impersonate + HTTP header 维度)
fingerprint = Column(JSON, nullable=True, default=None)
# 时间戳
created_at = Column(
DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), nullable=False

View File

@@ -654,6 +654,10 @@ class EndpointAPIKeyUpdate(BaseModel):
default=None,
description="Key 级别代理配置(覆盖 Provider 级别代理null=使用 Provider 级别代理",
)
fingerprint: dict[str, Any] | None = Field(
default=None,
description="请求指纹配置TLS + HTTP 头部)",
)
@field_validator("api_formats")
@classmethod
@@ -823,6 +827,10 @@ class EndpointAPIKeyResponse(BaseModel):
proxy: dict[str, Any] | None = Field(
None, description="Key 级别代理配置(覆盖 Provider 级别代理)"
)
fingerprint: dict[str, Any] | None = Field(
None,
description="请求指纹配置TLS + HTTP 头部)",
)
# 时间戳
last_used_at: datetime | None = None

View File

@@ -14,6 +14,8 @@ import uuid
# ============== API 端点 ==============
# 唯一定义在 core 层,此处 re-export 保持向后兼容
from src.core.provider_templates.fixed_providers import ANTIGRAVITY_PROD_URL as PROD_BASE_URL
from src.services.provider.fingerprint import resolve_platform_token
from src.services.provider.request_context import get_current_fingerprint
DAILY_BASE_URL = "https://daily-cloudcode-pa.googleapis.com"
SANDBOX_BASE_URL = "https://daily-cloudcode-pa.sandbox.googleapis.com"
@@ -39,11 +41,27 @@ def _detect_platform_info() -> str:
_PLATFORM_INFO = _detect_platform_info()
def _build_antigravity_http_user_agent(
*,
platform_token: str,
version: str,
chrome_version: str,
electron_version: str,
) -> str:
return (
f"Mozilla/5.0 ({platform_token}) AppleWebKit/537.36 (KHTML, like Gecko) "
f"Antigravity/{version} Chrome/{chrome_version} "
f"Electron/{electron_version} Safari/537.36"
)
# HTTP Header User-Agent对齐 AM constants.rs: 完整 Electron 浏览器格式)
HTTP_USER_AGENT = (
f"Mozilla/5.0 ({_PLATFORM_INFO}) AppleWebKit/537.36 (KHTML, like Gecko) "
f"Antigravity/{_FALLBACK_VERSION} Chrome/{_FALLBACK_CHROME} "
f"Electron/{_FALLBACK_ELECTRON} Safari/537.36"
HTTP_USER_AGENT = _build_antigravity_http_user_agent(
platform_token=_PLATFORM_INFO,
version=_FALLBACK_VERSION,
chrome_version=_FALLBACK_CHROME,
electron_version=_FALLBACK_ELECTRON,
)
# V1InternalRequest.userAgent 字段(固定值)
@@ -57,10 +75,11 @@ _ua_version: str = _FALLBACK_VERSION
def get_http_user_agent() -> str:
"""返回当前 HTTP User-Agent 字符串(对齐 AM Electron UA 格式)。"""
with _ua_lock:
return (
f"Mozilla/5.0 ({_PLATFORM_INFO}) AppleWebKit/537.36 (KHTML, like Gecko) "
f"Antigravity/{_ua_version} Chrome/{_FALLBACK_CHROME} "
f"Electron/{_FALLBACK_ELECTRON} Safari/537.36"
return _build_antigravity_http_user_agent(
platform_token=_PLATFORM_INFO,
version=_ua_version,
chrome_version=_FALLBACK_CHROME,
electron_version=_FALLBACK_ELECTRON,
)
@@ -72,10 +91,11 @@ def update_user_agent_version(version: str) -> None:
return
with _ua_lock:
_ua_version = version
HTTP_USER_AGENT = (
f"Mozilla/5.0 ({_PLATFORM_INFO}) AppleWebKit/537.36 (KHTML, like Gecko) "
f"Antigravity/{_ua_version} Chrome/{_FALLBACK_CHROME} "
f"Electron/{_FALLBACK_ELECTRON} Safari/537.36"
HTTP_USER_AGENT = _build_antigravity_http_user_agent(
platform_token=_PLATFORM_INFO,
version=_ua_version,
chrome_version=_FALLBACK_CHROME,
electron_version=_FALLBACK_ELECTRON,
)
@@ -94,16 +114,43 @@ URL_UNAVAILABLE_TTL_SECONDS = 300 # 5 分钟
_SESSION_ID = uuid.uuid4().hex # 每次进程启动生成一个固定 session ID
def _normalize_node_version(raw: str | None) -> str:
text = str(raw or "").strip()
if text.startswith(("v", "V")):
text = text[1:]
return text or "18.18.2"
def get_v1internal_extra_headers() -> dict[str, str]:
"""构建 v1internal 请求需要的额外 header对齐 AM upstream/client.rs"""
fp = get_current_fingerprint()
with _ua_lock:
version = _ua_version
user_agent = get_http_user_agent()
session_id = _SESSION_ID
node_version = "18.18.2"
if fp:
user_agent = _build_antigravity_http_user_agent(
platform_token=resolve_platform_token(
platform_info=fp.platform_info,
stainless_os=fp.stainless_os,
stainless_arch=fp.stainless_arch,
),
version=version,
chrome_version=fp.chrome_version or _FALLBACK_CHROME,
electron_version=fp.electron_version or _FALLBACK_ELECTRON,
)
session_id = fp.vscode_session_id or _SESSION_ID
node_version = _normalize_node_version(fp.node_version)
return {
"User-Agent": get_http_user_agent(),
"User-Agent": user_agent,
"x-client-name": "antigravity",
"x-client-version": version,
"x-vscode-sessionid": _SESSION_ID,
"x-goog-api-client": "gl-node/18.18.2 fire/0.8.6 grpc/1.10.x",
"x-vscode-sessionid": session_id,
"x-goog-api-client": f"gl-node/{node_version} fire/0.8.6 grpc/1.10.x",
}

View File

@@ -26,6 +26,7 @@ from src.services.provider.adapters.claude_code.context import (
get_claude_code_request_context,
set_claude_code_request_context,
)
from src.services.provider.request_context import get_current_fingerprint
_SESSION_MARKER = "_session_"
_DUMMY_THINKING_SIGNATURE = "skip_thought_signature_validator"
@@ -488,6 +489,7 @@ class ClaudeCodeEnvelope:
def extra_headers(self) -> dict[str, str] | None:
ctx = get_claude_code_request_context()
is_stream = bool(ctx.is_stream) if ctx else False
fp = get_current_fingerprint()
headers = dict(CLAUDE_CODE_DEFAULT_HEADERS)
headers["Accept"] = DEFAULT_ACCEPT
@@ -496,9 +498,17 @@ class ClaudeCodeEnvelope:
if is_stream:
headers["x-stainless-helper-method"] = STREAM_HELPER_METHOD
ua = str(getattr(config, "internal_user_agent_claude_cli", "") or "").strip()
if ua:
headers["User-Agent"] = ua
if fp:
headers["X-Stainless-Package-Version"] = fp.stainless_package_version
headers["X-Stainless-OS"] = fp.stainless_os
headers["X-Stainless-Arch"] = fp.stainless_arch
headers["X-Stainless-Runtime-Version"] = fp.stainless_runtime_version
headers["X-Stainless-Timeout"] = fp.stainless_timeout
headers["User-Agent"] = fp.user_agent
else:
ua = str(getattr(config, "internal_user_agent_claude_cli", "") or "").strip()
if ua:
headers["User-Agent"] = ua
return headers
@@ -586,6 +596,9 @@ class ClaudeCodeEnvelope:
is_stream=is_stream,
provider_id=provider_id,
)
fp = get_current_fingerprint()
if _ctx.enable_tls_fingerprint and fp:
return fp.impersonate
return tls_profile
async def post_wrap_request(self, request_body: dict[str, Any]) -> None:

View File

@@ -0,0 +1,358 @@
"""Per-key request fingerprint generation and lazy persistence helpers."""
from __future__ import annotations
import asyncio
import hashlib
import random
import secrets
import threading
import uuid
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from typing import Any
from src.core.logger import logger
# curl_cffi impersonate profile pool (Chrome family)
CHROME_IMPERSONATE_PROFILES: tuple[str, ...] = (
"chrome110",
"chrome116",
"chrome119",
"chrome120",
"chrome123",
"chrome124",
"chrome131",
"chrome133",
)
KNOWN_IMPERSONATE_PROFILES: frozenset[str] = frozenset(CHROME_IMPERSONATE_PROFILES)
_CHROME_VERSION_BY_PROFILE: dict[str, str] = {
"chrome110": "110.0.5481.177",
"chrome116": "116.0.5845.188",
"chrome119": "119.0.6045.214",
"chrome120": "120.0.6099.216",
"chrome123": "123.0.6312.122",
"chrome124": "124.0.6367.243",
"chrome131": "131.0.6778.265",
"chrome133": "133.0.6943.142",
}
_PLATFORM_VARIANTS: tuple[tuple[str, str, str, str], ...] = (
("Linux", "x64", "X11; Linux x86_64", "Linux x86_64"),
("Linux", "arm64", "X11; Linux arm64", "Linux arm64"),
("Windows", "x64", "Windows NT 10.0; Win64; x64", "Windows x64"),
("MacOS", "x64", "Macintosh; Intel Mac OS X 10_15_7", "Darwin x64"),
("MacOS", "arm64", "Macintosh; ARM Mac OS X 14_0_0", "Darwin arm64"),
)
_STAINLESS_PACKAGE_VERSIONS: tuple[str, ...] = ("0.68.0", "0.69.0", "0.70.0", "0.71.0")
_NODE_VERSIONS: tuple[str, ...] = ("v20.18.1", "v22.12.0", "v22.14.0", "v24.13.0")
_ELECTRON_VERSIONS: tuple[str, ...] = ("35.5.1", "36.7.1", "37.3.0", "38.7.0", "39.2.3")
_STAINLESS_TIMEOUTS: tuple[str, ...] = ("600", "900")
_PENDING_LAZY_PERSIST: set[str] = set()
_PENDING_LAZY_PERSIST_LOCK = threading.Lock()
_PENDING_LAZY_PERSIST_MAX = 2000
@dataclass(frozen=True, slots=True)
class FingerprintProfile:
# TLS layer
impersonate: str
# Claude Code / Claude Chat feature dimensions
stainless_package_version: str
stainless_os: str
stainless_arch: str
stainless_runtime_version: str
stainless_timeout: str
# Generic dimensions
user_agent: str
node_version: str
chrome_version: str
electron_version: str
vscode_session_id: str
platform_info: str
def _build_rng(seed: str | None) -> random.Random:
if seed is None:
return random.Random(secrets.randbits(64))
digest = hashlib.sha256(seed.encode("utf-8")).digest()
return random.Random(int.from_bytes(digest[:8], "big", signed=False))
def _normalize_impersonate(raw: Any, fallback: str) -> str:
value = str(raw or "").strip().lower()
return value if value in KNOWN_IMPERSONATE_PROFILES else fallback
def _build_user_agent(
platform_token: str,
chrome_version: str,
electron_version: str,
) -> str:
return (
f"Mozilla/5.0 ({platform_token}) AppleWebKit/537.36 (KHTML, like Gecko) "
f"Chrome/{chrome_version} Electron/{electron_version} Safari/537.36"
)
def resolve_platform_token(
fp: dict[str, str] | None = None,
*,
platform_info: str | None = None,
stainless_os: str | None = None,
stainless_arch: str | None = None,
) -> str:
"""Resolve a platform token string from fingerprint dict or explicit kwargs."""
if fp is not None:
os_name = str(fp.get("stainless_os") or "").lower()
arch = str(fp.get("stainless_arch") or "").lower()
info = str(fp.get("platform_info") or "").lower()
else:
os_name = str(stainless_os or "").lower()
arch = str(stainless_arch or "").lower()
info = str(platform_info or "").lower()
if os_name.startswith("win") or "windows" in info:
return "Windows NT 10.0; Win64; x64"
if os_name in {"darwin", "mac", "macos"} or "darwin" in info:
return (
"Macintosh; ARM Mac OS X 14_0_0"
if arch in {"arm64", "aarch64"}
else "Macintosh; Intel Mac OS X 10_15_7"
)
return "X11; Linux arm64" if arch in {"arm64", "aarch64"} else "X11; Linux x86_64"
def _normalize_text(value: Any, fallback: str) -> str:
text = str(value or "").strip()
return text or fallback
def _sanitize_fingerprint_dict(raw: dict[str, Any], key_id: str) -> dict[str, str]:
generated = generate_fingerprint(seed=key_id or None)
fp: dict[str, str] = {k: str(v) for k, v in generated.items()}
for key, value in raw.items():
if isinstance(value, str) and value.strip():
fp[key] = value.strip()
fp["impersonate"] = _normalize_impersonate(fp.get("impersonate"), generated["impersonate"])
fp["chrome_version"] = _normalize_text(
fp.get("chrome_version"),
_CHROME_VERSION_BY_PROFILE.get(fp["impersonate"], generated["chrome_version"]),
)
fp["node_version"] = _normalize_text(fp.get("node_version"), generated["node_version"])
fp["electron_version"] = _normalize_text(
fp.get("electron_version"),
generated["electron_version"],
)
fp["stainless_package_version"] = _normalize_text(
fp.get("stainless_package_version"),
generated["stainless_package_version"],
)
fp["stainless_os"] = _normalize_text(fp.get("stainless_os"), generated["stainless_os"])
fp["stainless_arch"] = _normalize_text(fp.get("stainless_arch"), generated["stainless_arch"])
fp["stainless_runtime_version"] = _normalize_text(
fp.get("stainless_runtime_version"),
generated["stainless_runtime_version"],
)
fp["stainless_timeout"] = _normalize_text(
fp.get("stainless_timeout"),
generated["stainless_timeout"],
)
fp["platform_info"] = _normalize_text(fp.get("platform_info"), generated["platform_info"])
fp["vscode_session_id"] = _normalize_text(
fp.get("vscode_session_id"),
generated["vscode_session_id"],
)
if not str(fp.get("user_agent") or "").strip():
fp["user_agent"] = _build_user_agent(
resolve_platform_token(fp),
fp["chrome_version"],
fp["electron_version"],
)
return fp
def generate_fingerprint(seed: str | None = None) -> dict[str, str]:
"""Generate a serializable fingerprint profile dict."""
rng = _build_rng(seed)
impersonate = rng.choice(CHROME_IMPERSONATE_PROFILES)
chrome_version = _CHROME_VERSION_BY_PROFILE.get(impersonate, "120.0.6099.216")
node_version = rng.choice(_NODE_VERSIONS)
electron_version = rng.choice(_ELECTRON_VERSIONS)
stainless_os, stainless_arch, platform_token, platform_info = rng.choice(_PLATFORM_VARIANTS)
if seed is None:
vscode_session_id = uuid.uuid4().hex
else:
vscode_session_id = uuid.uuid5(uuid.NAMESPACE_URL, f"aether:fingerprint:{seed}").hex
return {
"impersonate": impersonate,
"stainless_package_version": rng.choice(_STAINLESS_PACKAGE_VERSIONS),
"stainless_os": stainless_os,
"stainless_arch": stainless_arch,
"stainless_runtime_version": node_version,
"stainless_timeout": rng.choice(_STAINLESS_TIMEOUTS),
"node_version": node_version,
"chrome_version": chrome_version,
"electron_version": electron_version,
"vscode_session_id": vscode_session_id,
"platform_info": platform_info,
"user_agent": _build_user_agent(platform_token, chrome_version, electron_version),
}
def _dict_to_profile(d: dict[str, str]) -> FingerprintProfile:
return FingerprintProfile(
impersonate=d["impersonate"],
stainless_package_version=d["stainless_package_version"],
stainless_os=d["stainless_os"],
stainless_arch=d["stainless_arch"],
stainless_runtime_version=d["stainless_runtime_version"],
stainless_timeout=d["stainless_timeout"],
user_agent=d["user_agent"],
node_version=d["node_version"],
chrome_version=d["chrome_version"],
electron_version=d["electron_version"],
vscode_session_id=d["vscode_session_id"],
platform_info=d["platform_info"],
)
def load_fingerprint(raw: dict[str, Any] | None, key_id: str) -> FingerprintProfile:
"""Load fingerprint from DB JSON with deterministic fallback by key_id."""
if raw and isinstance(raw, dict):
normalized = _sanitize_fingerprint_dict(raw, key_id)
else:
normalized = generate_fingerprint(seed=key_id or None)
return _dict_to_profile(normalized)
def serialize_fingerprint(fp: FingerprintProfile) -> dict[str, str]:
return {k: str(v) for k, v in asdict(fp).items()}
def normalize_fingerprint(raw: dict[str, Any], key_id: str) -> dict[str, str]:
return serialize_fingerprint(load_fingerprint(raw, key_id))
def _persist_fingerprint_if_missing_sync(key_id: str, fp: dict[str, str]) -> None:
from src.database import create_session
from src.models.database import ProviderAPIKey
db = create_session()
try:
updated = (
db.query(ProviderAPIKey)
.filter(
ProviderAPIKey.id == key_id,
ProviderAPIKey.fingerprint.is_(None),
)
.update(
{
ProviderAPIKey.fingerprint: fp,
ProviderAPIKey.updated_at: datetime.now(timezone.utc),
},
synchronize_session=False,
)
)
if updated > 0:
db.commit()
except Exception as exc:
db.rollback()
logger.debug("lazy fingerprint persist failed for key {}: {}", key_id[:8], str(exc))
finally:
db.close()
def _mark_pending_persist(key_id: str) -> bool:
with _PENDING_LAZY_PERSIST_LOCK:
if key_id in _PENDING_LAZY_PERSIST:
return False
if len(_PENDING_LAZY_PERSIST) >= _PENDING_LAZY_PERSIST_MAX:
return False
_PENDING_LAZY_PERSIST.add(key_id)
return True
def _clear_pending_persist(key_id: str) -> None:
with _PENDING_LAZY_PERSIST_LOCK:
_PENDING_LAZY_PERSIST.discard(key_id)
def schedule_lazy_fingerprint_persist(key_id: str, fp: dict[str, str]) -> None:
"""Persist generated fingerprint in background when key.fingerprint is missing."""
key_id = str(key_id or "").strip()
if not key_id:
return
if not _mark_pending_persist(key_id):
return
payload = dict(fp)
async def _persist_async() -> None:
try:
await asyncio.to_thread(_persist_fingerprint_if_missing_sync, key_id, payload)
finally:
_clear_pending_persist(key_id)
try:
loop = asyncio.get_running_loop()
except RuntimeError:
try:
_persist_fingerprint_if_missing_sync(key_id, payload)
finally:
_clear_pending_persist(key_id)
return
task = loop.create_task(_persist_async())
def _on_done(done_task: asyncio.Task[None]) -> None:
try:
done_task.result()
except Exception as exc:
logger.debug("lazy fingerprint background task failed: {}", str(exc))
task.add_done_callback(_on_done)
def ensure_key_fingerprint(
key: Any,
*,
persist_if_missing: bool = False,
) -> FingerprintProfile:
"""Get a key fingerprint, generating deterministic fallback if missing."""
key_id = str(getattr(key, "id", "") or "").strip()
raw = getattr(key, "fingerprint", None)
if isinstance(raw, dict) and raw:
return load_fingerprint(raw, key_id)
generated = generate_fingerprint(seed=key_id or None)
if persist_if_missing and key_id:
schedule_lazy_fingerprint_persist(key_id, generated)
return _dict_to_profile(generated)
__all__ = [
"CHROME_IMPERSONATE_PROFILES",
"FingerprintProfile",
"KNOWN_IMPERSONATE_PROFILES",
"ensure_key_fingerprint",
"generate_fingerprint",
"load_fingerprint",
"normalize_fingerprint",
"resolve_platform_token",
"schedule_lazy_fingerprint_persist",
"serialize_fingerprint",
]

View File

@@ -10,11 +10,19 @@ per request.
from __future__ import annotations
import contextvars
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from src.services.provider.fingerprint import FingerprintProfile
_selected_base_url: contextvars.ContextVar[str | None] = contextvars.ContextVar(
"provider_selected_base_url",
default=None,
)
_current_fingerprint: contextvars.ContextVar[FingerprintProfile | None] = contextvars.ContextVar(
"provider_current_fingerprint",
default=None,
)
def set_selected_base_url(url: str | None) -> None:
@@ -25,4 +33,17 @@ def get_selected_base_url() -> str | None:
return _selected_base_url.get()
__all__ = ["get_selected_base_url", "set_selected_base_url"]
def set_current_fingerprint(fp: FingerprintProfile | None) -> None:
_current_fingerprint.set(fp)
def get_current_fingerprint() -> FingerprintProfile | None:
return _current_fingerprint.get()
__all__ = [
"get_current_fingerprint",
"get_selected_base_url",
"set_current_fingerprint",
"set_selected_base_url",
]

View File

@@ -23,6 +23,7 @@ from src.models.endpoint_models import (
EndpointAPIKeyResponse,
EndpointAPIKeyUpdate,
)
from src.services.provider.fingerprint import generate_fingerprint, normalize_fingerprint
from src.services.provider_keys.auth_type import normalize_auth_type
from src.services.provider_keys.duplicate_check import check_duplicate_key
from src.services.provider_keys.key_side_effects import (
@@ -282,6 +283,12 @@ def _prepare_update_key_payload(
else:
update_data["proxy"] = key_data.proxy.model_dump(exclude_none=True)
if "fingerprint" in key_data.model_fields_set:
if key_data.fingerprint is None:
update_data["fingerprint"] = None
else:
update_data["fingerprint"] = normalize_fingerprint(key_data.fingerprint, key_id)
return _UpdateKeyPreparation(
update_data=update_data,
auto_fetch_enabled_before=auto_fetch_enabled_before,
@@ -335,8 +342,10 @@ def _prepare_create_key_payload(
if key_data.auth_config:
encrypted_auth_config = crypto_service.encrypt(json.dumps(key_data.auth_config))
new_key_id = str(uuid.uuid4())
new_key = ProviderAPIKey(
id=str(uuid.uuid4()),
id=new_key_id,
provider_id=provider_id,
api_formats=key_data.api_formats,
auth_type=auth_type,
@@ -359,6 +368,7 @@ def _prepare_create_key_payload(
model_exclude_patterns=(
key_data.model_exclude_patterns if key_data.model_exclude_patterns else None
),
fingerprint=generate_fingerprint(seed=new_key_id),
request_count=0,
success_count=0,
error_count=0,

View File

@@ -21,13 +21,17 @@ from src.services.provider.adapters.claude_code.envelope import (
claude_code_envelope,
merge_anthropic_beta_tokens,
)
from src.services.provider.fingerprint import load_fingerprint
from src.services.provider.request_context import set_current_fingerprint
@pytest.fixture(autouse=True)
def _reset_claude_code_context():
def _reset_claude_code_context() -> None: # type: ignore[misc]
set_claude_code_request_context(None)
set_current_fingerprint(None)
yield
set_claude_code_request_context(None)
set_current_fingerprint(None)
def test_merge_anthropic_beta_tokens_adds_required_and_deduplicates() -> None:
@@ -45,7 +49,7 @@ def test_merge_anthropic_beta_tokens_adds_required_and_deduplicates() -> None:
def test_claude_code_envelope_extra_headers_include_required_defaults(
monkeypatch,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(config, "internal_user_agent_claude_cli", "claude-code/test")
@@ -61,6 +65,43 @@ def test_claude_code_envelope_extra_headers_include_required_defaults(
assert "x-stainless-helper-method" not in headers
def test_claude_code_envelope_extra_headers_use_current_fingerprint() -> None:
fp = load_fingerprint(
{
"stainless_package_version": "1.0.5",
"stainless_os": "Windows",
"stainless_arch": "x64",
"stainless_runtime_version": "v22.12.0",
"stainless_timeout": "900",
"user_agent": "Mozilla/5.0 test-fingerprint",
},
"key-fingerprint-1",
)
set_current_fingerprint(fp)
headers = claude_code_envelope.extra_headers() or {}
assert headers.get("X-Stainless-Package-Version") == "1.0.5"
assert headers.get("X-Stainless-OS") == "Windows"
assert headers.get("X-Stainless-Arch") == "x64"
assert headers.get("X-Stainless-Runtime-Version") == "v22.12.0"
assert headers.get("X-Stainless-Timeout") == "900"
assert headers.get("User-Agent") == "Mozilla/5.0 test-fingerprint"
def test_claude_code_prepare_context_prefers_fingerprint_tls_profile() -> None:
fp = load_fingerprint({"impersonate": "chrome124"}, "key-fingerprint-2")
set_current_fingerprint(fp)
tls_profile = claude_code_envelope.prepare_context(
provider_config={"claude_code_advanced": {"enable_tls_fingerprint": True}},
key_id="key-fingerprint-2",
is_stream=False,
provider_id="provider-1",
)
assert tls_profile == "chrome124"
def test_claude_code_envelope_adds_stream_helper_header_for_stream_request() -> None:
_, _ = claude_code_envelope.wrap_request(
{"stream": True},

View File

@@ -0,0 +1,39 @@
from __future__ import annotations
from types import SimpleNamespace
from src.services.provider.fingerprint import (
CHROME_IMPERSONATE_PROFILES,
ensure_key_fingerprint,
generate_fingerprint,
load_fingerprint,
)
def test_generate_fingerprint_is_deterministic_with_seed() -> None:
first = generate_fingerprint(seed="key-123")
second = generate_fingerprint(seed="key-123")
assert first == second
def test_load_fingerprint_falls_back_for_invalid_impersonate() -> None:
profile = load_fingerprint({"impersonate": "invalid-profile"}, "key-abc")
assert profile.impersonate in CHROME_IMPERSONATE_PROFILES
def test_ensure_key_fingerprint_returns_profile_when_missing() -> None:
key = SimpleNamespace(id="key-xyz", fingerprint=None)
profile = ensure_key_fingerprint(key, persist_if_missing=False)
assert profile.impersonate in CHROME_IMPERSONATE_PROFILES
# Deterministic: same key_id produces same profile
profile2 = ensure_key_fingerprint(key, persist_if_missing=False)
assert profile.impersonate == profile2.impersonate
def test_ensure_key_fingerprint_uses_existing_fingerprint() -> None:
existing = generate_fingerprint(seed="key-existing")
key = SimpleNamespace(id="key-existing", fingerprint=existing)
profile = ensure_key_fingerprint(key, persist_if_missing=False)
assert profile.impersonate == existing["impersonate"]