From f32d466845979f46b9dbee67ec2f4b07e135f535 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 20:37:19 +0800 Subject: [PATCH 01/23] Keep image-bearing failed tool results convertible Prefixing is_error tool results assumed string content, so a failed tool result containing an image block raised TypeError and the request was rejected with 400. The failure marker is now prepended as a text block when the converted content is a block list. Verified by extending the is_error differential test with an image-bearing case. --- app/adapters/anthropic_adapter.py | 7 +++++-- app/admin_api.py | 3 +++ app/auth_oauth.py | 23 +++++++++++++++++++++++ converter.py | 2 ++ tests/test_anthropic_adapter.py | 7 +++++++ 5 files changed, 40 insertions(+), 2 deletions(-) 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..e213e7e 100644 --- a/app/admin_api.py +++ b/app/admin_api.py @@ -343,6 +343,9 @@ def upload(body): 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/auth_oauth.py b/app/auth_oauth.py index 1b7374f..3da05f0 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 @@ -95,6 +96,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/converter.py b/converter.py index 91c16bd..3cf3a14 100644 --- a/converter.py +++ b/converter.py @@ -1633,6 +1633,8 @@ async def admin_add_credential(request: Request, 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): 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") From 556071c2dddbae98a3e1578c45bae85a0e04c532 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 20:38:46 +0800 Subject: [PATCH 02/23] Map Messages 404s to Anthropic not_found_error A 404 from model guard or model availability arrived with invalid_request_error because only the embedded error type was mapped. Anthropic clients expect not_found_error for HTTP 404, so the status code now wins on the /v1/messages error path. Verified by an endpoint test asserting not_found_error for a rejected model. --- converter.py | 4 +++- tests/test_runtime_endpoints.py | 11 +++++++++++ 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/converter.py b/converter.py index 3cf3a14..a1cb227 100644 --- a/converter.py +++ b/converter.py @@ -1373,7 +1373,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}, diff --git a/tests/test_runtime_endpoints.py b/tests/test_runtime_endpoints.py index f53ad64..c8fb09c 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()) From b07739c84e7504fae58a12cdbbbc4a5becb1ec16 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 20:40:24 +0800 Subject: [PATCH 03/23] Retry transient empty credits pages beyond page one The empty-Accounts retry only applied to the first page, so the known transient empty response on a later page ended pagination early and returned an understated balance with partial=false. Empty pages are now retried on every page before being accepted as the end of the list. Verified by a pagination test whose second page answers empty once. --- app/credits.py | 2 +- tests/test_credits.py | 22 ++++++++++++++++++++++ 2 files changed, 23 insertions(+), 1 deletion(-) 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/tests/test_credits.py b/tests/test_credits.py index 72e95b4..088293c 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") From 39f172835ab55db0b02c6f62bed7394a0e831275 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 20:44:13 +0800 Subject: [PATCH 04/23] Mark accounts that fail before any successful sync as stale An enabled account failing on the initial sync had no snapshot in usage_daily_accounts, so the publish loop never saw it: the aggregate from the successful accounts was published with partial=false and no stale_accounts, presenting an incomplete view as exact. Stale marking now applies to every enabled account that failed this round, with or without a prior snapshot. Verified by extending the sync regression test with a first-round-failure phase. --- converter.py | 4 +++- tests/test_credits.py | 8 ++++++++ 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/converter.py b/converter.py index a1cb227..da0283f 100644 --- a/converter.py +++ b/converter.py @@ -1260,7 +1260,9 @@ 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: diff --git a/tests/test_credits.py b/tests/test_credits.py index 088293c..74e9c72 100644 --- a/tests/test_credits.py +++ b/tests/test_credits.py @@ -651,6 +651,14 @@ 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: + # 首轮即有账号失败且无任何历史快照：也必须标 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 From d3c8580625d0245a05e74ff1cb5ad9d359f7370e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 20:46:13 +0800 Subject: [PATCH 05/23] Surface incomplete billing data on the subscription endpoint Billing totals could be partial (pagination cap or a failed account sync), but the subscription endpoint returned truncated balances and hard limits without any marker. It now carries codebuddy_partial and, when accounts are stale, codebuddy_stale_accounts. Verified by the billing identity test extended with the new field. --- converter.py | 3 +++ tests/test_credits.py | 1 + 2 files changed, 4 insertions(+) diff --git a/converter.py b/converter.py index da0283f..4073cc4 100644 --- a/converter.py +++ b/converter.py @@ -1846,6 +1846,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 {}), } diff --git a/tests/test_credits.py b/tests/test_credits.py index 74e9c72..8ab7935 100644 --- a/tests/test_credits.py +++ b/tests/test_credits.py @@ -756,6 +756,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") From 59dae80abb452e3fd33cd4cdbab7675ad8a4de39 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 20:49:32 +0800 Subject: [PATCH 06/23] Record the budget-exhausting generation and keep its usage through sanitization Two gaps hid retried tool-call spend: the branch raising after budget exhaustion never recorded that final generation, and the audit sanitizer's numeric allowlist dropped total_tokens/max_attempts even when recorded. The exhaustion path now logs a tool_args_exhausted attempt with usage, and safe_attempt accepts the two numeric fields. Verified by the extended retry-budget test (affected tests pass). --- app/audit_store.py | 2 +- converter.py | 5 +++++ tests/test_runtime_endpoints.py | 4 ++++ 3 files changed, 10 insertions(+), 1 deletion(-) 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/converter.py b/converter.py index 4073cc4..4668580 100644 --- a/converter.py +++ b/converter.py @@ -2597,6 +2597,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，账务不再只看见最后一次 diff --git a/tests/test_runtime_endpoints.py b/tests/test_runtime_endpoints.py index c8fb09c..b2b0745 100644 --- a/tests/test_runtime_endpoints.py +++ b/tests/test_runtime_endpoints.py @@ -232,6 +232,10 @@ 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 From 1598455e8b3388f142ce17df6e1fff4961b2216c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 20:50:52 +0800 Subject: [PATCH 07/23] Normalize credential token aliases before persisting imports Validation accepted access_token/token aliases but the runtime reads only accessToken, so an aliased credential imported successfully yet produced an empty Bearer header and failed refreshes. Both ingestion paths (file import and WebUI upload) now persist the normalized form via auth_oauth.normalize_cred_data, folding aliases into the canonical accessToken/refreshToken/tokenType fields. Verified by an upload test asserting the persisted content (affected tests pass). --- tests/test_admin_api.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) 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) From ad17ca4f7917c8798bd3380fb3c21bd8585a5014 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 20:55:43 +0800 Subject: [PATCH 08/23] Reject non-finite timestamps and non-standard JSON constants on credential intake Credential upload/import accepted NaN/Infinity (Python json allows them by default) and only type-checked expiresAt, so a non-finite expiry could break expiry checks and JSON serialization of admin responses. Both intake paths now parse with a strict JSON loader rejecting non-standard constants, and validate_cred_data rejects non-finite, boolean, or implausible expiresAt/lastRefreshTime values. Verified by new validation cases (affected tests pass). --- app/admin_api.py | 2 +- app/auth_oauth.py | 18 ++++++++++++++++++ converter.py | 2 +- tests/test_auth_oauth.py | 17 +++++++++++++++++ 4 files changed, 37 insertions(+), 2 deletions(-) diff --git a/app/admin_api.py b/app/admin_api.py index e213e7e..1910e45 100644 --- a/app/admin_api.py +++ b/app/admin_api.py @@ -336,7 +336,7 @@ 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("凭据格式无效") diff --git a/app/auth_oauth.py b/app/auth_oauth.py index 3da05f0..e3beafc 100644 --- a/app/auth_oauth.py +++ b/app/auth_oauth.py @@ -15,6 +15,7 @@ import base64 import copy import json +import math import re import threading import time @@ -70,6 +71,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): @@ -85,6 +95,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 + # bool 是 int 子类必须显式排除；NaN/Infinity 会让到期判断与 JSON 序列化行为异常 + if isinstance(value, bool) or not isinstance(value, (int, float)) \ + or not math.isfinite(value) or not 0 < float(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): diff --git a/converter.py b/converter.py index 4668580..70945bd 100644 --- a/converter.py +++ b/converter.py @@ -1630,7 +1630,7 @@ 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("凭据格式或站点校验失败") diff --git a/tests/test_auth_oauth.py b/tests/test_auth_oauth.py index 7098133..88a21e8 100644 --- a/tests/test_auth_oauth.py +++ b/tests/test_auth_oauth.py @@ -55,6 +55,23 @@ 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 bad_ts in (float("nan"), float("inf"), True, 0, -1, 99999999999999): + c = _cred() + c["auth"]["expiresAt"] = bad_ts + assert validate_cred_data(c)[1], bad_ts + 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") From ea58bd3914a53d5f9a7f71cbc5aabfca6df00fc5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 21:00:35 +0800 Subject: [PATCH 09/23] Bind loopback by default and refuse unauthenticated open binding The compose port mapping fell back to 0.0.0.0 with an empty key fallback, so a deployment that skipped .env setup exposed unauthenticated inference to the network. The mapping now defaults to 127.0.0.1, and a native run binding a non-loopback host with an empty API key refuses to start unless CODEBUDDY2API_ALLOW_OPEN_NOAUTH=true is set explicitly. Docker keeps working: the container listens on 0.0.0.0 by necessity and sets the opt-in flag, with exposure controlled by the port mapping. Verified by a startup guard test (refuse / opt-in / key-set paths). --- Dockerfile | 2 ++ converter.py | 6 ++++++ docker-compose.yml | 4 +++- tests/test_runtime_endpoints.py | 8 ++++++++ 4 files changed, 19 insertions(+), 1 deletion(-) 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/converter.py b/converter.py index 70945bd..f53fd25 100644 --- a/converter.py +++ b/converter.py @@ -3056,6 +3056,12 @@ def main(): if args.command == "login": return login(site=args.site, open_browser=not args.no_browser) + # 暴露到非回环地址且未设 API key = 匿名推理开放：默认拒启；Docker 由 compose 端口映射控制边界并显式放行 + if (args.host not in ("127.0.0.1", "::1", "localhost") and not args.api_key + and os.environ.get("CODEBUDDY2API_ALLOW_OPEN_NOAUTH", "").lower() not in ("1", "true", "yes")): + ap.error("非回环绑定且未设置 API key 会匿名开放推理额度；" + "请设置 CODEBUDDY2API_KEY，或确知风险后以 CODEBUDDY2API_ALLOW_OPEN_NOAUTH=true 显式放行") + for key in ("max_images", "image_policy", "max_request_bytes", "log_body_limit", "auto_trial", "tool_call_max_retry"): CONFIG[key] = getattr(args, key) 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/tests/test_runtime_endpoints.py b/tests/test_runtime_endpoints.py index b2b0745..66c1f87 100644 --- a/tests/test_runtime_endpoints.py +++ b/tests/test_runtime_endpoints.py @@ -602,6 +602,14 @@ 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_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"} From 7e080f7cb5f1ebb33525b4a4ef2463de8ac91499 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 21:06:31 +0800 Subject: [PATCH 10/23] Cap raw inbound request bytes at the ASGI layer before parsing The endpoints read request.json() before any limit ran, so an inbound body full of ignored fields or soon-to-be-stripped images allocated memory and CPU that the later 413 could not undo; max_request_bytes only governs the processed upstream body. A new outermost middleware counts raw bytes on the receive channel (chunked included, ignoring Content-Length) and answers 413 with the client protocol's error shape before any JSON parsing; --max-inbound-bytes / CODEBUDDY2API_MAX_INBOUND_BYTES (default 64 MiB) configures it, and /admin routes keep their own existing body limits. Verified by HTTP and chunked ASGI-level tests. --- app/inbound_limits.py | 70 +++++++++++++++++++++++++++++++++ app/runtime_management.py | 2 + converter.py | 6 ++- tests/test_runtime_endpoints.py | 55 ++++++++++++++++++++++++++ 4 files changed, 132 insertions(+), 1 deletion(-) create mode 100644 app/inbound_limits.py diff --git a/app/inbound_limits.py b/app/inbound_limits.py new file mode 100644 index 0000000..4b7200f --- /dev/null +++ b/app/inbound_limits.py @@ -0,0 +1,70 @@ +#!/usr/bin/env python3 +"""inbound_limits.py — 推理端点入站原始字节限量。 + +端点先 await request.json() 再检查处理后的上游请求体，原始入站大小无人管：大 JSON、 +被忽略的顶层字段、将被剥离的图片都在解析前已经占用了内存。本中间件在 ASGI receive 层 +累计原始字节（含 chunked 传输，不看 Content-Length），超限直接 413，不进入 JSON 解析。 +缓冲体随后原样回放给下游，端点行为不变。 +""" + +from __future__ import annotations + +import json + + +class InboundBodyLimitMiddleware: + """/v1/* 请求的原始字节上限；limit<=0 时关闭。""" + + def __init__(self, app, config): + self.app = app + self.config = config + + async def __call__(self, scope, receive, send): + if scope["type"] != "http" or not scope.get("path", "").startswith("/v1/"): + 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 {"type": "http.disconnect"} + 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/runtime_management.py b/app/runtime_management.py index 86b2c04..b83060f 100644 --- a/app/runtime_management.py +++ b/app/runtime_management.py @@ -91,6 +91,8 @@ def install(gateway): install_admin(app, config, config["management"]) app.add_middleware(AuditMiddleware, config=config) app.add_middleware(PolicyScopeMiddleware) + from .inbound_limits import InboundBodyLimitMiddleware + app.add_middleware(InboundBodyLimitMiddleware, config=config) # 最外层：解析前限原始字节 install_pages(app, Path(gateway.__file__).resolve().parent / "web" / "dist") diff --git a/converter.py b/converter.py index f53fd25..2aeb440 100644 --- a/converter.py +++ b/converter.py @@ -1396,6 +1396,7 @@ 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, "usage_daily": None, # 官方用量聚合视图（日期×模型 credit），供 billing/usage 出 daily_costs "usage_daily_accounts": None, # 按账号的用量快照；单账号失败不丢历史 "credit_price_cny": None, "credit_price_usd": None, "usd_rate": None, @@ -3041,6 +3042,9 @@ 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("--log-body-limit", type=_nonnegative_int, metavar="BYTES", default=os.environ.get("CODEBUDDY2API_LOG_BODY_LIMIT", "65536"), help="每条正文日志的预览字节上限，默认 64 KiB；0 只记录摘要") @@ -3063,7 +3067,7 @@ def main(): "请设置 CODEBUDDY2API_KEY，或确知风险后以 CODEBUDDY2API_ALLOW_OPEN_NOAUTH=true 显式放行") 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"): CONFIG[key] = getattr(args, key) CONFIG["api_key"] = args.api_key CONFIG["desensitize"] = args.desensitize diff --git a/tests/test_runtime_endpoints.py b/tests/test_runtime_endpoints.py index 66c1f87..cf40051 100644 --- a/tests/test_runtime_endpoints.py +++ b/tests/test_runtime_endpoints.py @@ -570,6 +570,61 @@ 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", "path": "/v1/chat/completions"}, receive, send)) + self.assertFalse(reached) # 超限请求不进入下游 + self.assertEqual(sent[0]["status"], 413) + + class ConfigurationTests(unittest.TestCase): def configure(self, env=None, flags=(), invalid=False): with contextlib.ExitStack() as stack: From 892bd544d2272a1ef3308555973ef4df0272a039 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 21:09:54 +0800 Subject: [PATCH 11/23] Move credential selection and refresh off the event loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cred picking runs get_headers under an RLock plus a cross-process file lock and may synchronously refresh a token with a 15s httpx client — all from async endpoints, so one expiring token stalled every in-flight stream. The three inference endpoints now call _route_chat through run_in_threadpool, moving selection/refresh/lock waits to a bounded worker pool while contextvars (and thus request observation) still propagate. Verified by a wiring regression test and the endpoint suite. --- converter.py | 10 +++++++--- tests/test_runtime_endpoints.py | 10 ++++++++++ 2 files changed, 17 insertions(+), 3 deletions(-) diff --git a/converter.py b/converter.py index 2aeb440..a7224fd 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: @@ -2307,7 +2308,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() @@ -2742,7 +2744,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() @@ -2836,7 +2839,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() diff --git a/tests/test_runtime_endpoints.py b/tests/test_runtime_endpoints.py index cf40051..102e52b 100644 --- a/tests/test_runtime_endpoints.py +++ b/tests/test_runtime_endpoints.py @@ -239,6 +239,16 @@ def flaky(request): 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": { From af6d117cf56772b9375e2a5ba93db4ef64f555a0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 21:21:54 +0800 Subject: [PATCH 12/23] Bound aggregated output, error bodies, and inference concurrency Aggregation buffered a whole upstream response with no budget: content, reasoning and tool arguments now share a collection byte budget (--max-collect-bytes, default 8 MiB) that fails with response_too_large instead of caching unbounded output; upstream error bodies are read through a 4 MiB bounded reader instead of aread; and a concurrency gate (--max-concurrent, default 64) rejects excess inference requests with 503 + Retry-After for the whole stream lifecycle instead of queueing them. Verified by budget, bounded-read and gate regression tests (affected test files pass). --- app/inbound_limits.py | 50 +++++++++++++++++++++++++++++++++ app/runtime_management.py | 6 ++-- app/upstream_io.py | 42 +++++++++++++++++++++------ converter.py | 19 +++++++++---- tests/test_runtime_endpoints.py | 38 +++++++++++++++++++++++++ tests/test_upstream_io.py | 28 ++++++++++++++++++ 6 files changed, 167 insertions(+), 16 deletions(-) diff --git a/app/inbound_limits.py b/app/inbound_limits.py index 4b7200f..4581744 100644 --- a/app/inbound_limits.py +++ b/app/inbound_limits.py @@ -12,6 +12,56 @@ import json +import asyncio + + +_GATED_PATHS = ("/v1/chat/completions", "/v1/responses", "/v1/messages") + + +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 not scope.get("path", "").startswith(_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(): # 无空闲名额：立即失败并给出重试提示 + raw = json.dumps({"error": {"message": "inference concurrency limit reached, retry later", + "type": "rate_limit_error", "code": "concurrency_limit"}}).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: """/v1/* 请求的原始字节上限；limit<=0 时关闭。""" diff --git a/app/runtime_management.py b/app/runtime_management.py index b83060f..493709e 100644 --- a/app/runtime_management.py +++ b/app/runtime_management.py @@ -91,8 +91,10 @@ def install(gateway): install_admin(app, config, config["management"]) app.add_middleware(AuditMiddleware, config=config) app.add_middleware(PolicyScopeMiddleware) - from .inbound_limits import InboundBodyLimitMiddleware - app.add_middleware(InboundBodyLimitMiddleware, config=config) # 最外层：解析前限原始字节 + from .inbound_limits import ConcurrencyLimitMiddleware, InboundBodyLimitMiddleware + app.add_middleware(InboundBodyLimitMiddleware, config=config) + # 最外层：并发名额先用完即 503，再进行请求体缓冲与处理 + app.add_middleware(ConcurrencyLimitMiddleware, 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..3e72606 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,19 @@ 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。""" + buf = bytearray() + async for chunk in response.aiter_bytes(): + if len(buf) >= limit: + break + buf.extend(chunk[: max(0, limit - len(buf))]) + 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 a7224fd..cc9eab4 100644 --- a/converter.py +++ b/converter.py @@ -72,7 +72,7 @@ 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.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 @@ -1398,6 +1398,7 @@ async def _protocol_http_exception(request: Request, exc: HTTPException): "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, @@ -2439,7 +2440,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() @@ -2557,11 +2558,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: @@ -2627,7 +2628,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(): @@ -3049,6 +3050,12 @@ def main(): 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 只记录摘要") @@ -3071,7 +3078,7 @@ def main(): "请设置 CODEBUDDY2API_KEY，或确知风险后以 CODEBUDDY2API_ALLOW_OPEN_NOAUTH=true 显式放行") for key in ("max_images", "image_policy", "max_request_bytes", "log_body_limit", "auto_trial", - "tool_call_max_retry", "max_inbound_bytes"): + "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 diff --git a/tests/test_runtime_endpoints.py b/tests/test_runtime_endpoints.py index 102e52b..4407b94 100644 --- a/tests/test_runtime_endpoints.py +++ b/tests/test_runtime_endpoints.py @@ -635,6 +635,44 @@ async def send(message): self.assertEqual(sent[0]["status"], 413) +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)) + await asyncio.wait_for(entered.wait(), 2) + await mw(scope, receive, send) + starts = [m for m in sent if m["type"] == "http.response.start"] + self.assertEqual(starts[-1]["status"], 503) + self.assertIn(b"retry-after", dict(starts[-1]["headers"])) + release.set() + await asyncio.wait_for(first, 2) + sent.clear() + await mw(scope, receive, send) + self.assertEqual(sent[0]["status"], 200) + + class ConfigurationTests(unittest.TestCase): def configure(self, env=None, flags=(), invalid=False): with contextlib.ExitStack() as stack: diff --git a/tests/test_upstream_io.py b/tests/test_upstream_io.py index b1c7611..d9eb093 100644 --- a/tests/test_upstream_io.py +++ b/tests/test_upstream_io.py @@ -222,6 +222,34 @@ 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 TransportTests(unittest.IsolatedAsyncioTestCase): async def test_empty_or_malformed_stream_never_replays_post(self): raw_cases = [b"", b"data: {}\n\ndata: [DONE]\n\n", From f79085d63fd5a360bf18dd34971a764b24b1926e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 21:22:37 +0800 Subject: [PATCH 13/23] Document the new resource limits and deployment guardrails Sync the advanced guides: --max-inbound-bytes / --max-collect-bytes / --max-concurrent options, the loopback-by-default binding with the explicit open-binding opt-in, and credential intake normalization with strict JSON parsing. --- docs/advanced.md | 8 ++++++++ docs/advanced.zh-CN.md | 8 ++++++++ 2 files changed, 16 insertions(+) diff --git a/docs/advanced.md b/docs/advanced.md index f745fe1..9c29e33 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 inbound body byte limit enforced at the ASGI layer before parsing (chunked included); 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` | Inference concurrency limit; excess requests get an immediate 503 with Retry-After; `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,6 +136,11 @@ 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); a native run bound to a non-loopback host with an empty API key refuses to start unless `CODEBUDDY2API_ALLOW_OPEN_NOAUTH=true` is set explicitly. +- 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. diff --git a/docs/advanced.zh-CN.md b/docs/advanced.zh-CN.md index 57bf7eb..1d5d34d 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` | 入站原始请求体字节上限（解析前在 ASGI 层生效，含 chunked），超限返回 413 | +| `--max-collect-bytes` | `8388608` | 聚合路径输出收集总字节上限（正文+思考+工具参数），超限返回 `response_too_large`；`0` 不限制 | +| `--max-concurrent` | `64` | 推理端点并发上限，占满立即 503（含 Retry-After）；`0` 不限制 | | `--max-request-bytes` | `33554432` | 处理后的上游 JSON 字节上限，须为正整数 | | `--log-body-limit` | `65536` | 兼容文本日志正文预览字节；`0` 只记摘要，不控制 SQLite 诊断预算 | @@ -133,6 +136,11 @@ WebUI 可以直接上传文件；以下限制针对 `POST /admin/credentials` - `/v1/messages/count_tokens` 返回字符启发式估算值，仅作预算参考，不是精确计数。 - 兼容文本日志和 SQLite 审计使用独立预算；日志仅记录有界、脱敏预览，不是完整原始请求。日志、凭证导出和备份仍须按私有数据保管。 +## 部署暴露与凭据导入 + +- Compose 端口映射默认只绑回环（`CODEBUDDY2API_BIND` 默认 127.0.0.1）；原生运行绑定非回环地址且未设 API key 时拒绝启动，须显式设 `CODEBUDDY2API_ALLOW_OPEN_NOAUTH=true`。 +- 凭据导入/上传在落盘前把 token 别名归一化为官方字段名；严格 JSON 解析拒绝 NaN/Infinity，`expiresAt`/`lastRefreshTime` 必须是合理的有限毫秒时间戳。 + ## 账务数据完整性 - 余额/用量来自官方接口的分页遍历；达到页数上限或任一账号同步失败时，响应带 `partial: true`（及 `stale_accounts` 列表），不伪装为精确全量。 From 629c4ae2b57eb6866e03d0ed970e7419a591a034 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 21:39:26 +0800 Subject: [PATCH 14/23] Preserve Messages error envelopes on concurrency rejection Return the Anthropic error envelope and api_error type for Messages 503 responses while preserving Retry-After and the concurrency_limit code. Keep OpenAI rejections unchanged. Verified saturation and recovery across all three protocols (1 test, 3 subtests). --- app/inbound_limits.py | 9 +++++++-- tests/test_runtime_endpoints.py | 29 ++++++++++++++++++++++------- 2 files changed, 29 insertions(+), 9 deletions(-) diff --git a/app/inbound_limits.py b/app/inbound_limits.py index 4581744..97a65c4 100644 --- a/app/inbound_limits.py +++ b/app/inbound_limits.py @@ -48,8 +48,13 @@ async def __call__(self, scope, receive, send): return await self.app(scope, receive, send) gate = self._gate() if gate.locked(): # 无空闲名额：立即失败并给出重试提示 - raw = json.dumps({"error": {"message": "inference concurrency limit reached, retry later", - "type": "rate_limit_error", "code": "concurrency_limit"}}).encode() + 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())]}) diff --git a/tests/test_runtime_endpoints.py b/tests/test_runtime_endpoints.py index 4407b94..0872342 100644 --- a/tests/test_runtime_endpoints.py +++ b/tests/test_runtime_endpoints.py @@ -661,13 +661,28 @@ async def send(message): sent.append(message) first = asyncio.create_task(mw(scope, receive, send)) - await asyncio.wait_for(entered.wait(), 2) - await mw(scope, receive, send) - starts = [m for m in sent if m["type"] == "http.response.start"] - self.assertEqual(starts[-1]["status"], 503) - self.assertIn(b"retry-after", dict(starts[-1]["headers"])) - release.set() - await asyncio.wait_for(first, 2) + 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) From 27fe051ea771a0bb28a4a9de73a668737a0fab83 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 21:41:47 +0800 Subject: [PATCH 15/23] Publish billing completeness when every usage sync fails Publish an empty partial view with stale accounts when no account snapshot exists, keeping fetched_at zero for quota-delta fallback. Preserve existing per-account snapshots on failures and clear obsolete aggregates when no credentials remain enabled. Sync the bilingual guides and verify both billing HTTP endpoints, first-sync failure, retained-history failure, recovery, disabling and removal (4 targeted tests passed). --- converter.py | 7 ++----- docs/advanced.md | 1 + docs/advanced.zh-CN.md | 1 + tests/test_credits.py | 44 ++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 48 insertions(+), 5 deletions(-) diff --git a/converter.py b/converter.py index cc9eab4..e395026 100644 --- a/converter.py +++ b/converter.py @@ -1240,12 +1240,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(): @@ -1266,8 +1264,7 @@ def _publish_usage_daily(pool, 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), @@ -1276,7 +1273,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): diff --git a/docs/advanced.md b/docs/advanced.md index 9c29e33..05ea11b 100644 --- a/docs/advanced.md +++ b/docs/advanced.md @@ -145,6 +145,7 @@ Credential domain / token issuer determine the product identity. Chat and refres - 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 1d5d34d..b09b9ff 100644 --- a/docs/advanced.zh-CN.md +++ b/docs/advanced.zh-CN.md @@ -145,6 +145,7 @@ WebUI 可以直接上传文件；以下限制针对 `POST /admin/credentials` - 余额/用量来自官方接口的分页遍历；达到页数上限或任一账号同步失败时，响应带 `partial: true`（及 `stale_accounts` 列表），不伪装为精确全量。 - 单账号同步失败保留其上次成功快照；接口返回 HTTP 200 但业务码失败或结构缺失时按错误处理，不覆盖历史。 +- 首次同步全部失败时，两个账务端点仍标记数据不完整及失败账号，并保留额度差回退；已有账号快照在失败时不更新其成功时间。 - daily_costs 按各站（国内/国际）本站单价逐日折算，不再使用全局平均单价。 ## 故障与重试 diff --git a/tests/test_credits.py b/tests/test_credits.py index 8ab7935..fee80ce 100644 --- a/tests/test_credits.py +++ b/tests/test_credits.py @@ -651,6 +651,31 @@ 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) @@ -664,6 +689,15 @@ def fake_fetch(token, uid="", domain=""): 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"] @@ -682,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"], \ From a94efafa7578dcb12136903e1379c1bb45cb3d02 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 21:43:09 +0800 Subject: [PATCH 16/23] Preserve real disconnect events after inbound body replay Forward subsequent receive calls to the server instead of synthesizing http.disconnect, which cancelled streamed responses after their first asynchronous yield. Verify both middleware layers with Starlette StreamingResponse: all three routes finish normally, and a genuine disconnect closes the stream and releases capacity (5 tests, 6 subtests passed). --- app/inbound_limits.py | 2 +- tests/test_runtime_endpoints.py | 64 +++++++++++++++++++++++++++++++++ 2 files changed, 65 insertions(+), 1 deletion(-) diff --git a/app/inbound_limits.py b/app/inbound_limits.py index 97a65c4..d025fcc 100644 --- a/app/inbound_limits.py +++ b/app/inbound_limits.py @@ -102,7 +102,7 @@ async def __call__(self, scope, receive, send): async def replay(): nonlocal replayed if replayed: - return {"type": "http.disconnect"} + return await receive() # 请求体结束不等于断连，继续监听真实连接。 replayed = True return {"type": "http.request", "body": buffered, "more_body": False} diff --git a/tests/test_runtime_endpoints.py b/tests/test_runtime_endpoints.py index 0872342..6732d6c 100644 --- a/tests/test_runtime_endpoints.py +++ b/tests/test_runtime_endpoints.py @@ -635,6 +635,70 @@ async def send(message): 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），释放后恢复。""" From da9ffb291774cfa167554e2d38f848105d3b53ff Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 22:05:20 +0800 Subject: [PATCH 17/23] Limit concurrency admission to exact generation routes Keep local token counting and unrelated method/path handling outside the generation gate. Verify real count_tokens responses while the gate is full, normal 404/405 handling, and unchanged three-protocol saturation errors (2 tests and 3 subtests passed). --- app/inbound_limits.py | 2 +- tests/test_runtime_endpoints.py | 25 +++++++++++++++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/app/inbound_limits.py b/app/inbound_limits.py index d025fcc..0de6dcd 100644 --- a/app/inbound_limits.py +++ b/app/inbound_limits.py @@ -41,7 +41,7 @@ def _limit(self) -> int: async def __call__(self, scope, receive, send): if (scope["type"] != "http" or scope.get("method") != "POST" - or not scope.get("path", "").startswith(_GATED_PATHS)): + or scope.get("path", "") not in _GATED_PATHS): return await self.app(scope, receive, send) limit = self._limit() if limit <= 0: diff --git a/tests/test_runtime_endpoints.py b/tests/test_runtime_endpoints.py index 6732d6c..b9b901c 100644 --- a/tests/test_runtime_endpoints.py +++ b/tests/test_runtime_endpoints.py @@ -752,6 +752,31 @@ async def send(message): 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): with contextlib.ExitStack() as stack: From 017e0d4058ee4e9e87661e73efefabdc560df6cd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 22:10:42 +0800 Subject: [PATCH 18/23] Authenticate inference headers before buffering or admission Share API-key validation between endpoints and an outer header-only middleware. Reject unauthorized bodies before acquiring generation capacity, preserve Bearer/X-Api-Key precedence and live key changes, and keep protocol-shaped 401 responses even when the gate is full or disabled. Sync bilingual documentation. Verified production middleware ordering, header-only rejection, cookie isolation, capacity recovery and token counting (75 affected tests plus 33 management/observation tests; the new direct-run entry point also passed). --- app/inference_auth.py | 40 +++++++++ app/runtime_management.py | 4 +- converter.py | 12 +-- docs/advanced.md | 3 +- docs/advanced.zh-CN.md | 3 +- tests/test_inference_admission.py | 137 ++++++++++++++++++++++++++++++ 6 files changed, 186 insertions(+), 13 deletions(-) create mode 100644 app/inference_auth.py create mode 100644 tests/test_inference_admission.py diff --git a/app/inference_auth.py b/app/inference_auth.py new file mode 100644 index 0000000..10ebd29 --- /dev/null +++ b/app/inference_auth.py @@ -0,0 +1,40 @@ +"""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 + + +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 not scope.get("path", "").startswith("/v1/"): + 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 493709e..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 @@ -93,8 +94,9 @@ def install(gateway): app.add_middleware(PolicyScopeMiddleware) from .inbound_limits import ConcurrencyLimitMiddleware, InboundBodyLimitMiddleware app.add_middleware(InboundBodyLimitMiddleware, config=config) - # 最外层：并发名额先用完即 503，再进行请求体缓冲与处理 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/converter.py b/converter.py index e395026..50dd055 100644 --- a/converter.py +++ b/converter.py @@ -73,6 +73,7 @@ def desensitize_body(body, roles=("system",), desensitize_harness_user=False, 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, 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 @@ -1468,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]): diff --git a/docs/advanced.md b/docs/advanced.md index 05ea11b..736ad62 100644 --- a/docs/advanced.md +++ b/docs/advanced.md @@ -32,7 +32,7 @@ Compose explicitly passes some environment variables and CLI flags, so deleting | `--tool-call-max-retry` | `3` | Extra generations after malformed tool calls (each consumes credits); `0` disables retries | | `--max-inbound-bytes` | `67108864` | Raw inbound body byte limit enforced at the ASGI layer before parsing (chunked included); 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` | Inference concurrency limit; excess requests get an immediate 503 with Retry-After; `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 | @@ -139,6 +139,7 @@ Credential domain / token issuer determine the product identity. Chat and refres ## Deployment exposure and credential intake - The compose port mapping binds loopback by default (`CODEBUDDY2API_BIND` defaults to 127.0.0.1); a native run bound to a non-loopback host with an empty API key refuses to start unless `CODEBUDDY2API_ALLOW_OPEN_NOAUTH=true` is set explicitly. +- When a key is configured, `/v1/*` verifies request headers before buffering bodies or reserving inference capacity; invalid keys return 401 even while generation slots are full. - 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 diff --git a/docs/advanced.zh-CN.md b/docs/advanced.zh-CN.md index b09b9ff..397e415 100644 --- a/docs/advanced.zh-CN.md +++ b/docs/advanced.zh-CN.md @@ -32,7 +32,7 @@ Compose 会显式传入部分环境变量及 CLI 参数，删除 `.env` 中的 | `--tool-call-max-retry` | `3` | 工具参数损坏时的额外生成上限（每次都消耗额度）；`0` 不重试 | | `--max-inbound-bytes` | `67108864` | 入站原始请求体字节上限（解析前在 ASGI 层生效，含 chunked），超限返回 413 | | `--max-collect-bytes` | `8388608` | 聚合路径输出收集总字节上限（正文+思考+工具参数），超限返回 `response_too_large`；`0` 不限制 | -| `--max-concurrent` | `64` | 推理端点并发上限，占满立即 503（含 Retry-After）；`0` 不限制 | +| `--max-concurrent` | `64` | 仅限制三个生成端点；占满立即 503（含 Retry-After），不限制 token 估算；`0` 不限制 | | `--max-request-bytes` | `33554432` | 处理后的上游 JSON 字节上限，须为正整数 | | `--log-body-limit` | `65536` | 兼容文本日志正文预览字节；`0` 只记摘要，不控制 SQLite 诊断预算 | @@ -139,6 +139,7 @@ WebUI 可以直接上传文件；以下限制针对 `POST /admin/credentials` ## 部署暴露与凭据导入 - Compose 端口映射默认只绑回环（`CODEBUDDY2API_BIND` 默认 127.0.0.1）；原生运行绑定非回环地址且未设 API key 时拒绝启动，须显式设 `CODEBUDDY2API_ALLOW_OPEN_NOAUTH=true`。 +- 配置 key 时，`/v1/*` 在缓冲请求体、预留推理名额之前校验请求头；即使名额已满，无效 key 仍返回 401。 - 凭据导入/上传在落盘前把 token 别名归一化为官方字段名；严格 JSON 解析拒绝 NaN/Infinity，`expiresAt`/`lastRefreshTime` 必须是合理的有限毫秒时间戳。 ## 账务数据完整性 diff --git a/tests/test_inference_admission.py b/tests/test_inference_admission.py new file mode 100644 index 0000000..9fa5b23 --- /dev/null +++ b/tests/test_inference_admission.py @@ -0,0 +1,137 @@ +"""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())} + + 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): + 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": "POST", "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) + return start["status"], json.loads(body) + + 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_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) From 19a4c50402449ffb4bf0976b4eaf734c506c9500 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 22:16:14 +0800 Subject: [PATCH 19/23] Preserve routing errors outside inference body admission Scope early authentication to exact generation and token-count POST routes, retaining ordinary 404/405 behavior for unknown paths and unsupported methods. This fixes the browser integration regression without weakening header-only rejection before buffering and capacity admission. Updated the guides; all 5 admission tests and 32 subtests passed, with the unchanged runtime endpoint suite also passing. --- app/inference_auth.py | 7 ++++++- docs/advanced.md | 2 +- docs/advanced.zh-CN.md | 2 +- tests/test_inference_admission.py | 18 +++++++++++++++--- 4 files changed, 23 insertions(+), 6 deletions(-) diff --git a/app/inference_auth.py b/app/inference_auth.py index 10ebd29..cb5a368 100644 --- a/app/inference_auth.py +++ b/app/inference_auth.py @@ -6,6 +6,10 @@ from starlette.datastructures import Headers from starlette.responses import JSONResponse +from .inbound_limits import _GATED_PATHS + +_BODY_PATHS = (*_GATED_PATHS, "/v1/messages/count_tokens") + def require_api_key(key, authorization=None, x_api_key=None): """Preserve Bearer precedence, X-Api-Key fallback, and optional empty keys.""" @@ -26,7 +30,8 @@ def __init__(self, app, config): self.config = config async def __call__(self, scope, receive, send): - if scope["type"] != "http" or not scope.get("path", "").startswith("/v1/"): + 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: diff --git a/docs/advanced.md b/docs/advanced.md index 736ad62..3631093 100644 --- a/docs/advanced.md +++ b/docs/advanced.md @@ -139,7 +139,7 @@ Credential domain / token issuer determine the product identity. Chat and refres ## Deployment exposure and credential intake - The compose port mapping binds loopback by default (`CODEBUDDY2API_BIND` defaults to 127.0.0.1); a native run bound to a non-loopback host with an empty API key refuses to start unless `CODEBUDDY2API_ALLOW_OPEN_NOAUTH=true` is set explicitly. -- When a key is configured, `/v1/*` verifies request headers before buffering bodies or reserving inference capacity; invalid keys return 401 even while generation slots are full. +- 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 diff --git a/docs/advanced.zh-CN.md b/docs/advanced.zh-CN.md index 397e415..26d56bb 100644 --- a/docs/advanced.zh-CN.md +++ b/docs/advanced.zh-CN.md @@ -139,7 +139,7 @@ WebUI 可以直接上传文件；以下限制针对 `POST /admin/credentials` ## 部署暴露与凭据导入 - Compose 端口映射默认只绑回环（`CODEBUDDY2API_BIND` 默认 127.0.0.1）；原生运行绑定非回环地址且未设 API key 时拒绝启动，须显式设 `CODEBUDDY2API_ALLOW_OPEN_NOAUTH=true`。 -- 配置 key 时，`/v1/*` 在缓冲请求体、预留推理名额之前校验请求头；即使名额已满，无效 key 仍返回 401。 +- 配置 key 时，生成及 token 估算 POST 在缓冲请求体、预留推理名额之前校验请求头；即使名额已满，无效 key 仍返回 401。其他路由保留原有鉴权和路由行为。 - 凭据导入/上传在落盘前把 token 别名归一化为官方字段名；严格 JSON 解析拒绝 NaN/Infinity，`expiresAt`/`lastRefreshTime` 必须是合理的有限毫秒时间戳。 ## 账务数据完整性 diff --git a/tests/test_inference_admission.py b/tests/test_inference_admission.py index 9fa5b23..1d9f2ed 100644 --- a/tests/test_inference_admission.py +++ b/tests/test_inference_admission.py @@ -39,7 +39,7 @@ async def endpoint(request: Request): layer = layer.app self.gate = layer - async def request(self, path, headers=(), *, forbid_receive=False): + async def request(self, path, headers=(), *, forbid_receive=False, method="POST"): sent = [] replayed = False @@ -55,14 +55,16 @@ async def receive(): async def send(message): sent.append(message) - scope = {"type": "http", "method": "POST", "path": path, "raw_path": path.encode(), + 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) - return start["status"], json.loads(body) + 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) @@ -120,6 +122,16 @@ async def test_current_key_and_header_precedence_are_preserved(self): 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) + self.assertEqual(status, expected, body) + + 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) From 2e7b90dfedfa8171eca3d74780f68989cb4124e5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 22:20:44 +0800 Subject: [PATCH 20/23] Validate the effective bind address after settings resolution Check the resolved host and effective API key after persisted settings are applied, before credential scanning, worker startup or binding. Close initialized stores when refusing startup. Verify saved IPv4/IPv6 wildcard hosts, explicit opt-in, key configuration, CLI overrides and rejection cleanup (7 configuration tests and 22 subtests passed); update the bilingual guides. --- converter.py | 12 ++++++------ docs/advanced.md | 2 +- docs/advanced.zh-CN.md | 2 +- tests/test_runtime_endpoints.py | 28 +++++++++++++++++++++++++++- 4 files changed, 35 insertions(+), 9 deletions(-) diff --git a/converter.py b/converter.py index 50dd055..40f96dc 100644 --- a/converter.py +++ b/converter.py @@ -3060,12 +3060,6 @@ def main(): if args.command == "login": return login(site=args.site, open_browser=not args.no_browser) - # 暴露到非回环地址且未设 API key = 匿名推理开放：默认拒启；Docker 由 compose 端口映射控制边界并显式放行 - if (args.host not in ("127.0.0.1", "::1", "localhost") and not args.api_key - and os.environ.get("CODEBUDDY2API_ALLOW_OPEN_NOAUTH", "").lower() not in ("1", "true", "yes")): - ap.error("非回环绑定且未设置 API key 会匿名开放推理额度；" - "请设置 CODEBUDDY2API_KEY，或确知风险后以 CODEBUDDY2API_ALLOW_OPEN_NOAUTH=true 显式放行") - for key in ("max_images", "image_policy", "max_request_bytes", "log_body_limit", "auto_trial", "tool_call_max_retry", "max_inbound_bytes", "max_collect_bytes", "max_concurrent"): CONFIG[key] = getattr(args, key) @@ -3080,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/docs/advanced.md b/docs/advanced.md index 3631093..c1292bd 100644 --- a/docs/advanced.md +++ b/docs/advanced.md @@ -138,7 +138,7 @@ Credential domain / token issuer determine the product identity. Chat and refres ## Deployment exposure and credential intake -- The compose port mapping binds loopback by default (`CODEBUDDY2API_BIND` defaults to 127.0.0.1); a native run bound to a non-loopback host with an empty API key refuses to start unless `CODEBUDDY2API_ALLOW_OPEN_NOAUTH=true` is set explicitly. +- 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. diff --git a/docs/advanced.zh-CN.md b/docs/advanced.zh-CN.md index 26d56bb..cb39c70 100644 --- a/docs/advanced.zh-CN.md +++ b/docs/advanced.zh-CN.md @@ -138,7 +138,7 @@ WebUI 可以直接上传文件；以下限制针对 `POST /admin/credentials` ## 部署暴露与凭据导入 -- Compose 端口映射默认只绑回环（`CODEBUDDY2API_BIND` 默认 127.0.0.1）；原生运行绑定非回环地址且未设 API key 时拒绝启动，须显式设 `CODEBUDDY2API_ALLOW_OPEN_NOAUTH=true`。 +- 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` 必须是合理的有限毫秒时间戳。 diff --git a/tests/test_runtime_endpoints.py b/tests/test_runtime_endpoints.py index b9b901c..ef5ac8f 100644 --- a/tests/test_runtime_endpoints.py +++ b/tests/test_runtime_endpoints.py @@ -778,9 +778,16 @@ async def test_saturated_generation_gate_does_not_block_token_counting(self): 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)) @@ -792,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")} @@ -817,6 +830,19 @@ def test_open_binding_without_key_requires_explicit_opt_in(self): # 非回环但设了 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"} From 4e810d1c406ae64721758f068fe45dcb94cbc3d9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 22:35:04 +0800 Subject: [PATCH 21/23] Buffer bodies only on authenticated request-consuming routes Share an exact generation/token-count POST allowlist between early authentication and inbound buffering. Bodyless and unknown routes no longer wait for attacker-controlled uploads before authentication or routing. Regressions require zero receive calls for model GETs and 404/405 paths while retaining POST size limits, token counting and SSE lifecycle behavior; update both guides. --- app/inbound_limits.py | 6 ++++-- app/inference_auth.py | 4 +--- docs/advanced.md | 2 +- docs/advanced.zh-CN.md | 2 +- tests/test_inference_admission.py | 16 +++++++++++++++- tests/test_runtime_endpoints.py | 2 +- 6 files changed, 23 insertions(+), 9 deletions(-) diff --git a/app/inbound_limits.py b/app/inbound_limits.py index 0de6dcd..6eec137 100644 --- a/app/inbound_limits.py +++ b/app/inbound_limits.py @@ -16,6 +16,7 @@ _GATED_PATHS = ("/v1/chat/completions", "/v1/responses", "/v1/messages") +_BODY_PATHS = (*_GATED_PATHS, "/v1/messages/count_tokens") class ConcurrencyLimitMiddleware: @@ -68,14 +69,15 @@ async def __call__(self, scope, receive, send): class InboundBodyLimitMiddleware: - """/v1/* 请求的原始字节上限；limit<=0 时关闭。""" + """仅缓冲生成及 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 not scope.get("path", "").startswith("/v1/"): + 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) diff --git a/app/inference_auth.py b/app/inference_auth.py index cb5a368..7b4a5f8 100644 --- a/app/inference_auth.py +++ b/app/inference_auth.py @@ -6,9 +6,7 @@ from starlette.datastructures import Headers from starlette.responses import JSONResponse -from .inbound_limits import _GATED_PATHS - -_BODY_PATHS = (*_GATED_PATHS, "/v1/messages/count_tokens") +from .inbound_limits import _BODY_PATHS def require_api_key(key, authorization=None, x_api_key=None): diff --git a/docs/advanced.md b/docs/advanced.md index c1292bd..1a71daf 100644 --- a/docs/advanced.md +++ b/docs/advanced.md @@ -30,7 +30,7 @@ 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 inbound body byte limit enforced at the ASGI layer before parsing (chunked included); 413 beyond it | +| `--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 | diff --git a/docs/advanced.zh-CN.md b/docs/advanced.zh-CN.md index cb39c70..ea75e03 100644 --- a/docs/advanced.zh-CN.md +++ b/docs/advanced.zh-CN.md @@ -30,7 +30,7 @@ Compose 会显式传入部分环境变量及 CLI 参数，删除 `.env` 中的 | `--max-images` | `16` | 单请求图片总数；`0` 不允许图片 | | `--image-policy` | `truncate` | 保留最新图片；设为 `error` 时超限返回 413 | | `--tool-call-max-retry` | `3` | 工具参数损坏时的额外生成上限（每次都消耗额度）；`0` 不重试 | -| `--max-inbound-bytes` | `67108864` | 入站原始请求体字节上限（解析前在 ASGI 层生效，含 chunked），超限返回 413 | +| `--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 字节上限，须为正整数 | diff --git a/tests/test_inference_admission.py b/tests/test_inference_admission.py index 1d9f2ed..82ced46 100644 --- a/tests/test_inference_admission.py +++ b/tests/test_inference_admission.py @@ -28,6 +28,12 @@ def setUp(self): 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")) @@ -128,8 +134,16 @@ async def test_unknown_routes_and_methods_preserve_routing_errors(self): ("/v1/messages/unknown", "POST", 404), ("/v1/messages", "GET", 405)): with self.subTest(path=path, method=method): - status, body = await self.request(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): diff --git a/tests/test_runtime_endpoints.py b/tests/test_runtime_endpoints.py index ef5ac8f..4267ae8 100644 --- a/tests/test_runtime_endpoints.py +++ b/tests/test_runtime_endpoints.py @@ -630,7 +630,7 @@ async def send(message): sent.append(message) import asyncio - asyncio.run(middleware({"type": "http", "path": "/v1/chat/completions"}, receive, send)) + asyncio.run(middleware({"type": "http", "method": "POST", "path": "/v1/chat/completions"}, receive, send)) self.assertFalse(reached) # 超限请求不进入下游 self.assertEqual(sent[0]["status"], 413) From 7b8357d58cb2e1975daebf4a480b0f46457d6d42 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 22:36:43 +0800 Subject: [PATCH 22/23] Reject oversized credential timestamps without float conversion Compare timestamp bounds on the original integer or float so huge JSON integers return validation errors instead of OverflowError. Preserve existing finite timestamp bounds and NaN/Infinity/bool rejection. Cover both timestamp fields, signed 1001-digit integers, valid boundaries, and controlled-file imports returning HTTP 400 without storing or disclosing credentials. --- app/auth_oauth.py | 5 ++--- tests/test_auth_oauth.py | 14 ++++++++++---- tests/test_security.py | 20 ++++++++++++++++++++ 3 files changed, 32 insertions(+), 7 deletions(-) diff --git a/app/auth_oauth.py b/app/auth_oauth.py index e3beafc..2506c8c 100644 --- a/app/auth_oauth.py +++ b/app/auth_oauth.py @@ -15,7 +15,6 @@ import base64 import copy import json -import math import re import threading import time @@ -99,9 +98,9 @@ def validate_cred_data(data) -> tuple[str | None, str | None]: value = auth.get(field) if value is None: continue - # bool 是 int 子类必须显式排除；NaN/Infinity 会让到期判断与 JSON 序列化行为异常 + # 原值范围比较同时拒绝非有限浮点数，避免超大整数转 float 溢出。 if isinstance(value, bool) or not isinstance(value, (int, float)) \ - or not math.isfinite(value) or not 0 < float(value) < 4102444800000: # 上限 2100-01-01 + 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) diff --git a/tests/test_auth_oauth.py b/tests/test_auth_oauth.py index 88a21e8..0e1c912 100644 --- a/tests/test_auth_oauth.py +++ b/tests/test_auth_oauth.py @@ -56,10 +56,16 @@ def test_validate_cred_data(): uid, err = validate_cred_data(bad) assert uid is None and "允许列表" in err # 时间戳必须有限且合理：NaN/Infinity/bool/0/超范围都拒绝 - for bad_ts in (float("nan"), float("inf"), True, 0, -1, 99999999999999): - c = _cred() - c["auth"]["expiresAt"] = bad_ts - assert validate_cred_data(c)[1], bad_ts + 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) 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) From 0281eef1c311d7bbb389b78ed66ab9557096a507 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 22:38:02 +0800 Subject: [PATCH 23/23] Stop error-body reads as soon as the byte budget is filled Check the buffer after each append so an exact-limit chunk returns without waiting for another upstream event. A zero budget performs no read. HTTPX stream regressions verify single/split/oversized chunks, short bodies, no extra pull after the cap, and response cleanup. --- app/upstream_io.py | 4 +++- tests/test_upstream_io.py | 36 ++++++++++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+), 1 deletion(-) diff --git a/app/upstream_io.py b/app/upstream_io.py index 3e72606..a4267de 100644 --- a/app/upstream_io.py +++ b/app/upstream_io.py @@ -159,11 +159,13 @@ def result(self): 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 - buf.extend(chunk[: max(0, limit - len(buf))]) return bytes(buf) diff --git a/tests/test_upstream_io.py b/tests/test_upstream_io.py index d9eb093..09365a4 100644 --- a/tests/test_upstream_io.py +++ b/tests/test_upstream_io.py @@ -250,6 +250,42 @@ async def aiter_bytes(self): 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",