diff --git a/Dockerfile b/Dockerfile index 9fb83a7..6883204 100644 --- a/Dockerfile +++ b/Dockerfile @@ -28,4 +28,6 @@ COPY converter.py ./ COPY app/ ./app/ COPY --from=frontend /web/dist/ ./web/dist/ EXPOSE 8787 +# 容器内监听 0.0.0.0 是硬需求；对外暴露边界在端口映射层（compose 默认只绑回环） +ENV CODEBUDDY2API_ALLOW_OPEN_NOAUTH=true CMD ["python3", "converter.py", "--host", "0.0.0.0", "--port", "8787", "--skip-check"] diff --git a/app/adapters/anthropic_adapter.py b/app/adapters/anthropic_adapter.py index 4668478..8b3fd65 100644 --- a/app/adapters/anthropic_adapter.py +++ b/app/adapters/anthropic_adapter.py @@ -155,8 +155,11 @@ def _convert_anthropic_message(msg: dict) -> list[dict]: if isinstance(output, list): output = _convert_content_blocks(output) if block.get("is_error") is True: - # Chat 协议无等价字段：失败标记编码进正文前缀，模型不再把失败当有效结果 - output = "[tool execution failed]\n" + (output or "") + # Chat 协议无等价字段：失败标记编码进正文；含图片的列表内容前置文本块而非拼接 + if isinstance(output, list): + output = [{"type": "text", "text": "[tool execution failed]"}] + output + else: + output = "[tool execution failed]\n" + (output or "") result.append({"role": "tool", "tool_call_id": tc_id, "content": output}) if user_blocks: # 工具结果必须紧随 assistant 的 tool_calls；普通文本排在它们之后 diff --git a/app/admin_api.py b/app/admin_api.py index c6b256b..1910e45 100644 --- a/app/admin_api.py +++ b/app/admin_api.py @@ -336,13 +336,16 @@ def upload(body): directory = gateway.managed_auth_dir().resolve() for name, content in prepared: try: - data = json.loads(content) + data = auth_oauth.loads_strict(content) # 严格解析：拒绝 NaN/Infinity 常量 uid, invalid = auth_oauth.validate_cred_data(data) if invalid or not isinstance(data.get("account") or {}, dict) or type(data["auth"].get("expiresAt", 0)) not in (int, float): raise CredentialFileError("凭据格式无效") if not body.get("replace", False) and (directory / name).exists(): results.append({"name": name, "ok": False, "error": "文件已存在，需明确允许替换"}) continue + # 落盘统一为规范形态：token 别名折叠为官方字段名，运行时只读 accessToken + content = json.dumps(auth_oauth.normalize_cred_data(data), + ensure_ascii=False).encode("utf-8") gateway._store_credential(directory, name, content, uid, replace_existing=body.get("replace", False)) results.append({"name": name, "ok": True}) except Exception: diff --git a/app/audit_store.py b/app/audit_store.py index 526e7f4..9cff907 100644 --- a/app/audit_store.py +++ b/app/audit_store.py @@ -60,7 +60,7 @@ def safe_attempt(value: Any) -> dict: clean = safe_label(value.get(key)) if clean is not None: result[key] = clean - for key in ("status_code", "duration_ms", "attempt", "retry_after"): + for key in ("status_code", "duration_ms", "attempt", "retry_after", "max_attempts", "total_tokens"): clean = number(value.get(key)) if clean is not None: result[key] = clean diff --git a/app/auth_oauth.py b/app/auth_oauth.py index 1b7374f..2506c8c 100644 --- a/app/auth_oauth.py +++ b/app/auth_oauth.py @@ -13,6 +13,7 @@ from __future__ import annotations import base64 +import copy import json import re import threading @@ -69,6 +70,15 @@ def _token_issuer_origin(access_token: str) -> str: return "" +def _reject_constant(value): + raise ValueError(f"非标准 JSON 常量: {value}") + + +def loads_strict(text): + """严格 JSON 解析：拒绝 NaN/Infinity 等非标准常量（json.loads 默认接受）。""" + return json.loads(text, parse_constant=_reject_constant) + + def validate_cred_data(data) -> tuple[str | None, str | None]: """入库校验（严格模式）：返回 (uid, None) 或 (None, 原因)。""" if not isinstance(data, dict): @@ -84,6 +94,14 @@ def validate_cred_data(data) -> tuple[str | None, str | None]: token = auth.get("accessToken") or auth.get("access_token") or auth.get("token") if not isinstance(token, str) or not token: return None, "缺少有效的 accessToken" + for field in ("expiresAt", "lastRefreshTime"): + value = auth.get(field) + if value is None: + continue + # 原值范围比较同时拒绝非有限浮点数，避免超大整数转 float 溢出。 + if isinstance(value, bool) or not isinstance(value, (int, float)) \ + or not 0 < value < 4102444800000: # 上限 2100-01-01 + return None, f"{field} 必须是合理范围内的有限毫秒时间戳" domain = _normalize_origin(auth.get("domain") or auth.get("issuer") or "") issuer = _token_issuer_origin(token) if not any(o in ALLOWED_ORIGINS for o in (domain, issuer) if o): @@ -95,6 +113,28 @@ def validate_cred_data(data) -> tuple[str | None, str | None]: return uid, None +def normalize_cred_data(data: dict) -> dict: + """校验通过后生成唯一规范形态：token 别名折叠为官方字段名。 + + 运行时（client_profiles.credential_headers 等）只读 accessToken/refreshToken； + 导入侧若接受别名却不归一化，会得到「导入成功但认证头为空」的凭据。""" + out = copy.deepcopy(data) + auth = out.get("auth") + if not isinstance(auth, dict): + return out + for canonical, aliases in (("accessToken", ("access_token", "token")), + ("refreshToken", ("refresh_token",)), + ("tokenType", ("token_type",))): + if not auth.get(canonical): + for alias in aliases: + if auth.get(alias): + auth[canonical] = auth[alias] + break + for alias in aliases: + auth.pop(alias, None) + return out + + def _norm_ts(v) -> int | None: """时间戳归一化：秒/毫秒/数字字符串 → 毫秒；无效返回 None。""" if isinstance(v, bool): diff --git a/app/credits.py b/app/credits.py index dfba871..4da1c7b 100644 --- a/app/credits.py +++ b/app/credits.py @@ -343,7 +343,7 @@ def fetch_credits(access_token: str, uid: str = "", domain: str = "") -> dict: if page > CREDITS_MAX_PAGES: partial = True break - rows = _fetch_accounts_page(client, url, headers, page, retry_empty=(page == 1)) + rows = _fetch_accounts_page(client, url, headers, page, retry_empty=True) # 空页在任何页都可能是瞬时现象，一律重试 accounts.extend(rows) if len(rows) < CREDITS_PAGE_SIZE: break diff --git a/app/inbound_limits.py b/app/inbound_limits.py new file mode 100644 index 0000000..6eec137 --- /dev/null +++ b/app/inbound_limits.py @@ -0,0 +1,127 @@ +#!/usr/bin/env python3 +"""inbound_limits.py — 推理端点入站原始字节限量。 + +端点先 await request.json() 再检查处理后的上游请求体，原始入站大小无人管：大 JSON、 +被忽略的顶层字段、将被剥离的图片都在解析前已经占用了内存。本中间件在 ASGI receive 层 +累计原始字节（含 chunked 传输，不看 Content-Length），超限直接 413，不进入 JSON 解析。 +缓冲体随后原样回放给下游，端点行为不变。 +""" + +from __future__ import annotations + +import json + + +import asyncio + + +_GATED_PATHS = ("/v1/chat/completions", "/v1/responses", "/v1/messages") +_BODY_PATHS = (*_GATED_PATHS, "/v1/messages/count_tokens") + + +class ConcurrencyLimitMiddleware: + """推理端点并发上限：占满立即 503，不排队放大聚合内存。 + + 信号量从进入持有到响应体发完（含流式），覆盖整个上游连接生命周期。""" + + def __init__(self, app, config): + self.app = app + self.config = config + self._semaphore = None + + def _gate(self): + if self._semaphore is None: + self._semaphore = asyncio.Semaphore(self._limit()) + return self._semaphore + + def _limit(self) -> int: + try: + return max(0, int(self.config.get("max_concurrent") or 0)) + except (TypeError, ValueError): + return 0 + + async def __call__(self, scope, receive, send): + if (scope["type"] != "http" or scope.get("method") != "POST" + or scope.get("path", "") not in _GATED_PATHS): + return await self.app(scope, receive, send) + limit = self._limit() + if limit <= 0: + return await self.app(scope, receive, send) + gate = self._gate() + if gate.locked(): # 无空闲名额：立即失败并给出重试提示 + error = {"message": "inference concurrency limit reached, retry later", + "type": "rate_limit_error", "code": "concurrency_limit"} + payload = {"error": error} + if scope["path"] == "/v1/messages": + error["type"] = "api_error" + payload["type"] = "error" + raw = json.dumps(payload).encode() + await send({"type": "http.response.start", "status": 503, + "headers": [(b"content-type", b"application/json"), (b"retry-after", b"3"), + (b"content-length", str(len(raw)).encode())]}) + await send({"type": "http.response.body", "body": raw}) + return + await gate.acquire() + try: + await self.app(scope, receive, send) + finally: + gate.release() + + +class InboundBodyLimitMiddleware: + """仅缓冲生成及 token 估算 POST；其他路由不读取请求体。""" + + def __init__(self, app, config): + self.app = app + self.config = config + + async def __call__(self, scope, receive, send): + if (scope["type"] != "http" or scope.get("method") != "POST" + or scope.get("path", "") not in _BODY_PATHS): + return await self.app(scope, receive, send) + try: + limit = int(self.config.get("max_inbound_bytes") or 0) + except (TypeError, ValueError): + limit = 0 + if limit <= 0: + return await self.app(scope, receive, send) + + body = bytearray() + more = True + while more: + message = await receive() + if message["type"] == "http.request": + body.extend(message.get("body", b"")) + more = bool(message.get("more_body")) + if len(body) > limit: + return await self._reject(send, scope["path"], limit) + elif message["type"] == "http.disconnect": + return + + buffered = bytes(body) + replayed = False + + async def replay(): + nonlocal replayed + if replayed: + return await receive() # 请求体结束不等于断连，继续监听真实连接。 + replayed = True + return {"type": "http.request", "body": buffered, "more_body": False} + + await self.app(scope, replay, send) + + @staticmethod + async def _reject(send, path: str, limit: int): + # 与协议化错误处理一致的外形（middle ware 在路由之前，自行成形） + if path.startswith("/v1/messages"): + payload = {"type": "error", "error": {"type": "invalid_request_error", + "message": f"request body exceeds {limit} bytes", + "code": "request_too_large"}} + else: + payload = {"error": {"message": f"request body exceeds {limit} bytes", + "type": "invalid_request_error", "code": "request_too_large"}} + raw = json.dumps(payload).encode("utf-8") + await send({"type": "http.response.start", "status": 413, + "headers": [(b"content-type", b"application/json"), + (b"content-length", str(len(raw)).encode())]}) + await send({"type": "http.response.body", "body": raw}) diff --git a/app/inference_auth.py b/app/inference_auth.py new file mode 100644 index 0000000..7b4a5f8 --- /dev/null +++ b/app/inference_auth.py @@ -0,0 +1,43 @@ +"""Shared API-key checks and header-only admission before inference buffering.""" + +import secrets + +from fastapi import HTTPException +from starlette.datastructures import Headers +from starlette.responses import JSONResponse + +from .inbound_limits import _BODY_PATHS + + +def require_api_key(key, authorization=None, x_api_key=None): + """Preserve Bearer precedence, X-Api-Key fallback, and optional empty keys.""" + if not key: + return + token = "" + if isinstance(authorization, str) and authorization.startswith("Bearer "): + token = authorization[7:].strip() + if not token and isinstance(x_api_key, str): + token = x_api_key + if not secrets.compare_digest(token.encode(), key.encode()): + raise HTTPException(status_code=401, detail={"error": {"message": "invalid api key", "type": "auth_error"}}) + + +class InferenceAuthMiddleware: + def __init__(self, app, config): + self.app = app + self.config = config + + async def __call__(self, scope, receive, send): + if (scope["type"] != "http" or scope.get("method") != "POST" + or scope.get("path", "") not in _BODY_PATHS): + return await self.app(scope, receive, send) + headers = Headers(scope=scope) + try: + require_api_key(self.config.get("api_key"), headers.get("authorization"), headers.get("x-api-key")) + except HTTPException as error: + payload = error.detail + if scope["path"].startswith("/v1/messages"): + payload = {"type": "error", "error": {**payload["error"], "type": "authentication_error"}} + response = JSONResponse(payload, status_code=error.status_code, headers=error.headers) + return await response(scope, receive, send) + await self.app(scope, receive, send) diff --git a/app/runtime_management.py b/app/runtime_management.py index 86b2c04..cf68197 100644 --- a/app/runtime_management.py +++ b/app/runtime_management.py @@ -8,6 +8,7 @@ from .audit_store import AuditStore from .control_store import ControlStore from .gateway_management import Management, install_pages +from .inference_auth import InferenceAuthMiddleware from .model_policy import PolicyScopeMiddleware from .observability import AuditMiddleware from .settings import SCHEMA, apply_persisted_settings @@ -91,6 +92,11 @@ def install(gateway): install_admin(app, config, config["management"]) app.add_middleware(AuditMiddleware, config=config) app.add_middleware(PolicyScopeMiddleware) + from .inbound_limits import ConcurrencyLimitMiddleware, InboundBodyLimitMiddleware + app.add_middleware(InboundBodyLimitMiddleware, config=config) + app.add_middleware(ConcurrencyLimitMiddleware, config=config) + # 最外层先校验请求头；未鉴权的慢请求不得占用推理名额或进入请求体缓冲。 + app.add_middleware(InferenceAuthMiddleware, config=config) install_pages(app, Path(gateway.__file__).resolve().parent / "web" / "dist") diff --git a/app/upstream_io.py b/app/upstream_io.py index bf7d8ef..a4267de 100644 --- a/app/upstream_io.py +++ b/app/upstream_io.py @@ -21,8 +21,10 @@ def __init__(self, status, raw): class ChatSSEAccumulator: """聚合 Chat SSE，拒绝错误事件、空输出和无结束标记的残流。""" - def __init__(self, *, collect=True): + def __init__(self, *, collect=True, max_collect_bytes: int = 0): self.collect = collect + self.max_collect_bytes = max(0, int(max_collect_bytes or 0)) + self.collected_bytes = 0 self.content = [] self.reasoning = [] self.refusal = [] @@ -92,12 +94,11 @@ def _consume_chunk(self, chunk): self.saw_output = True if "tool_calls" in delta and not isinstance(delta["tool_calls"], list): raise ValueError("tool_calls") - if self.collect and delta.get("content"): - self.content.append(delta["content"]) - if self.collect and delta.get("reasoning_content"): - self.reasoning.append(delta["reasoning_content"]) - if self.collect and delta.get("refusal"): - self.refusal.append(delta["refusal"]) + if self.collect: + for key in ("content", "reasoning_content", "refusal"): + if delta.get(key): + getattr(self, key if key != "reasoning_content" else "reasoning").append(delta[key]) + self._charge(len(delta[key].encode("utf-8"))) for tool in delta.get("tool_calls") or []: if not isinstance(tool, dict): raise ValueError("tool") @@ -118,9 +119,21 @@ def _consume_chunk(self, chunk): self.saw_output = True slot["name"] = function.get("name") or slot["name"] if self.collect: - slot["arguments"] += function.get("arguments") or "" + piece = function.get("arguments") or "" + slot["arguments"] += piece + self._charge(len(piece.encode("utf-8"))) self.filter_detector.feed(delta, choice.get("finish_reason")) + def _charge(self, size: int): + """聚合收集总字节预算：超限即失败，不把无界输出缓存在内存里。""" + if not self.collect or not self.max_collect_bytes: + return + self.collected_bytes += size + if self.collected_bytes > self.max_collect_bytes: + raise UpstreamResponseError(502, json.dumps({"error": { + "message": f"upstream response exceeds the {self.max_collect_bytes}-byte collection budget", + "type": "upstream_error", "code": "response_too_large"}}).encode()) + def result(self): if not self.saw_choice or not (self.done or self.finish_reason): raise httpx.RemoteProtocolError("Upstream SSE ended without a completion marker") @@ -141,6 +154,21 @@ def result(self): "usage": self.usage, "model": self.model} +ERROR_BODY_LIMIT = 4 * 1024 * 1024 # 错误响应读取上限：错误页不应撑爆内存 + + +async def read_bounded_error(response, limit: int = ERROR_BODY_LIMIT) -> bytes: + """错误体有界读取：超限即截断，不再整段 aread。""" + if limit <= 0: + return b"" + buf = bytearray() + async for chunk in response.aiter_bytes(): + buf.extend(chunk[:limit - len(buf)]) + if len(buf) >= limit: + break + return bytes(buf) + + @asynccontextmanager async def open_backend_stream(url, headers, body, *, read_timeout=300, on_retry=None): """只重试一次建连失败，其他错误交给调用方按协议返回。""" diff --git a/converter.py b/converter.py index 91c16bd..40f96dc 100644 --- a/converter.py +++ b/converter.py @@ -43,6 +43,7 @@ from fastapi import FastAPI, Header, HTTPException, Request from fastapi.exception_handlers import http_exception_handler as _default_http_exception_handler from fastapi.responses import JSONResponse, StreamingResponse +from starlette.concurrency import run_in_threadpool import uvicorn try: @@ -71,7 +72,8 @@ def desensitize_body(body, roles=("system",), desensitize_harness_user=False, observe_attempt, observe_failure) from app.credential_io import (CredentialFileError, read_import_file, atomic_write_credential, credential_file_lock) -from app.upstream_io import ChatSSEAccumulator, UpstreamResponseError, open_backend_stream +from app.upstream_io import ChatSSEAccumulator, UpstreamResponseError, open_backend_stream, read_bounded_error +from app.inference_auth import require_api_key from app.content_filter import ContentFilterDetector, is_filter_error from app.request_limits import ImageLimitError, apply_image_policy from app.safe_logging import format_log_body, sanitize_log_text @@ -1239,12 +1241,10 @@ def _publish_usage_daily(pool, stale=()): used, count = 0.0, 0 partial = False newest = 0.0 - included = False stale_out = [] for cred_id, snap in accounts.items(): if cred_id not in enabled: continue - included = True site = snap.get("site") or "domestic" group = groups.setdefault(site, {"by_day": {}, "total_credits": 0.0, "requests": 0}) for day, models in (snap.get("by_day") or {}).items(): @@ -1260,11 +1260,12 @@ def _publish_usage_daily(pool, stale=()): newest = max(newest, float(snap.get("fetched_at") or 0)) if snap.get("partial"): partial = True - if cred_id in stale: + # 本轮失败的启用账号即使没有任何历史快照也必须可见，否则不完整聚合被当成精确值 + for cred_id in stale: + if cred_id in enabled: partial = True stale_out.append(Path(cred_id).name) - if not included: - return # 还没有任何成功快照：不覆盖已有视图 + # 无成功快照也发布完整性标记；fetched_at=0 使账务继续使用额度差回退。 for group in groups.values(): group["total_credits"] = round(group["total_credits"], 2) out = {"by_day": by_day, "groups": groups, "total_credits": round(used, 2), @@ -1273,7 +1274,7 @@ def _publish_usage_daily(pool, stale=()): out["stale_accounts"] = sorted(stale_out) CONFIG["usage_daily"] = out _log(f"[usage] 明细已同步: {count} 请求 / {used:.2f} credits" - + (f" | {len(stale_out)} 账号保留历史快照" if stale_out else "")) + + (f" | {len(stale_out)} 账号同步失败" if stale_out else "")) def _housekeep_once(pool: CredentialPool, ledger, *, pending_only=False): @@ -1373,7 +1374,9 @@ async def _protocol_http_exception(request: Request, exc: HTTPException): if path.startswith("/v1/messages"): # Anthropic：{"type": "error", "error": {...}}；上游业务 code 原样保留，客户端仍可识别 content_filter etype = _ANTHROPIC_ERROR_TYPES.get(str(err.get("type") or "")) - if etype is None: + if exc.status_code == 404: + etype = "not_found_error" # Anthropic 约定：404 恒为 not_found_error + elif etype is None: etype = "api_error" if exc.status_code >= 500 else "invalid_request_error" error_obj = {**err, "type": etype, "message": message} # code/param/image_count 等结构化字段原样保留 return JSONResponse({"type": "error", "error": error_obj}, @@ -1392,6 +1395,8 @@ async def _protocol_http_exception(request: Request, exc: HTTPException): "model_guard": True, # 表外模型本地拦截，不转发上游 "max_images": 16, "image_policy": "truncate", "max_request_bytes": 32 * 1024 * 1024, "log_body_limit": 65536, + "max_inbound_bytes": 64 * 1024 * 1024, + "max_collect_bytes": 8 * 1024 * 1024, "max_concurrent": 64, "usage_daily": None, # 官方用量聚合视图（日期×模型 credit），供 billing/usage 出 daily_costs "usage_daily_accounts": None, # 按账号的用量快照；单账号失败不丢历史 "credit_price_cny": None, "credit_price_usd": None, "usd_rate": None, @@ -1464,16 +1469,7 @@ def _truncate(s: str, n: int = 80) -> str: def _check_auth(authorization: Optional[str], x_api_key: Optional[str]): - key = CONFIG["api_key"] - if not key: - return - token = "" - if isinstance(authorization, str) and authorization.startswith("Bearer "): - token = authorization[7:].strip() - if not token and isinstance(x_api_key, str): - token = x_api_key - if not secrets.compare_digest(token.encode(), key.encode()): - raise HTTPException(status_code=401, detail={"error": {"message": "invalid api key", "type": "auth_error"}}) + require_api_key(CONFIG["api_key"], authorization, x_api_key) def _check_admin_auth(authorization: Optional[str], x_api_key: Optional[str]): @@ -1626,13 +1622,15 @@ async def admin_add_credential(request: Request, import_dir = Path(os.environ.get("CODEBUDDY_IMPORT_DIR") or dst_dir / "imports") try: name, content = read_import_file(import_dir, body.get("path")) - cred_data = json.loads(content.decode("utf-8")) + cred_data = auth_oauth.loads_strict(content.decode("utf-8")) src_uid, verr = auth_oauth.validate_cred_data(cred_data) if verr: raise CredentialFileError("凭据格式或站点校验失败") if (not isinstance(cred_data.get("account") or {}, dict) or not isinstance(cred_data["auth"].get("expiresAt", 0), (int, float))): raise CredentialFileError("凭据账号或过期时间格式无效") + # 与上传路径一致：落盘前折叠 token 别名为官方字段名 + content = json.dumps(auth_oauth.normalize_cred_data(cred_data), ensure_ascii=False).encode("utf-8") except CredentialFileError: raise HTTPException(status_code=400, detail={"error": {"message": "凭据文件不符合导入要求", "type": "invalid_request_error"}}) from None except (ValueError, UnicodeError, RecursionError): @@ -1840,6 +1838,9 @@ def billing_subscription(authorization: Optional[str] = Header(default=None), "codebuddy_balance_usd": t["remaining_usd"], "codebuddy_balance_cny": t["remaining_cny"], "codebuddy_sites": t["groups"], + # 余额/用量不完整（分页到顶或账号同步失败）时调用方必须能看到 + "codebuddy_partial": t["partial"], + **({"codebuddy_stale_accounts": stale} if (stale := (CONFIG.get("usage_daily") or {}).get("stale_accounts")) else {}), } @@ -2297,7 +2298,8 @@ async def chat_completions(request: Request, _log(f"[{rid}] ▶ REQUEST {model_name} | stream={client_wants_stream} | msgs={len(messages)}" + (f" | tools={tool_names}" if tool_names else "") + (f" | last_user={_truncate(last_user, 60)!r}" if last_user else "")) - body, cred, headers, url = _route_chat(payload, body, rid) + # 凭据选择/到期刷新持线程锁与文件锁并可能同步访问网络：放到受限线程池，不占事件循环 + body, cred, headers, url = await run_in_threadpool(_route_chat, payload, body, rid) _log_json(f"[{rid}] REQUEST BODY (发往后端，预览)", body) t0 = time.time() @@ -2427,7 +2429,7 @@ def _tool_calls_healthy(tool_calls, body: dict | None = None) -> bool: def _merge_chat_sse_text(text: str) -> dict: """文本路径与异步流路径使用同一聚合器。""" - accumulator = ChatSSEAccumulator() + accumulator = ChatSSEAccumulator(max_collect_bytes=CONFIG.get("max_collect_bytes", 0)) for line in text.splitlines(): accumulator.feed_line(line) return accumulator.result() @@ -2545,11 +2547,11 @@ async def _fetch_checked_chat(url, headers, body, model_name, rid, cred=None, *, tool_attempt = 0 filter_retried = False while True: - accumulator = ChatSSEAccumulator() + accumulator = ChatSSEAccumulator(max_collect_bytes=CONFIG.get("max_collect_bytes", 0)) rejection = None async with _backend_stream(url, headers, body, rid=rid, model_name=model_name) as response: if response.status_code != 200: - _check_upstream_status(response.status_code, await response.aread(), cred, body.get("model")) + _check_upstream_status(response.status_code, await read_bounded_error(response), cred, body.get("model")) else: _note_cred_model_ok(cred, body.get("model")) try: @@ -2588,6 +2590,11 @@ async def _fetch_checked_chat(url, headers, body, model_name, rid, cred=None, *, # 审核拒绝不是工具损坏，不因 required 工具选择而重复生成。 budget = CONFIG.get("tool_call_max_retry", _TOOL_CALL_MAX_RETRY) if detector.detected or not body.get("tools") or tool_attempt >= budget: + if not detector.detected and body.get("tools"): + # 耗尽预算的末次生成同样消耗额度：记入 attempts 再报错 + exhausted = result.get("usage") or {} + observe_attempt("tool_args_exhausted", attempt=tool_attempt, max_attempts=budget, + total_tokens=exhausted.get("total_tokens")) raise UpstreamResponseError(502, b"Invalid upstream tool_calls after retries") tool_attempt += 1 # 被丢弃的这次生成也是真实消耗：连同序号记进 attempts，账务不再只看见最后一次 @@ -2610,7 +2617,7 @@ async def _chat_sse_lines(url, headers, body, model_name, t0, rid, cred=None, *, budget = CONFIG["log_body_limit"] if CONFIG.get("log_path") else 0 async with _backend_stream(url, headers, body, rid=rid, model_name=model_name) as response: if response.status_code != 200: - _check_upstream_status(response.status_code, await response.aread(), cred, body.get("model")) + _check_upstream_status(response.status_code, await read_bounded_error(response), cred, body.get("model")) else: _note_cred_model_ok(cred, body.get("model")) async for line in response.aiter_lines(): @@ -2727,7 +2734,8 @@ async def create_response(request: Request, f"| dropped_harness={projection_stats.get('dropped_harness_messages', 0)} " f"| anchor_user={projection_stats.get('anchor_user_preserved', False)}" ) - chat_body, cred, headers, url = _route_chat(payload, chat_body, rid) + # 同上：凭据选择/刷新是阻塞操作，移出事件循环 + chat_body, cred, headers, url = await run_in_threadpool(_route_chat, payload, chat_body, rid) _log_json(f"[{rid}] RESPONSES → CHAT BODY (预览)", chat_body) t0 = time.time() @@ -2821,7 +2829,8 @@ async def create_message(request: Request, chat_messages = chat_body.get("messages", []) rid = os.urandom(4).hex() _log(f"[{rid}] ▶ ANTHROPIC {model_name} | msgs={len(chat_messages)} | anthropic_msgs={len(messages)}") - chat_body, cred, headers, url = _route_chat(payload, chat_body, rid) + # 同上：凭据选择/刷新是阻塞操作，移出事件循环 + chat_body, cred, headers, url = await run_in_threadpool(_route_chat, payload, chat_body, rid) _log_json(f"[{rid}] ANTHROPIC → CHAT BODY (预览)", chat_body) t0 = time.time() @@ -3027,6 +3036,15 @@ def main(): ap.add_argument("--max-request-bytes", type=_positive_int, metavar="BYTES", default=os.environ.get("CODEBUDDY2API_MAX_REQUEST_BYTES", str(32 * 1024 * 1024)), help="图片处理与适配后请求体的字节上限，默认 32 MiB") + ap.add_argument("--max-inbound-bytes", type=_positive_int, metavar="BYTES", + default=os.environ.get("CODEBUDDY2API_MAX_INBOUND_BYTES", str(64 * 1024 * 1024)), + help="入站原始请求体字节上限（解析前生效，含 chunked），默认 64 MiB") + ap.add_argument("--max-collect-bytes", type=_nonnegative_int, metavar="BYTES", + default=os.environ.get("CODEBUDDY2API_MAX_COLLECT_BYTES", str(8 * 1024 * 1024)), + help="聚合路径输出收集总字节上限（正文+思考+工具参数），默认 8 MiB；0 不限制") + ap.add_argument("--max-concurrent", type=_nonnegative_int, metavar="N", + default=os.environ.get("CODEBUDDY2API_MAX_CONCURRENT", "64"), + help="推理端点并发上限（超出立即 503），默认 64；0 不限制") ap.add_argument("--log-body-limit", type=_nonnegative_int, metavar="BYTES", default=os.environ.get("CODEBUDDY2API_LOG_BODY_LIMIT", "65536"), help="每条正文日志的预览字节上限，默认 64 KiB；0 只记录摘要") @@ -3043,7 +3061,7 @@ def main(): return login(site=args.site, open_browser=not args.no_browser) for key in ("max_images", "image_policy", "max_request_bytes", "log_body_limit", "auto_trial", - "tool_call_max_retry"): + "tool_call_max_retry", "max_inbound_bytes", "max_collect_bytes", "max_concurrent"): CONFIG[key] = getattr(args, key) CONFIG["api_key"] = args.api_key CONFIG["desensitize"] = args.desensitize @@ -3056,6 +3074,12 @@ def main(): CONFIG["log_path"] = args.log if args.log else os.environ.get("CODEBUDDY2API_LOG") from app import runtime_management runtime_management.initialize(sys.modules[__name__], args) + # 持久化设置解析后再核对实际监听地址和生效 key，且必须早于凭据扫描、线程及监听。 + if (args.host not in ("127.0.0.1", "::1", "localhost") and not CONFIG.get("api_key") + and os.environ.get("CODEBUDDY2API_ALLOW_OPEN_NOAUTH", "").lower() not in ("1", "true", "yes")): + runtime_management.close(CONFIG) + ap.error("非回环绑定且未设置 API key 会匿名开放推理额度；" + "请设置 CODEBUDDY2API_KEY，或确知风险后以 CODEBUDDY2API_ALLOW_OPEN_NOAUTH=true 显式放行") files = [Path(p) for p in args.auth_file] if not files: seed_credentials() # 自管模式：启动时把桌面端缺失凭据复制进 auth/ diff --git a/docker-compose.yml b/docker-compose.yml index 50f830e..c0c2c2c 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -8,7 +8,7 @@ services: build: . container_name: codebuddy2api ports: - - "${CODEBUDDY2API_BIND:-0.0.0.0}:${CODEBUDDY2API_PORT:-8787}:8787" + - "${CODEBUDDY2API_BIND:-127.0.0.1}:${CODEBUDDY2API_PORT:-8787}:8787" volumes: # .info 凭证、control.sqlite3、logs.sqlite3 及 SQLite WAL/SHM 共用持久目录。 # SQLite 需要可写的本地文件系统；不要只挂载单个数据库文件。 @@ -24,6 +24,8 @@ services: CODEBUDDY2API_LOG_BODY_LIMIT: ${CODEBUDDY2API_LOG_BODY_LIMIT:-65536} CODEBUDDY2API_LOG: ${CODEBUDDY2API_LOG:-} CODEBUDDY2API_AUTO_TRIAL: ${CODEBUDDY2API_AUTO_TRIAL:-false} + # 容器内固定 0.0.0.0；对外暴露边界由上方 BIND 端口映射控制（默认仅回环） + CODEBUDDY2API_ALLOW_OPEN_NOAUTH: "true" CODEBUDDY_AUTH_DIR: /data/auth command: > python3 converter.py diff --git a/docs/advanced.md b/docs/advanced.md index f745fe1..1a71daf 100644 --- a/docs/advanced.md +++ b/docs/advanced.md @@ -30,6 +30,9 @@ Compose explicitly passes some environment variables and CLI flags, so deleting | `--max-images` | `16` | Total images per request; `0` permits no images | | `--image-policy` | `truncate` | Keep newest images; `error` rejects excess images with 413 | | `--tool-call-max-retry` | `3` | Extra generations after malformed tool calls (each consumes credits); `0` disables retries | +| `--max-inbound-bytes` | `67108864` | Raw body limit for generation and token-count POSTs, before parsing (chunked included); other routes are not buffered; 413 beyond it | +| `--max-collect-bytes` | `8388608` | Total collection budget for aggregated output (content + reasoning + tool arguments); `response_too_large` beyond it; `0` disables | +| `--max-concurrent` | `64` | Concurrency limit for the three generation endpoints only; excess requests get 503 with Retry-After; token counting is unaffected; `0` disables | | `--max-request-bytes` | `33554432` | Positive byte limit for the processed upstream JSON | | `--log-body-limit` | `65536` | Text-log body preview bytes; `0` logs summaries only, not the SQLite diagnostic budget | @@ -133,10 +136,17 @@ Credential domain / token issuer determine the product identity. Chat and refres - `/v1/messages/count_tokens` returns a character-based heuristic estimate for budgeting, not an exact count. - Text logs and SQLite auditing have separate budgets. Logs contain bounded, redacted previews, not complete original requests. Treat logs, credential exports and backups as private data. +## Deployment exposure and credential intake + +- The compose port mapping binds loopback by default (`CODEBUDDY2API_BIND` defaults to 127.0.0.1); after resolving CLI, environment and saved settings, a native non-loopback bind with an empty effective API key refuses to start unless `CODEBUDDY2API_ALLOW_OPEN_NOAUTH=true` is set explicitly. +- When a key is configured, generation and token-count POSTs verify request headers before buffering bodies or reserving inference capacity; invalid keys return 401 even while generation slots are full. Other routes retain their existing authentication and routing behavior. +- Credential imports/uploads persist the normalized form (token aliases folded into the canonical fields); strict JSON parsing rejects NaN/Infinity, and `expiresAt`/`lastRefreshTime` must be plausible finite millisecond timestamps. + ## Billing data integrity - Balances and usage are paginated in full; when a page cap is hit or an account's sync fails, responses carry `partial: true` (and `stale_accounts`) instead of pretending to be exact. - A failed account keeps its last good snapshot; HTTP 200 responses with a failing business code or missing structure are treated as errors and never overwrite history. +- If every account fails before a first snapshot, both billing endpoints still report partial data and stale accounts; quota-delta fallback remains in use. Existing account snapshots retain their original fetch time on failures. - daily_costs are priced per site per day at that site's price, not at one blended average. ## Troubleshooting and retries diff --git a/docs/advanced.zh-CN.md b/docs/advanced.zh-CN.md index 57bf7eb..ea75e03 100644 --- a/docs/advanced.zh-CN.md +++ b/docs/advanced.zh-CN.md @@ -30,6 +30,9 @@ Compose 会显式传入部分环境变量及 CLI 参数，删除 `.env` 中的 | `--max-images` | `16` | 单请求图片总数；`0` 不允许图片 | | `--image-policy` | `truncate` | 保留最新图片；设为 `error` 时超限返回 413 | | `--tool-call-max-retry` | `3` | 工具参数损坏时的额外生成上限（每次都消耗额度）；`0` 不重试 | +| `--max-inbound-bytes` | `67108864` | 生成及 token 估算 POST 的解析前原始字节上限（含 chunked），超限 413；其他路由不缓冲请求体 | +| `--max-collect-bytes` | `8388608` | 聚合路径输出收集总字节上限（正文+思考+工具参数），超限返回 `response_too_large`；`0` 不限制 | +| `--max-concurrent` | `64` | 仅限制三个生成端点；占满立即 503（含 Retry-After），不限制 token 估算；`0` 不限制 | | `--max-request-bytes` | `33554432` | 处理后的上游 JSON 字节上限，须为正整数 | | `--log-body-limit` | `65536` | 兼容文本日志正文预览字节；`0` 只记摘要，不控制 SQLite 诊断预算 | @@ -133,10 +136,17 @@ WebUI 可以直接上传文件；以下限制针对 `POST /admin/credentials` - `/v1/messages/count_tokens` 返回字符启发式估算值，仅作预算参考，不是精确计数。 - 兼容文本日志和 SQLite 审计使用独立预算；日志仅记录有界、脱敏预览，不是完整原始请求。日志、凭证导出和备份仍须按私有数据保管。 +## 部署暴露与凭据导入 + +- Compose 端口映射默认只绑回环（`CODEBUDDY2API_BIND` 默认 127.0.0.1）；原生运行在合并 CLI、环境和持久化设置后，若实际地址非回环且生效 key 为空则拒绝启动，须显式设 `CODEBUDDY2API_ALLOW_OPEN_NOAUTH=true` 放行。 +- 配置 key 时，生成及 token 估算 POST 在缓冲请求体、预留推理名额之前校验请求头；即使名额已满，无效 key 仍返回 401。其他路由保留原有鉴权和路由行为。 +- 凭据导入/上传在落盘前把 token 别名归一化为官方字段名；严格 JSON 解析拒绝 NaN/Infinity，`expiresAt`/`lastRefreshTime` 必须是合理的有限毫秒时间戳。 + ## 账务数据完整性 - 余额/用量来自官方接口的分页遍历；达到页数上限或任一账号同步失败时，响应带 `partial: true`（及 `stale_accounts` 列表），不伪装为精确全量。 - 单账号同步失败保留其上次成功快照；接口返回 HTTP 200 但业务码失败或结构缺失时按错误处理，不覆盖历史。 +- 首次同步全部失败时，两个账务端点仍标记数据不完整及失败账号，并保留额度差回退；已有账号快照在失败时不更新其成功时间。 - daily_costs 按各站（国内/国际）本站单价逐日折算，不再使用全局平均单价。 ## 故障与重试 diff --git a/tests/test_admin_api.py b/tests/test_admin_api.py index 716cebc..2317f15 100644 --- a/tests/test_admin_api.py +++ b/tests/test_admin_api.py @@ -401,6 +401,23 @@ def test_valid_upload_and_replace_confirmation(self): response = self.client.post("/admin/credentials/upload", headers=self.headers, json=payload) self.assertFalse(response.json()["results"][0]["ok"]) + def test_upload_normalizes_token_aliases_to_canonical_fields(self): + """只含 access_token/token 别名的凭据：落盘内容必须折叠为 accessToken，别名键移除。""" + data = self.credential() + auth = data.pop("auth") + data["auth"] = {**{k: v for k, v in auth.items() if k != "accessToken"}, + "access_token": auth["accessToken"], "token_type": "Bearer", + "refresh_token": "synthetic-refresh"} + payload = {"files": [{"name": "first.info", "content": json.dumps(data)}]} + response = self.client.post("/admin/credentials/upload", headers=self.headers, json=payload) + self.assertTrue(response.json()["results"][0]["ok"], response.text) + saved = json.loads(self.gateway._store_credential.call_args.args[2].decode("utf-8")) + self.assertEqual(saved["auth"]["accessToken"], "synthetic-token") + self.assertEqual(saved["auth"]["refreshToken"], "synthetic-refresh") + self.assertNotIn("access_token", saved["auth"]) + self.assertNotIn("refresh_token", saved["auth"]) + self.assertNotIn("token", saved["auth"]) + def test_export_only_selected_info_and_symlink_identity_protection(self): content = json.dumps(self.credential()) (self.root / "first.info").write_text(content) diff --git a/tests/test_anthropic_adapter.py b/tests/test_anthropic_adapter.py index d88ba30..eaa8715 100644 --- a/tests/test_anthropic_adapter.py +++ b/tests/test_anthropic_adapter.py @@ -449,6 +449,13 @@ def conv(is_error): assert conv(True).startswith("[tool execution failed]\nexit 1") assert conv(False) == "exit 1" assert conv(None) == "exit 1" + # 含图片的失败结果：标记为前置文本块，不做字符串拼接 + req = {"model": "auto", "max_tokens": 64, "messages": [{"role": "user", "content": [ + {"type": "tool_result", "tool_use_id": "toolu_2", "is_error": True, "content": [ + {"type": "image", "source": {"type": "url", "url": "https://synthetic.invalid/x.png"}}, + {"type": "text", "text": "boom"}]}]}]} + content = anthropic_request_to_chat(req)["messages"][0]["content"] + assert isinstance(content, list) and content[0] == {"type": "text", "text": "[tool execution failed]"} print("✅ test_tool_result_is_error_is_preserved") diff --git a/tests/test_auth_oauth.py b/tests/test_auth_oauth.py index 7098133..0e1c912 100644 --- a/tests/test_auth_oauth.py +++ b/tests/test_auth_oauth.py @@ -55,6 +55,29 @@ def test_validate_cred_data(): bad["auth"]["accessToken"] = "not-a-jwt" uid, err = validate_cred_data(bad) assert uid is None and "允许列表" in err + # 时间戳必须有限且合理：NaN/Infinity/bool/0/超范围都拒绝 + for field in ("expiresAt", "lastRefreshTime"): + for bad_ts in (float("nan"), float("inf"), float("-inf"), True, 0, -1, + 4102444800000, 99999999999999, 10**1000, -(10**1000)): + c = _cred() + c["auth"][field] = bad_ts + assert validate_cred_data(c)[1], (field, bad_ts) + for valid_ts in (1, 1893456000000, 4102444799999, 1893456000000.5): + c = _cred() + c["auth"][field] = valid_ts + assert validate_cred_data(c) == ("u1", None) + c = _cred() + c["auth"]["expiresAt"] = 1893456000000 # 2030-01-01 毫秒 + assert validate_cred_data(c) == ("u1", None) + # 严格解析拒绝非标准常量 + from app.auth_oauth import loads_strict + for text in ('{"a": NaN}', '{"a": Infinity}', '{"a": -Infinity}'): + try: + loads_strict(text) + raise AssertionError(text) + except ValueError: + pass + assert loads_strict('{"a": 1.5}') == {"a": 1.5} print("✅ test_validate_cred_data") diff --git a/tests/test_credits.py b/tests/test_credits.py index 72e95b4..fee80ce 100644 --- a/tests/test_credits.py +++ b/tests/test_credits.py @@ -583,6 +583,28 @@ def test_fetch_credits_paginates_until_short_page(): lambda: credits.fetch_credits(token)) assert len(seen) == credits.CREDITS_MAX_PAGES assert result["partial"] is True + + # 第 2 页的瞬时空响应也要重试：不能在非首页把空页当作结束 + calls = {"n": 0} + sequence = [full_page, {"code": 0, "data": {"Response": {"Data": {"Accounts": []}}}}, short_page] + + class FlakyClient: + def __enter__(self): + return self + def __exit__(self, *a): + return False + def post(self, url, headers=None, json=None, timeout=None): + calls["n"] += 1 + payload = sequence[min(calls["n"] - 1, len(sequence) - 1)] + + class Resp: + status_code = 200 + def json(self): + return payload + return Resp() + + result = _with_client(FlakyClient, lambda: credits.fetch_credits(token)) + assert result["count"] == 101 and result["partial"] is False, result print("✅ test_fetch_credits_paginates_until_short_page") @@ -629,11 +651,53 @@ def fake_fetch(token, uid="", domain=""): credits.fetch_request_usage = fake_fetch converter.CONFIG.update(usage_daily=None, usage_daily_accounts=None, control_store=None) try: + from fastapi.testclient import TestClient + + def check_billing_stale(names): + with patch.dict(converter.CONFIG, {"api_key": "", "ledger": None}), \ + TestClient(converter.app) as client: + sub = client.get("/v1/dashboard/billing/subscription") + usage = client.get("/v1/dashboard/billing/usage") + assert sub.status_code == usage.status_code == 200 + assert sub.json()["codebuddy_partial"] is bool(names) + assert sub.json().get("codebuddy_stale_accounts", []) == names + assert usage.json().get("partial", False) is bool(names) + assert usage.json().get("stale_accounts", []) == names + + failing.update(snapshots) + for previous in (None, {"total_credits": 999, "fetched_at": 123, "partial": False}): + converter.CONFIG["usage_daily"] = previous + converter._sync_usage(pool) + view = converter.CONFIG["usage_daily"] + assert view["by_day"] == view["groups"] == {} + assert view["total_credits"] == view["requests"] == view["fetched_at"] == 0 + assert view["partial"] is True and view["stale_accounts"] == ["u1.info", "u2.info"] + check_billing_stale(["u1.info", "u2.info"]) + assert converter._billing_totals()["used_source"] == "quota_delta" + failing.clear() + + # 首轮即有账号失败且无任何历史快照：也必须标 stale/partial，不能装作精确 + failing.add("token-u2") + converter._sync_usage(pool) + view = converter.CONFIG["usage_daily"] + assert view["total_credits"] == 10.0 and view["requests"] == 1 + assert view["partial"] is True and view["stale_accounts"] == ["u2.info"] + + failing.clear() converter._sync_usage(pool) view = converter.CONFIG["usage_daily"] assert view["total_credits"] == 30.0 and view["requests"] == 3 assert view["partial"] is False and "stale_accounts" not in view + failing.update(snapshots) + last_good = dict(view) + converter._sync_usage(pool) + view = converter.CONFIG["usage_daily"] + for key in ("by_day", "groups", "total_credits", "requests", "fetched_at"): + assert view[key] == last_good[key] + check_billing_stale(["u1.info", "u2.info"]) + failing.clear() + failing.add("token-u2") converter._sync_usage(pool) view = converter.CONFIG["usage_daily"] @@ -652,6 +716,16 @@ def fake_fetch(token, uid="", domain=""): converter._sync_usage(pool) view = converter.CONFIG["usage_daily"] assert view["total_credits"] == 10.0 # 凭证删除后其快照不再计入 + + with patch.object(converter.model_policy, "credential_enabled", return_value=False): + converter._sync_usage(pool) + assert converter.CONFIG["usage_daily"]["total_credits"] == 0 + check_billing_stale([]) + paths[0].unlink() + pool.prune() + converter._sync_usage(pool) + assert converter.CONFIG["usage_daily"]["groups"] == {} + check_billing_stale([]) finally: credits.fetch_request_usage = orig_fetch converter.CONFIG["usage_daily"], converter.CONFIG["usage_daily_accounts"], \ @@ -726,6 +800,7 @@ def test_billing_balance_identity(): # 端点级恒等式：客户端按 hard_limit_usd − total_usage/100 算出的正是真实剩余 sub = converter.billing_subscription(None, None) usage = converter.billing_usage(None, None, None, None) + assert sub["codebuddy_partial"] is False # 数据完整时显式 False assert abs(sub["hard_limit_usd"] - usage["total_usage"] / 100 - t["remaining_usd"]) < 0.01 assert sub["codebuddy_credits_remaining"] == 1000.0 assert sub["plan"]["title"].startswith("CodeBuddy Credits") diff --git a/tests/test_inference_admission.py b/tests/test_inference_admission.py new file mode 100644 index 0000000..82ced46 --- /dev/null +++ b/tests/test_inference_admission.py @@ -0,0 +1,163 @@ +"""Admission regressions using the production middleware installation order.""" + +import asyncio +import json +import sys +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock, patch + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from fastapi import FastAPI, Request + +from app import runtime_management +from app.inbound_limits import ConcurrencyLimitMiddleware + + +ROUTES = ("/v1/chat/completions", "/v1/responses", "/v1/messages") + + +class InferenceAdmissionTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.config = {"api_key": "synthetic-key", "max_concurrent": 1, + "max_inbound_bytes": 1024, "management": Mock()} + app = FastAPI() + + async def endpoint(request: Request): + return {"size": len(await request.body())} + + @app.get("/v1/models") + async def models(request: Request): + from app.inference_auth import require_api_key + require_api_key(self.config["api_key"], request.headers.get("authorization")) + return {"object": "list", "data": []} + + for path in (*ROUTES, "/v1/messages/count_tokens", "/admin/test"): + app.add_api_route(path, endpoint, methods=["POST"]) + gateway = SimpleNamespace(app=app, CONFIG=self.config, __file__=str(Path(__file__).parent.parent / "converter.py")) + with patch.object(runtime_management, "install_admin"), patch.object(runtime_management, "install_pages"): + runtime_management.install(gateway) + self.app = app.build_middleware_stack() + layer = self.app + while not isinstance(layer, ConcurrencyLimitMiddleware): + layer = layer.app + self.gate = layer + + async def request(self, path, headers=(), *, forbid_receive=False, method="POST"): + sent = [] + replayed = False + + async def receive(): + nonlocal replayed + if forbid_receive: + self.fail("rejected headers must not read or wait for a request body") + if replayed: + await asyncio.Event().wait() + replayed = True + return {"type": "http.request", "body": b"{}", "more_body": False} + + async def send(message): + sent.append(message) + + scope = {"type": "http", "method": method, "path": path, "raw_path": path.encode(), + "query_string": b"", "headers": list(headers), "scheme": "http", "http_version": "1.1", + "server": ("test", 80), "client": ("127.0.0.1", 1234), + "asgi": {"version": "3.0", "spec_version": "2.3"}} + await asyncio.wait_for(self.app(scope, receive, send), 2) + start = next(message for message in sent if message["type"] == "http.response.start") + body = b"".join(message.get("body", b"") for message in sent) + content_type = dict(start.get("headers", [])).get(b"content-type", b"") + payload = json.loads(body) if b"application/json" in content_type else body.decode() + return start["status"], payload + + def assert_unauthorized(self, path, status, body): + self.assertEqual(status, 401, body) + self.assertEqual(body["error"]["message"], "invalid api key") + if path.startswith("/v1/messages"): + self.assertEqual(body["type"], "error") + self.assertEqual(body["error"]["type"], "authentication_error") + else: + self.assertNotIn("type", body) + self.assertEqual(body["error"]["type"], "auth_error") + self.assertNotIn("synthetic-key", json.dumps(body)) + + async def test_unauthorized_bodies_never_reserve_or_consume_inference_capacity(self): + invalid = ((), ((b"authorization", b"Bearer wrong"),), ((b"x-api-key", b"wrong"),), + ((b"cookie", b"codebuddy_admin=synthetic-cookie"),)) + for path in (*ROUTES, "/v1/messages/count_tokens"): + for headers in invalid: + with self.subTest(path=path, headers=headers): + status, body = await self.request(path, headers, forbid_receive=True) + self.assert_unauthorized(path, status, body) + self.assertIsNone(self.gate._semaphore) + status, body = await self.request(ROUTES[0], ((b"authorization", b"Bearer synthetic-key"),)) + self.assertEqual(status, 200, body) + + async def test_full_gate_still_authenticates_before_returning_capacity_errors(self): + gate = self.gate._gate() + await gate.acquire() + try: + for path in ROUTES: + with self.subTest(path=path): + status, body = await self.request(path, forbid_receive=True) + self.assert_unauthorized(path, status, body) + status, body = await self.request(path, ((b"x-api-key", b"synthetic-key"),), forbid_receive=True) + self.assertEqual(status, 503, body) + self.assertEqual(body["error"]["code"], "concurrency_limit") + status, body = await self.request("/v1/messages/count_tokens", ((b"x-api-key", b"synthetic-key"),)) + self.assertEqual(status, 200, body) + finally: + gate.release() + + async def test_current_key_and_header_precedence_are_preserved(self): + valid = (((b"authorization", b"Bearer synthetic-key"),), ((b"x-api-key", b"synthetic-key"),), + ((b"authorization", b"Bearer "), (b"x-api-key", b"synthetic-key"))) + for path in ROUTES: + for headers in valid: + with self.subTest(path=path, headers=headers): + status, body = await self.request(path, headers) + self.assertEqual(status, 200, body) + status, body = await self.request(path, ((b"authorization", b"Bearer wrong"), + (b"x-api-key", b"synthetic-key")), forbid_receive=True) + self.assert_unauthorized(path, status, body) + self.config["api_key"] = "replacement-key" + status, body = await self.request(ROUTES[0], ((b"x-api-key", b"synthetic-key"),), forbid_receive=True) + self.assert_unauthorized(ROUTES[0], status, body) + status, body = await self.request(ROUTES[0], ((b"x-api-key", b"replacement-key"),)) + self.assertEqual(status, 200, body) + + async def test_unknown_routes_and_methods_preserve_routing_errors(self): + for path, method, expected in (("/v1/missing", "GET", 404), + ("/v1/missing", "POST", 404), + ("/v1/messages/unknown", "POST", 404), + ("/v1/messages", "GET", 405)): + with self.subTest(path=path, method=method): + status, body = await self.request(path, method=method, forbid_receive=True) + self.assertEqual(status, expected, body) + + + async def test_bodyless_models_reaches_route_auth_without_buffering(self): + for headers, expected in (((), 401), (((b"authorization", b"Bearer synthetic-key"),), 200)): + with self.subTest(status=expected): + status, body = await self.request("/v1/models", headers, method="GET", forbid_receive=True) + self.assertEqual(status, expected, body) + self.assertIsNone(self.gate._semaphore) + + + async def test_disabled_gate_still_authenticates_and_empty_key_remains_optional(self): + self.config["max_concurrent"] = 0 + status, body = await self.request(ROUTES[0], forbid_receive=True) + self.assert_unauthorized(ROUTES[0], status, body) + self.config["api_key"] = "" + status, body = await self.request(ROUTES[0]) + self.assertEqual(status, 200, body) + self.assertIsNone(self.gate._semaphore) + self.config["api_key"] = "synthetic-key" + status, body = await self.request("/admin/test") + self.assertEqual(status, 200, body) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/tests/test_runtime_endpoints.py b/tests/test_runtime_endpoints.py index f53ad64..4267ae8 100644 --- a/tests/test_runtime_endpoints.py +++ b/tests/test_runtime_endpoints.py @@ -160,6 +160,17 @@ def test_inference_errors_follow_the_client_protocol_shape(self): body = response.json() self.assertEqual(body["type"], "error") self.assertEqual(body["error"]["type"], "authentication_error") + # Anthropic 约定 404 → not_found_error，即使内层写的是 invalid_request_error + converter.CONFIG["model_guard"] = True + try: + missing = self.client.post("/v1/messages", json={ + "model": "no-such-model", "max_tokens": 64, + "messages": [{"role": "user", "content": "hi"}]}, + headers={"Authorization": "Bearer secret"}) + self.assertEqual(missing.status_code, 404, missing.text) + self.assertEqual(missing.json()["error"]["type"], "not_found_error") + finally: + converter.CONFIG["model_guard"] = False response = self.client.get("/admin/credentials") self.assertEqual(response.status_code, 401, response.text) self.assertIn("detail", response.json()) @@ -221,9 +232,23 @@ def flaky(request): response = self.client.post(ROUTES[0], json=nonstream) self.assertEqual(response.status_code, 502, response.text) self.assertEqual(len(self.requests), 1) # 预算 0：不重试 + # 耗尽预算的末次生成也必须带着用量出现在 attempts 里 + exhausted = [kw for stage, kw in attempts if stage == "tool_args_exhausted"] + self.assertEqual(len(exhausted), 1) + self.assertIn("total_tokens", exhausted[0]) finally: converter.CONFIG["tool_call_max_retry"] = 3 + def test_credential_selection_runs_off_the_event_loop(self): + """_route_chat 内含线程锁/文件锁/同步刷新：三个端点都必须经线程池调用它。""" + import inspect + import re + src = inspect.getsource(converter) + direct = re.findall(r"^\s+(?:body|chat_body), cred, headers, url = _route_chat\(", src, re.M) + pooled = re.findall(r"await run_in_threadpool\(_route_chat", src) + self.assertEqual(direct, []) + self.assertEqual(len(pooled), 3) + def test_tool_metadata_policy_reaches_all_protocols(self): description = "Read sandbox data without destructive changes." schema = {"type": "object", "title": "Lookup inputs", "properties": { @@ -555,10 +580,214 @@ def handler(request): self.assertEqual(calls, 1) +class InboundBodyLimitTests(unittest.TestCase): + """入站原始字节限量：解析前 413，chunked 同样受限，/admin 不受影响。""" + + def _app(self, limit): + from app.inbound_limits import InboundBodyLimitMiddleware + from fastapi import Request + app = FastAPI() + + @app.post("/v1/chat/completions") + async def inference(request: Request): + return {"size": len(await request.body())} + + @app.post("/admin/x") + async def admin(request: Request): + return {"size": len(await request.body())} + + app.add_middleware(InboundBodyLimitMiddleware, config={"max_inbound_bytes": limit}) + return TestClient(app) + + def test_over_limit_rejected_before_parsing_and_under_limit_passes(self): + client = self._app(1024) + ok = client.post("/v1/chat/completions", json={"messages": []}) + self.assertEqual(ok.status_code, 200, ok.text) + big = client.post("/v1/chat/completions", content=b"x" * 2048, + headers={"Content-Type": "application/json"}) + self.assertEqual(big.status_code, 413) + self.assertEqual(big.json()["error"]["code"], "request_too_large") + self.assertNotIn("detail", big.json()) + big_admin = client.post("/admin/x", content=b"x" * 2048) + self.assertEqual(big_admin.status_code, 200) # 管理路由不在此限量范围 + + def test_chunked_body_is_counted_and_rejected(self): + from app.inbound_limits import InboundBodyLimitMiddleware + reached = [] + + async def app(scope, receive, send): + reached.append(True) + + middleware = InboundBodyLimitMiddleware(app, {"max_inbound_bytes": 10}) + chunks = [{"type": "http.request", "body": b"12345678", "more_body": True}, + {"type": "http.request", "body": b"9" * 8, "more_body": False}] + sent = [] + + async def receive(): + return chunks.pop(0) if chunks else {"type": "http.disconnect"} + + async def send(message): + sent.append(message) + + import asyncio + asyncio.run(middleware({"type": "http", "method": "POST", "path": "/v1/chat/completions"}, receive, send)) + self.assertFalse(reached) # 超限请求不进入下游 + self.assertEqual(sent[0]["status"], 413) + + +class InboundStreamingTests(unittest.IsolatedAsyncioTestCase): + async def exercise_stream(self, path, disconnect=False): + from app.inbound_limits import ConcurrencyLimitMiddleware, InboundBodyLimitMiddleware + from starlette.responses import StreamingResponse + + disconnected = asyncio.Event() + closed = asyncio.Event() + chunks = [{"type": "http.request", "body": b"{", "more_body": True}, + {"type": "http.request", "body": b"}", "more_body": False}] + sent = [] + + async def receive(): + if chunks: + return chunks.pop(0) + await disconnected.wait() + return {"type": "http.disconnect"} + + async def send(message): + sent.append(message) + if disconnect and message["type"] == "http.response.body" and message.get("body"): + disconnected.set() + + async def stream(): + try: + yield b"data: first\n\n" + if disconnect: + await asyncio.Event().wait() + else: + await asyncio.sleep(0) + yield b"data: [DONE]\n\n" + finally: + closed.set() + + async def app(scope, receive, send): + request = await receive() + self.assertEqual(request["body"], b"{}") + self.assertFalse(request["more_body"]) + await StreamingResponse(stream(), media_type="text/event-stream")(scope, receive, send) + + config = {"max_inbound_bytes": 1024, "max_concurrent": 1} + middleware = ConcurrencyLimitMiddleware(InboundBodyLimitMiddleware(app, config), config) + scope = {"type": "http", "method": "POST", "path": path, + "asgi": {"version": "3.0", "spec_version": "2.3"}} + await asyncio.wait_for(middleware(scope, receive, send), 2) + self.assertTrue(closed.is_set()) + self.assertFalse(middleware._gate().locked()) + body = b"".join(m.get("body", b"") for m in sent) + self.assertIn(b"data: first", body) + if disconnect: + self.assertNotIn(b"[DONE]", body) + else: + self.assertIn(b"[DONE]", body) + self.assertFalse(sent[-1].get("more_body", False)) + + async def test_buffered_requests_keep_streaming_until_completion(self): + for path in ROUTES: + with self.subTest(path=path): + await self.exercise_stream(path) + + async def test_real_disconnect_closes_the_stream_and_releases_capacity(self): + await self.exercise_stream("/v1/messages", disconnect=True) + + + +class ConcurrencyLimitTests(unittest.IsolatedAsyncioTestCase): + """并发上限：名额占满立即 503（含 Retry-After），释放后恢复。""" + + async def test_full_gate_returns_503_and_recovers(self): + from app.inbound_limits import ConcurrencyLimitMiddleware + entered = asyncio.Event() + release = asyncio.Event() + + async def slow_app(scope, receive, send): + entered.set() + await release.wait() + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"{}"}) + + mw = ConcurrencyLimitMiddleware(slow_app, {"max_concurrent": 1}) + scope = {"type": "http", "method": "POST", "path": "/v1/chat/completions"} + + async def receive(): + return {"type": "http.request", "body": b"{}", "more_body": False} + + sent = [] + + async def send(message): + sent.append(message) + + first = asyncio.create_task(mw(scope, receive, send)) + try: + await asyncio.wait_for(entered.wait(), 2) + for path in ROUTES: + with self.subTest(path=path): + sent.clear() + await mw({**scope, "path": path}, receive, send) + self.assertEqual(sent[0]["status"], 503) + headers = dict(sent[0]["headers"]) + self.assertEqual(headers[b"retry-after"], b"3") + body = sent[1]["body"] + self.assertEqual(int(headers[b"content-length"]), len(body)) + payload = json.loads(body) + self.assertEqual(payload["error"]["code"], "concurrency_limit") + if path == "/v1/messages": + self.assertEqual(payload["type"], "error") + self.assertEqual(payload["error"]["type"], "api_error") + else: + self.assertNotIn("type", payload) + self.assertEqual(payload["error"]["type"], "rate_limit_error") + finally: + release.set() + await asyncio.wait_for(first, 2) + sent.clear() + await mw(scope, receive, send) + self.assertEqual(sent[0]["status"], 200) + + +class AuxiliaryCapacityTests(unittest.IsolatedAsyncioTestCase): + async def test_saturated_generation_gate_does_not_block_token_counting(self): + from app.inbound_limits import ConcurrencyLimitMiddleware + + middleware = ConcurrencyLimitMiddleware(converter.app, {"max_concurrent": 1}) + gate = middleware._gate() + await gate.acquire() + try: + with patch.dict(converter.CONFIG, {"api_key": ""}): + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=middleware), + base_url="http://test") as client: + response = await client.post("/v1/messages/count_tokens", json={ + "model": "auto", "messages": [{"role": "user", "content": "hello"}]}) + self.assertEqual(response.status_code, 200, response.text) + self.assertGreater(response.json()["input_tokens"], 0) + unknown = await client.post("/v1/messages/unknown", json={}) + self.assertEqual(unknown.status_code, 404) + wrong_method = await client.get("/v1/messages") + self.assertEqual(wrong_method.status_code, 405) + self.assertTrue(gate.locked()) + finally: + gate.release() + + + class ConfigurationTests(unittest.TestCase): - def configure(self, env=None, flags=(), invalid=False): + def configure(self, env=None, flags=(), invalid=False, stored=None, expected_host=None): with contextlib.ExitStack() as stack: directory = stack.enter_context(tempfile.TemporaryDirectory()) + if stored: + from app.control_store import ControlStore + store = ControlStore(Path(directory) / "control.sqlite3") + try: + store.update_settings(stored, store.snapshot()["revision"]) + finally: + store.close() stack.enter_context(patch.object(converter, "managed_auth_dir", return_value=Path(directory))) stack.enter_context(patch.object(converter, "app", FastAPI())) stack.enter_context(patch.dict(os.environ, env or {}, clear=True)) @@ -570,15 +799,21 @@ def configure(self, env=None, flags=(), invalid=False): stack.enter_context(patch.object(converter, "credits_mod", None)) stack.enter_context(patch.object(converter.threading, "Thread")) server = stack.enter_context(patch.object(converter.uvicorn, "run")) + from app import runtime_management + close = stack.enter_context(patch.object(runtime_management, "close", wraps=runtime_management.close)) if invalid: with self.assertRaises(SystemExit) as caught: converter.main() self.assertEqual(caught.exception.code, 2) seed.assert_not_called() server.assert_not_called() + if stored: + close.assert_called_once_with(converter.CONFIG) return converter.main() server.assert_called_once() + if expected_host is not None: + self.assertEqual(server.call_args.kwargs["host"], expected_host) return {key: converter.CONFIG[key] for key in ( "max_images", "image_policy", "max_request_bytes", "log_body_limit", "admin_csrf", "keep_tool_metadata")} @@ -587,6 +822,27 @@ def test_defaults(self): "max_request_bytes": 33554432, "log_body_limit": 65536, "admin_csrf": True, "keep_tool_metadata": False}) + def test_open_binding_without_key_requires_explicit_opt_in(self): + # 非回环 + 空 key：默认拒启（SystemExit 2） + self.configure(flags=("--host", "0.0.0.0"), invalid=True) + # 显式放行环境变量后可启动 + self.configure(env={"CODEBUDDY2API_ALLOW_OPEN_NOAUTH": "true"}, flags=("--host", "0.0.0.0")) + # 非回环但设了 key：正常 + self.configure(env={"CODEBUDDY2API_KEY": "k"}, flags=("--host", "0.0.0.0")) + + def test_persisted_host_is_validated_after_configuration_resolution(self): + for host in ("0.0.0.0", "::"): + with self.subTest(host=host): + self.configure(stored={"host": host}, invalid=True) + self.configure(stored={"host": host}, env={"CODEBUDDY2API_KEY": "k"}, expected_host=host) + self.configure(stored={"host": host}, env={"CODEBUDDY2API_ALLOW_OPEN_NOAUTH": "true"}, + expected_host=host) + self.configure(stored={"host": "0.0.0.0"}, flags=("--host", "127.0.0.1"), expected_host="127.0.0.1") + self.configure(stored={"host": "127.0.0.1"}, flags=("--host", "0.0.0.0"), invalid=True) + self.configure(stored={"host": "0.0.0.0"}, env={"CODEBUDDY2API_KEY": ""}, + flags=("--api-key", "k"), expected_host="0.0.0.0") + + def test_environment_and_explicit_cli_precedence(self): env = {"CODEBUDDY2API_MAX_IMAGES": "8", "CODEBUDDY2API_IMAGE_POLICY": "error", "CODEBUDDY2API_MAX_REQUEST_BYTES": "100000", "CODEBUDDY2API_LOG_BODY_LIMIT": "0"} diff --git a/tests/test_security.py b/tests/test_security.py index 25b85f1..78c2bc7 100644 --- a/tests/test_security.py +++ b/tests/test_security.py @@ -209,6 +209,26 @@ def test_invalid_credentials_do_not_write_or_leak(self): self.assert_http(400, lambda: self.post(source.name)) self.assertFalse((self.auth / source.name).exists()) + def test_oversized_integer_timestamps_return_400_without_writes(self): + from fastapi.testclient import TestClient + + with TestClient(converter.app) as client, patch.object(converter, "_store_credential") as store: + for field in ("expiresAt", "lastRefreshTime"): + for value in (10**1000, -(10**1000)): + with self.subTest(field=field, negative=value < 0): + data = credential(token="synthetic-secret") + data["auth"][field] = value + source = self.source("invalid-time.info", data) + response = client.post("/admin/credentials", json={"path": source.name}, + headers={"Authorization": "Bearer synthetic-admin-key"}) + self.assertEqual(response.status_code, 400, response.text) + self.assertNotIn("synthetic-secret", response.text) + self.assertNotIn(str(self.root), response.text) + self.assertFalse((self.auth / source.name).exists()) + store.assert_not_called() + self.assertEqual(self.pool.entries(), []) + + def test_same_name_updates_pool(self): source = self.source() self.post(source.name) diff --git a/tests/test_upstream_io.py b/tests/test_upstream_io.py index b1c7611..09365a4 100644 --- a/tests/test_upstream_io.py +++ b/tests/test_upstream_io.py @@ -222,6 +222,70 @@ def test_refusal_text_and_content_filter_text_are_preserved(self): self.assertEqual(result["finish_reason"], "content_filter") +class OutputBudgetTests(unittest.TestCase): + """聚合收集预算与错误体有界读取。""" + + def test_collect_budget_aborts_oversized_aggregation(self): + acc = ChatSSEAccumulator(max_collect_bytes=10) # 两片各 8B，第二片超预算 + acc.feed_line('data: {"choices":[{"index":0,"delta":{"content":"12345678"}}]}') + with self.assertRaises(UpstreamResponseError) as caught: + acc.feed_line('data: {"choices":[{"index":0,"delta":{"content":"12345678"}}]}') + self.assertEqual(caught.exception.status, 502) + self.assertIn(b"response_too_large", caught.exception.raw) + # 预算内不受影响 + acc = ChatSSEAccumulator(max_collect_bytes=1024) + acc.feed_line('data: {"choices":[{"index":0,"delta":{"content":"ok"},"finish_reason":"stop"}]}') + acc.feed_line("data: [DONE]") + self.assertEqual(acc.result()["content"], "ok") + + def test_error_body_read_is_bounded(self): + import asyncio + + class BigError: + async def aiter_bytes(self): + for _ in range(8): + yield b"x" * 1024 * 1024 + + raw = asyncio.run(upstream_io.read_bounded_error(BigError(), limit=1024)) + self.assertEqual(len(raw), 1024) + + +class BoundedErrorReadTests(unittest.IsolatedAsyncioTestCase): + async def test_filling_the_limit_never_pulls_another_chunk_and_closes_response(self): + for chunks in ((b"abcdefgh",), (b"abcd", b"efgh"), (b"abcd", b"efgh-tail")): + with self.subTest(chunks=chunks): + class CappedStream(httpx.AsyncByteStream): + closed = False + + async def __aiter__(self): + for chunk in chunks: + yield chunk + raise AssertionError("reader waited for data after reaching its budget") + + async def aclose(self): + self.closed = True + + stream = CappedStream() + transport = httpx.MockTransport(lambda request: httpx.Response(500, stream=stream)) + async with httpx.AsyncClient(transport=transport) as client: + async with client.stream("POST", "https://synthetic.invalid") as response: + raw = await upstream_io.read_bounded_error(response, limit=8) + self.assertEqual(raw, b"abcdefgh") + self.assertTrue(stream.closed) + + async def test_short_error_body_is_preserved(self): + response = httpx.Response(400, content=b"short") + self.assertEqual(await upstream_io.read_bounded_error(response, limit=8), b"short") + + async def test_zero_budget_does_not_open_the_iterator(self): + class NoRead: + def aiter_bytes(self): + raise AssertionError("zero budget must not read upstream") + + self.assertEqual(await upstream_io.read_bounded_error(NoRead(), limit=0), b"") + + + class TransportTests(unittest.IsolatedAsyncioTestCase): async def test_empty_or_malformed_stream_never_replays_post(self): raw_cases = [b"", b"data: {}\n\ndata: [DONE]\n\n",