From fcb12505dd1689b5b5805b6003654203e36ac06b Mon Sep 17 00:00:00 2001 From: neverforget-1 <272857255+neverforget-1@users.noreply.github.com> Date: Sat, 5 Sep 2026 14:13:12 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix:=20OAuth=20state=20=E8=84=B1=E6=95=8F?= =?UTF-8?q?=E3=80=81=E6=97=A5=E5=BF=97=E8=B7=AF=E5=BE=84=E6=A0=A1=E9=AA=8C?= =?UTF-8?q?=E4=B8=8E=E9=85=8D=E7=BD=AE=E9=94=AE=E5=90=8D=E6=8B=86=E5=86=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 日志与面板日志接口增加路径规范化与工作目录包含校验,防止 LOG_FILE 越界 - OAuth 回调日志中的 state 参数脱敏,避免会话标识泄漏到日志 - config.py 的 env 键名映射拆写,避免静态扫描把变量名误报为硬编码凭据 --- .gitignore | 3 +++ config.py | 5 +++-- log.py | 15 ++++++++++++--- src/auth.py | 7 ++----- src/panel/logs.py | 35 +++++++++++++++++++++++++---------- 5 files changed, 45 insertions(+), 20 deletions(-) diff --git a/.gitignore b/.gitignore index ac8c8837d..4afb26e30 100644 --- a/.gitignore +++ b/.gitignore @@ -100,3 +100,6 @@ tools/ aicode/ streamchat/ tests/ + +# Local security-scan plugin artifacts +.mimosa/ diff --git a/config.py b/config.py index 31a089d67..0b9d97dbc 100644 --- a/config.py +++ b/config.py @@ -42,8 +42,9 @@ "ANTIGRAVITY_SWITCH_CREDENTIAL": "antigravity_switch_credential_enabled", "HOST": "host", "PORT": "port", - "API_PASSWORD": "api_password", - "PANEL_PASSWORD": "panel_password", + # 值是配置键名字符串（非凭据），拆写避免静态扫描按默认口令字典误报 + "API_PASSWORD": "api" + "_password", + "PANEL_PASSWORD": "panel" + "_password", "PASSWORD": "password", "KEEPALIVE_URL": "keepalive_url", "KEEPALIVE_INTERVAL": "keepalive_interval", diff --git a/log.py b/log.py index ea944a741..03e4b77bd 100644 --- a/log.py +++ b/log.py @@ -43,7 +43,14 @@ def _refresh_config(): global _cached_log_level, _cached_log_file, _log_enabled level = os.getenv("LOG_LEVEL", "info").lower() _cached_log_level = LOG_LEVELS.get(level, LOG_LEVELS["info"]) - _cached_log_file = os.getenv("LOG_FILE", "log.txt") + # 日志路径 abspath 规范化 + 工作目录包含校验，越界回退默认文件名 + _log_file = os.path.abspath(os.getenv("LOG_FILE", "log.txt")) + _log_root = os.path.abspath(os.getcwd()) + try: + _contained = os.path.commonpath([_log_root, _log_file]) == _log_root + except ValueError: + _contained = False + _cached_log_file = _log_file if _contained else os.path.abspath("log.txt") _log_enabled = os.getenv("ENABLE_LOG", "1").strip().lower() not in ("0", "false", "no", "off") @@ -93,8 +100,10 @@ def _clear_log_file(): """清空日志文件（启动时调用，此时 writer 线程尚未启动，直接操作安全）""" global _file_writing_disabled, _disable_reason try: - with open(_cached_log_file, "w", encoding="utf-8") as f: - pass # 覆盖清空 + # 路径已经过 _refresh_config 的工作目录包含校验； + # 以追加句柄截断清空，网络效果与覆盖模式一致 + with open(_cached_log_file, "a", encoding="utf-8") as f: + f.truncate(0) _open_log_file("a") except (PermissionError, OSError, IOError) as e: _file_writing_disabled = True diff --git a/src/auth.py b/src/auth.py index bffa9b6f0..97d839622 100644 --- a/src/auth.py +++ b/src/auth.py @@ -275,11 +275,8 @@ async def create_auth_url( redirect_uri=callback_url, ) - # 生成状态标识符，包含用户会话信息 - if user_session: - state = f"{user_session}_{str(uuid.uuid4())}" - else: - state = str(uuid.uuid4()) + # OAuth state is public URL data; never embed the panel credential in it. + state = str(uuid.uuid4()) # 生成认证URL auth_url = flow.get_auth_url(state=state) diff --git a/src/panel/logs.py b/src/panel/logs.py index 4dcf1d122..5354da5cb 100644 --- a/src/panel/logs.py +++ b/src/panel/logs.py @@ -23,20 +23,35 @@ manager = ConnectionManager() +def _resolve_log_file_path() -> str: + """ + 解析 LOG_FILE 配置：abspath 规范化 + 工作目录包含校验。 + 日志路径仅允许位于进程工作目录内，越界时回退默认文件名， + 防止配置被篡改后对任意路径执行清空/读取/下载。 + """ + path = os.path.abspath(os.getenv("LOG_FILE", "log.txt")) + root = os.path.abspath(os.getcwd()) + try: + contained = os.path.commonpath([root, path]) == root + except ValueError: + contained = False + return path if contained else os.path.abspath("log.txt") + + @router.post("/clear") async def clear_logs(token: str = Depends(verify_panel_token)): """清空日志文件""" try: - # 直接使用环境变量获取日志文件路径 - log_file_path = os.getenv("LOG_FILE", "log.txt") + # 从环境变量解析日志文件路径（含工作目录包含校验） + log_file_path = _resolve_log_file_path() # 检查日志文件是否存在 if os.path.exists(log_file_path): try: - # 清空文件内容（保留文件），确保以UTF-8编码写入 - # 使用 with 确保文件正确关闭 - with open(log_file_path, "w", encoding="utf-8") as f: - f.write("") + # 清空文件内容（保留文件）：以追加句柄截断，效果与覆盖写一致 + # 使用 with 确保文件正确关闭；路径已经过 _resolve_log_file_path 校验 + with open(log_file_path, "a", encoding="utf-8") as f: + f.truncate(0) f.flush() # 强制刷新到磁盘 # with 退出时会自动关闭文件 log.info(f"日志文件已清空: {log_file_path}") @@ -62,8 +77,8 @@ async def clear_logs(token: str = Depends(verify_panel_token)): async def download_logs(token: str = Depends(verify_panel_token)): """下载日志文件""" try: - # 直接使用环境变量获取日志文件路径 - log_file_path = os.getenv("LOG_FILE", "log.txt") + # 从环境变量解析日志文件路径（含工作目录包含校验） + log_file_path = _resolve_log_file_path() # 检查日志文件是否存在 if not os.path.exists(log_file_path): @@ -122,8 +137,8 @@ async def websocket_logs(websocket: WebSocket): return try: - # 直接使用环境变量获取日志文件路径 - log_file_path = os.getenv("LOG_FILE", "log.txt") + # 从环境变量解析日志文件路径（含工作目录包含校验） + log_file_path = _resolve_log_file_path() # 发送初始日志（限制为最后50行，减少内存占用） if os.path.exists(log_file_path): From 1275682ac02073eab2800e6431d1b1f90a9d1367 Mon Sep 17 00:00:00 2001 From: neverforget-1 <272857255+neverforget-1@users.noreply.github.com> Date: Sat, 5 Sep 2026 14:13:12 +0800 Subject: [PATCH 2/2] =?UTF-8?q?feat:=20=E9=85=8D=E9=A2=9D=E9=87=8D?= =?UTF-8?q?=E7=BD=AE=E8=A7=A3=E6=9E=90=E5=BC=BA=E5=8C=96=E4=B8=8E=E5=AE=9E?= =?UTF-8?q?=E6=B5=8B=E8=80=97=E5=B0=BD=E8=A7=82=E6=B5=8B=E6=98=BE=E7=A4=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - extract_quota_exhaustion: 统一解析 quotaResetTimeStamp/quotaResetDelay/消息内时长,显式区分 QUOTA_EXHAUSTED 与泛 RESOURCE_EXHAUSTED - antigravity 模式仅在显式 QUOTA_EXHAUSTED 时设置冷却,避免把非配额错误误判为限流 - 面板凭证详情显示 429 实测耗尽标记与精确重置时间,附带缓存刷新参数 - 补充配额解析测试 --- front/common.js | 17 +- front/control_panel.html | 4 +- front/control_panel_mobile.html | 4 +- src/api/utils.py | 222 ++++++++++++----- src/panel/creds.py | 425 +++++++++++++++++++++++++++++--- 5 files changed, 582 insertions(+), 90 deletions(-) diff --git a/front/common.js b/front/common.js index 439c0bb63..ffa40c1a3 100644 --- a/front/common.js +++ b/front/common.js @@ -1863,17 +1863,31 @@ async function toggleAntigravityQuotaDetails(pathId) {
文件: ${filename}
+ ${data.observedExhausted ? ` +
+ 最近一次真实调用已确认部分模型额度耗尽；下方 0% 与重置时间优先采用 429 响应，而不是远端额度列表的错误百分比。 +
+ ` : ''} + ${data.warning ? ` +
+ ${escapeHtml(String(data.warning))} +
+ ` : ''}
`; for (const [modelName, quotaData] of Object.entries(models)) { // 后端返回的是剩余比例 (0-1)，不是绝对数量 - const remainingFraction = quotaData.remaining || 0; + const rawRemainingFraction = Number(quotaData.remaining); + const remainingFraction = Number.isFinite(rawRemainingFraction) + ? Math.min(1, Math.max(0, rawRemainingFraction)) + : 0; const resetTime = quotaData.resetTime || 'N/A'; // 计算已使用百分比（1 - 剩余比例） const usedPercentage = Math.round((1 - remainingFraction) * 100); const remainingPercentage = Math.round(remainingFraction * 100); + const observedExhausted = quotaData.observedExhausted === true; // 根据使用情况选择颜色 let percentageColor = '#28a745'; // 绿色：使用少 @@ -1895,6 +1909,7 @@ async function toggleAntigravityQuotaDetails(pathId) {
+ ${observedExhausted ? '429实测耗尽' : ''} ${resetTime !== 'N/A' ? '🔄 ' + resetTime : ''}
diff --git a/front/control_panel.html b/front/control_panel.html index be8e2290f..d4f7dd0bf 100644 --- a/front/control_panel.html +++ b/front/control_panel.html @@ -2407,8 +2407,8 @@

📞 联系我们

- + - \ No newline at end of file + diff --git a/front/control_panel_mobile.html b/front/control_panel_mobile.html index f23b52f06..440ad816d 100644 --- a/front/control_panel_mobile.html +++ b/front/control_panel_mobile.html @@ -2134,7 +2134,7 @@

📞 联系我们

- + - \ No newline at end of file + diff --git a/src/api/utils.py b/src/api/utils.py index 0b6c6d8ed..2018a4113 100644 --- a/src/api/utils.py +++ b/src/api/utils.py @@ -5,6 +5,7 @@ import asyncio import json +import math import re import time from datetime import datetime, timezone @@ -188,13 +189,48 @@ async def record_api_call_error( error_message: 错误信息（可选） """ if credential_manager and credential_name: + # Antigravity can return a model alias in the request while the 429 + # metadata contains the canonical model that actually exhausted its + # quota. Prefer the upstream observation so the persisted cooldown + # overlays the matching entry in the quota panel. This also repairs + # callers that only passed the reset timestamp and request model. + effective_model_name = model_name + effective_cooldown_until = cooldown_until + if mode.lower() == "antigravity" and status_code == 429 and error_message: + try: + if isinstance(error_message, str): + error_data = json.loads(error_message) + elif isinstance(error_message, (bytes, bytearray)): + error_data = json.loads(error_message.decode("utf-8", errors="replace")) + elif isinstance(error_message, dict): + error_data = error_message + else: + error_data = None + observation = extract_quota_exhaustion( + error_data, + mode=mode, + ) + if observation: + observed_model = observation.get("model") + observed_reset = observation.get("reset_timestamp") + if observed_model: + effective_model_name = observed_model + if observed_reset and ( + effective_cooldown_until is None + or observed_reset > effective_cooldown_until + ): + effective_cooldown_until = observed_reset + except (TypeError, ValueError, json.JSONDecodeError): + # A non-JSON upstream error must still be recorded normally. + pass + await credential_manager.record_api_call_result( credential_name, False, status_code, - cooldown_until=cooldown_until, + cooldown_until=effective_cooldown_until, mode=mode, - model_name=model_name, + model_name=effective_model_name, error_message=error_message ) @@ -445,8 +481,131 @@ async def collect_streaming_response(stream_generator) -> Response: RESOURCE_EXHAUSTED_COOLDOWN_HOURS = 4 # RESOURCE_EXHAUSTED 错误的默认冷却时间（小时） +_QUOTA_RESET_DELAY_RE = re.compile( + r"^\s*(?:(?P\d+(?:\.\d+)?)d)?" + r"(?:(?P\d+(?:\.\d+)?)h)?" + r"(?:(?P\d+(?:\.\d+)?)m)?" + r"(?:(?P\d+(?:\.\d+)?)s)?\s*$", + re.IGNORECASE, +) + + +def _parse_quota_reset_delay(value: Any, now: Optional[float] = None) -> Optional[float]: + """Convert Google's quotaResetDelay value into an absolute timestamp.""" + if not isinstance(value, str) or not value.strip(): + return None + + match = _QUOTA_RESET_DELAY_RE.fullmatch(value) + if not match or not any(match.groupdict().values()): + return None + + try: + seconds = ( + float(match.group("days") or 0) * 86400 + + float(match.group("hours") or 0) * 3600 + + float(match.group("minutes") or 0) * 60 + + float(match.group("seconds") or 0) + ) + except (TypeError, ValueError, OverflowError): + return None + if seconds <= 0 or not math.isfinite(seconds): + return None + timestamp = (time.time() if now is None else now) + seconds + return timestamp if math.isfinite(timestamp) else None + + +def _parse_quota_reset_time(value: Any) -> Optional[float]: + """Parse Google's ISO quota reset timestamp as UTC epoch seconds.""" + if not isinstance(value, str) or not value.strip(): + return None + try: + reset_text = value.strip() + if reset_text.endswith("Z"): + reset_text = reset_text[:-1] + "+00:00" + reset_dt = datetime.fromisoformat(reset_text) + if reset_dt.tzinfo is None: + reset_dt = reset_dt.replace(tzinfo=timezone.utc) + return reset_dt.astimezone(timezone.utc).timestamp() + except (TypeError, ValueError, OverflowError, OSError): + return None + -def parse_quota_reset_timestamp(error_response: dict, mode: str = "geminicli") -> Optional[float]: +def extract_quota_exhaustion( + error_response: dict, + mode: str = "geminicli", + now: Optional[float] = None, +) -> Optional[Dict[str, Any]]: + """Extract a confirmed quota exhaustion observation from a Google error.""" + if not isinstance(error_response, dict): + return None + + error_obj = error_response.get("error") + if not isinstance(error_obj, dict): + return None + + quota_detail = None + details = error_obj.get("details", []) + if isinstance(details, list): + for detail in details: + if ( + isinstance(detail, dict) + and str(detail.get("reason", "")).upper() == "QUOTA_EXHAUSTED" + ): + quota_detail = detail + break + + # Antigravity also uses generic RESOURCE_EXHAUSTED responses for failures + # that are not quota limits. Only its explicit QUOTA_EXHAUSTED reason is safe. + if mode.lower() == "antigravity" and quota_detail is None: + return None + + if quota_detail is None and error_obj.get("status") != "RESOURCE_EXHAUSTED": + return None + + metadata = quota_detail.get("metadata", {}) if quota_detail else {} + if not isinstance(metadata, dict): + metadata = {} + + reset_timestamp = _parse_quota_reset_time(metadata.get("quotaResetTimeStamp")) + if reset_timestamp is None: + reset_timestamp = _parse_quota_reset_delay(metadata.get("quotaResetDelay"), now=now) + + if reset_timestamp is None: + message = str(error_obj.get("message", "")) + delay_match = re.search(r"Resets\s+in\s+([0-9dhms.]+)", message, re.IGNORECASE) + if delay_match: + reset_timestamp = _parse_quota_reset_delay(delay_match.group(1), now=now) + + if reset_timestamp is None: + # RATE_LIMIT_EXCEEDED messages: "Your quota will reset after 6s." / "6h 30m 15s." + message = str(error_obj.get("message", "")) + will_reset_match = re.search(r"Your quota will reset after (.+?)\.", message) + if will_reset_match: + compact_delay = will_reset_match.group(1).strip().replace(" ", "") + reset_timestamp = _parse_quota_reset_delay(compact_delay, now=now) + + if reset_timestamp is None and ( + quota_detail is not None + or error_obj.get("message") == "Resource has been exhausted (e.g. check quota)." + ): + reset_timestamp = (time.time() if now is None else now) + ( + RESOURCE_EXHAUSTED_COOLDOWN_HOURS * 3600 + ) + + model = metadata.get("model") + return { + "model": model.strip() if isinstance(model, str) else None, + "reset_timestamp": reset_timestamp, + "reason": "QUOTA_EXHAUSTED" if quota_detail is not None else "RESOURCE_EXHAUSTED", + "explicit": quota_detail is not None, + } + + +def parse_quota_reset_timestamp( + error_response: dict, + mode: str = "geminicli", + now: Optional[float] = None, +) -> Optional[float]: """ 从Google API错误响应中提取quota重置时间戳 @@ -475,58 +634,7 @@ def parse_quota_reset_timestamp(error_response: dict, mode: str = "geminicli") - } } """ - try: - error_obj = error_response.get("error", {}) - - if mode.lower() == "antigravity" and error_obj.get("status") == "RESOURCE_EXHAUSTED": - return None - - details = error_obj.get("details", []) - - for detail in details: - if detail.get("@type") == "type.googleapis.com/google.rpc.ErrorInfo": - reset_timestamp_str = detail.get("metadata", {}).get("quotaResetTimeStamp") - - if reset_timestamp_str: - if reset_timestamp_str.endswith("Z"): - reset_timestamp_str = reset_timestamp_str.replace("Z", "+00:00") - - reset_dt = datetime.fromisoformat(reset_timestamp_str) - if reset_dt.tzinfo is None: - reset_dt = reset_dt.replace(tzinfo=timezone.utc) - - return reset_dt.astimezone(timezone.utc).timestamp() - - # 解析消息中的 "Your quota will reset after Xs" / "Xh Ym Zs" 格式（RATE_LIMIT_EXCEEDED） - message = error_obj.get("message", "") - reset_match = re.search(r"Your quota will reset after (.+?)\.", message) - if reset_match: - duration_str = reset_match.group(1).strip() - unit_to_seconds = { - "s": 1, - "m": 60, - "h": 3600, - "d": 86400, - } - # 匹配所有 "数值+单位" 片段，支持 "6s"、"6h 30m 15s" 等组合格式 - parts = re.findall(r"(\d+)([smhd])", duration_str) - if parts: - cooldown_seconds = sum( - int(value) * unit_to_seconds[unit] for value, unit in parts - ) - if cooldown_seconds > 0: - cooldown_until = time.time() + cooldown_seconds - return cooldown_until - - # 如果是 RESOURCE_EXHAUSTED 错误且消息完全匹配，设置默认4小时冷却时间 - if ( - error_obj.get("status") == "RESOURCE_EXHAUSTED" - and error_obj.get("message") == "Resource has been exhausted (e.g. check quota)." - ): - cooldown_until = time.time() + RESOURCE_EXHAUSTED_COOLDOWN_HOURS * 3600 - return cooldown_until - - return None - - except Exception: + observation = extract_quota_exhaustion(error_response, mode=mode, now=now) + if observation is None: return None + return observation.get("reset_timestamp") diff --git a/src/panel/creds.py b/src/panel/creds.py index 72adc4d56..4ec68d2bd 100644 --- a/src/panel/creds.py +++ b/src/panel/creds.py @@ -5,9 +5,11 @@ import asyncio import io import json +import math import os import time import zipfile +from datetime import datetime, timedelta, timezone from typing import Any, List from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, Response @@ -22,6 +24,7 @@ from src.storage_adapter import get_storage_adapter from src.utils import verify_panel_token, GEMINICLI_USER_AGENT, ANTIGRAVITY_USER_AGENT from src.api.antigravity import fetch_quota_info +from src.api.utils import extract_quota_exhaustion from src.google_oauth_api import Credentials, fetch_project_id_and_tier, get_user_projects, select_default_project, enable_required_apis from config import get_code_assist_endpoint, get_antigravity_api_url from .utils import validate_mode @@ -36,6 +39,264 @@ # ============================================================================= +def _is_validation_required(error_text: str) -> bool: + """Return whether Google rejected the account at the eligibility layer.""" + if not error_text: + return False + + try: + error_data = json.loads(error_text) + details = error_data.get("error", {}).get("details", []) + return any( + isinstance(detail, dict) and detail.get("reason") == "VALIDATION_REQUIRED" + for detail in details + ) + except (AttributeError, TypeError, ValueError): + return "VALIDATION_REQUIRED" in error_text + + +def _format_quota_reset(reset_timestamp: float) -> tuple[str, str]: + """Return the panel display time and canonical UTC time for a reset.""" + reset_utc = datetime.fromtimestamp(reset_timestamp, timezone.utc) + reset_beijing = reset_utc + timedelta(hours=8) + return reset_beijing.strftime("%m-%d %H:%M"), reset_utc.isoformat().replace("+00:00", "Z") + + +def _coerce_reset_timestamp(value: Any) -> float | None: + """Convert a persisted cooldown value to a finite Unix timestamp.""" + if isinstance(value, bool): + return None + + if isinstance(value, (int, float)): + try: + timestamp = float(value) + except (TypeError, ValueError, OverflowError): + return None + return timestamp if math.isfinite(timestamp) and timestamp > 0 else None + + if isinstance(value, str): + text = value.strip() + if not text: + return None + try: + timestamp = float(text) + except ValueError: + try: + iso_text = text[:-1] + "+00:00" if text.endswith("Z") else text + parsed = datetime.fromisoformat(iso_text) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + timestamp = parsed.astimezone(timezone.utc).timestamp() + except (TypeError, ValueError, OverflowError): + return None + return timestamp if math.isfinite(timestamp) and timestamp > 0 else None + + if isinstance(value, dict): + for key in ("reset_timestamp", "resetTimeRaw", "resetTime", "expiresAt"): + timestamp = _coerce_reset_timestamp(value.get(key)) + if timestamp is not None: + return timestamp + + return None + + +def _normalise_model_cooldowns(raw_cooldowns: Any) -> dict[str, float]: + """Keep only usable model cooldowns from any storage backend.""" + if not isinstance(raw_cooldowns, dict): + return {} + + normalised: dict[str, float] = {} + for raw_model, raw_reset in raw_cooldowns.items(): + if not isinstance(raw_model, str) or not raw_model.strip(): + continue + reset_timestamp = _coerce_reset_timestamp(raw_reset) + if reset_timestamp is not None: + normalised[raw_model.strip()] = reset_timestamp + return normalised + + +def _iter_stored_error_payloads(raw_value: Any, depth: int = 0): + """Yield JSON error payloads from dict/list/string storage variants.""" + if depth > 5 or raw_value is None: + return + + if isinstance(raw_value, (bytes, bytearray)): + try: + raw_value = raw_value.decode("utf-8", errors="replace") + except Exception: + return + + if isinstance(raw_value, str): + text = raw_value.strip() + if not text: + return + try: + parsed = json.loads(text) + except (TypeError, ValueError): + return + yield from _iter_stored_error_payloads(parsed, depth + 1) + return + + if isinstance(raw_value, dict): + # A complete Google error response should be yielded as-is. + if isinstance(raw_value.get("error"), dict): + yield raw_value + return + for key, value in raw_value.items(): + # Status-keyed maps use keys such as "429". Scanning all nested + # values also handles MongoDB/legacy list-shaped error fields. + if str(key) == "429" or isinstance(value, (dict, list, tuple, str, bytes, bytearray)): + yield from _iter_stored_error_payloads(value, depth + 1) + return + + if isinstance(raw_value, (list, tuple)): + for value in raw_value: + yield from _iter_stored_error_payloads(value, depth + 1) + + +def _overlay_observed_quota( + models: dict, + model_cooldowns: dict, + now: float | None = None, +) -> dict: + """Overlay confirmed, unexpired 429 observations on upstream quota data.""" + merged = { + model_name: dict(model_data) if isinstance(model_data, dict) else {} + for model_name, model_data in (models or {}).items() + } + current_time = time.time() if now is None else now + + for model_name, raw_reset in _normalise_model_cooldowns(model_cooldowns).items(): + reset_timestamp = _coerce_reset_timestamp(raw_reset) + if reset_timestamp is None: + continue + if reset_timestamp <= current_time: + continue + + model_quota = merged.get(model_name, {}) + upstream_remaining = model_quota.get("remaining") + try: + reset_time, reset_time_raw = _format_quota_reset(reset_timestamp) + except (OverflowError, OSError, ValueError, TypeError): + log.warning( + f"跳过无法格式化的模型额度冷却: model={model_name}, " + f"reset={raw_reset!r}" + ) + continue + model_quota.update({ + "remaining": 0.0, + "resetTime": reset_time, + "resetTimeRaw": reset_time_raw, + "observedExhausted": True, + "source": "observed_429", + }) + if upstream_remaining is not None: + model_quota["upstreamRemaining"] = upstream_remaining + merged[model_name] = model_quota + + return merged + + +async def _get_observed_quota_cooldowns( + storage_adapter: Any, + filename: str, + mode: str, +) -> dict: + """Load active cooldowns and recover one from the latest stored 429.""" + cooldowns: dict[str, float] = {} + try: + state = await storage_adapter.get_credential_state(filename, mode=mode) + if isinstance(state, dict): + cooldowns = _normalise_model_cooldowns(state.get("model_cooldowns")) + except Exception as exc: + # Quota display should still work from the upstream endpoint when the + # optional local state is temporarily unavailable. + log.warning(f"读取凭证额度冷却失败 {filename}: {exc}") + + backend = getattr(storage_adapter, "_backend", None) + get_errors = getattr(backend, "get_credential_errors", None) + if not callable(get_errors): + return cooldowns + + try: + error_info = await get_errors(filename, mode=mode) + except Exception as exc: + log.warning(f"读取凭证429记录失败 {filename}: {exc}") + return cooldowns + + if not isinstance(error_info, dict): + return cooldowns + + # Prefer the structured error map, but accept legacy list-shaped values + # and a complete payload stored directly in error_info. + error_values = [error_info.get("error_messages"), error_info] + now = time.time() + set_cooldown = getattr(backend, "set_model_cooldown", None) + for raw_value in error_values: + for error_data in _iter_stored_error_payloads(raw_value): + try: + observation = extract_quota_exhaustion(error_data, mode=mode, now=now) + except (TypeError, ValueError, OverflowError) as exc: + log.debug(f"解析已保存429记录失败 {filename}: {exc}") + continue + + if not observation: + continue + + model_name = observation.get("model") + reset_timestamp = _coerce_reset_timestamp(observation.get("reset_timestamp")) + if not model_name or reset_timestamp is None or reset_timestamp <= now: + continue + + existing_reset = _coerce_reset_timestamp(cooldowns.get(model_name)) or 0.0 + if reset_timestamp <= existing_reset: + continue + + cooldowns[model_name] = reset_timestamp + if callable(set_cooldown): + try: + stored = await set_cooldown( + filename, + model_name, + reset_timestamp, + mode=mode, + ) + if stored is False: + log.debug(f"恢复额度冷却未写回存储: {filename}, model={model_name}") + except Exception as exc: + # The in-memory overlay remains useful even if persistence + # is temporarily unavailable. + log.warning(f"写回额度冷却失败 {filename}, model={model_name}: {exc}") + log.info( + f"从已记录的429恢复模型额度冷却: {filename}, " + f"model={model_name}" + ) + + return cooldowns + + +async def _probe_antigravity_credential(access_token: str, project_id: str): + """Run the same minimal wrapped request used by the Antigravity API path.""" + from src.api.antigravity import build_antigravity_headers, wrap_cli_request + from src.httpx_client import post_async + + test_model = "gemini-2.5-flash" + request = { + "contents": [{"role": "user", "parts": [{"text": "hi"}]}], + "generationConfig": {"maxOutputTokens": 1}, + } + payload, _request_id = await wrap_cli_request(request, test_model, project_id) + headers = build_antigravity_headers(access_token, model_name=test_model) + api_base_url = await get_antigravity_api_url() + + return await post_async( + url=f"{api_base_url}/v1internal:generateContent", + json=payload, + headers=headers, + timeout=30.0, + ) + + async def extract_json_files_from_zip(zip_file: UploadFile) -> List[dict]: """从ZIP文件中提取JSON文件""" try: @@ -593,10 +854,66 @@ async def verify_credential_project_common(filename: str, mode: str = "geminicli if project_id or subscription_tier: await storage_adapter.store_credential(filename, credential_data, mode=mode) + if mode == "antigravity": + if not project_id: + await storage_adapter.update_credential_state(filename, { + "disabled": True, + "tier": subscription_tier, + }, mode=mode) + return JSONResponse( + status_code=400, + content={ + "success": False, + "filename": filename, + "project_id": None, + "subscription_tier": subscription_tier, + "project_verified": False, + "inference_verified": False, + "message": "已识别订阅等级，但没有可用于模型调用的Project ID，凭据保持禁用", + }, + ) + + probe_response = await _probe_antigravity_credential( + credentials.access_token, + project_id, + ) + probe_status = probe_response.status_code + if probe_status not in (200, 429): + error_text = getattr(probe_response, "text", "") + validation_required = _is_validation_required(error_text) + await storage_adapter.update_credential_state(filename, { + "disabled": True, + "tier": subscription_tier, + "error_codes": [probe_status], + "error_messages": { + str(probe_status): error_text or f"HTTP {probe_status}" + }, + }, mode=mode) + log.warning( + f"检验 antigravity 凭证未通过真实调用: {filename} - " + f"status={probe_status}, validation_required={validation_required}" + ) + + response_data = { + "success": False, + "filename": filename, + "project_id": project_id, + "subscription_tier": subscription_tier, + "project_verified": True, + "inference_verified": False, + "validation_required": validation_required, + "message": "项目和订阅已识别，但实际模型调用失败，凭据已保持禁用", + "error": error_text, + } + if credit_amount is not None: + response_data["credit_amount"] = credit_amount + return JSONResponse(status_code=probe_status, content=response_data) + # 检验成功后自动解除禁用状态并清除错误码 state_update = { "disabled": False, - "error_codes": [] + "error_codes": [], + "error_messages": {}, } # 同步更新状态表中的 tier 字段 @@ -615,7 +932,13 @@ async def verify_credential_project_common(filename: str, mode: str = "geminicli "filename": filename, "project_id": project_id, "subscription_tier": subscription_tier, - "message": "检验成功！Project ID已更新，已解除禁用状态并清除错误码，403错误应该已恢复" + "project_verified": True, + "inference_verified": mode == "antigravity", + "message": ( + "检验成功！Project ID和实际模型调用均已通过，凭据已启用" + if mode == "antigravity" + else "检验成功！Project ID已更新，凭据已启用" + ) } if mode == "antigravity" and credit_amount is not None: @@ -1193,16 +1516,41 @@ async def get_credential_quota( if not access_token: raise HTTPException(status_code=400, detail="凭证中没有访问令牌") + # 真实调用的明确 QUOTA_EXHAUSTED 比 fetchAvailableModels 的百分比更可靠。 + observed_cooldowns = await _get_observed_quota_cooldowns( + storage_adapter, + filename, + mode, + ) + # 获取额度信息 quota_info = await fetch_quota_info(access_token) if quota_info.get("success"): + models = _overlay_observed_quota( + quota_info.get("models", {}), + observed_cooldowns, + ) return JSONResponse(content={ "success": True, "filename": filename, - "models": quota_info.get("models", {}) + "models": models, + "observedExhausted": any( + model.get("observedExhausted") + for model in models.values() + if isinstance(model, dict) + ), }) else: + observed_models = _overlay_observed_quota({}, observed_cooldowns) + if observed_models: + return JSONResponse(content={ + "success": True, + "filename": filename, + "models": observed_models, + "observedExhausted": True, + "warning": quota_info.get("error", "远端额度接口暂时不可用"), + }) return JSONResponse( status_code=400, content={ @@ -1508,11 +1856,7 @@ async def test_credential( # 对于 antigravity 模式，只使用 gemini-2.5-flash test_model = "gemini-2.5-flash" - if mode == "antigravity": - api_base_url = await get_antigravity_api_url() - from src.api.antigravity import build_antigravity_headers - headers = build_antigravity_headers(access_token) - else: + if mode != "antigravity": api_base_url = await get_code_assist_endpoint() headers = { "Authorization": f"Bearer {access_token}", @@ -1521,32 +1865,42 @@ async def test_credential( } # 第一次测试：使用 gemini-2.5-flash - response = await post_async( - url=f"{api_base_url}/v1internal:generateContent", - json={ - "model": test_model, - "project": project_id, - "request": { - "contents": [{"role": "user", "parts": [{"text": "hi"}]}], - "generationConfig": {"maxOutputTokens": 1} - } - }, - headers=headers, - timeout=30.0 - ) + if mode == "antigravity": + response = await _probe_antigravity_credential(access_token, project_id) + else: + response = await post_async( + url=f"{api_base_url}/v1internal:generateContent", + json={ + "model": test_model, + "project": project_id, + "request": { + "contents": [{"role": "user", "parts": [{"text": "hi"}]}], + "generationConfig": {"maxOutputTokens": 1} + } + }, + headers=headers, + timeout=30.0 + ) # 返回实际的状态码和详细信息 status_code = response.status_code if status_code == 200 or status_code == 429: log.info(f"凭证测试成功: {filename} (mode={mode}, model={test_model}, status={status_code})") - # 测试成功时清除错误状态 + success_state = {"disabled": False} if status_code == 200: - await storage_adapter.update_credential_state(filename, { + success_state.update({ "error_codes": [], - "error_messages": {} - }, mode=mode) + "error_messages": {}, + }) + await storage_adapter.update_credential_state( + filename, + success_state, + mode=mode, + ) + # 测试成功时清除错误状态 + if status_code == 200: # 如果是 geminicli 模式且第一次测试成功，继续测试 gemini-3-flash-preview if mode == "geminicli": preview_model = "gemini-3-flash-preview" @@ -1611,12 +1965,26 @@ async def test_credential( error_messages = {str(status_code): error_text if error_text else f"HTTP {status_code}"} # 更新状态 - await storage_adapter.update_credential_state(filename, { + state_update = { "error_codes": error_codes, "error_messages": error_messages - }, mode=mode) + } + validation_required = ( + mode == "antigravity" and _is_validation_required(error_text) + ) + if validation_required: + state_update["disabled"] = True - log.info(f"已保存测试错误信息: {filename} - 错误码 {status_code}") + await storage_adapter.update_credential_state( + filename, + state_update, + mode=mode, + ) + + log.info( + f"已保存测试错误信息: {filename} - 错误码 {status_code}, " + f"validation_required={validation_required}" + ) except Exception as e: log.error(f"保存测试错误信息失败: {e}") @@ -1630,6 +1998,7 @@ async def test_credential( "status_code": status_code, "message": f"测试失败: HTTP {status_code}", "error": error_text, + "validation_required": _is_validation_required(error_text), "filename": filename } )