Harden request, credential and deployment boundaries

Dockerfile

@@ -20,7 +20,7 @@
 
 FROM python:3.12-slim AS runtime
 WORKDIR /app
 ENV CODEBUDDY_AUTH_DIR=/data/auth
 COPY requirements.txt .
 RUN pip install --no-cache-dir -r requirements.txt
 COPY VERSION ./
@@ -28,4 +28,6 @@
 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"]

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；普通文本排在它们之后

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:

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

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

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

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

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)

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

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):
     """只重试一次建连失败，其他错误交给调用方按协议返回。"""