refactor(kiro): 拆分 auth_region/api_region,统一 region 解析逻辑

KiroAuthConfig 新增 auth_region(token 刷新端点)和 api_region(q.{region} 服务端点)字段,
通过 effective_auth_region() / effective_api_region() 方法统一各处散落的 region 回退逻辑,
与 kiro.rs 的 region 语义对齐。
This commit is contained in:
fawney19
2026-02-21 16:54:03 +08:00
parent f470ab6ec8
commit 19dd297138
6 changed files with 57 additions and 22 deletions

View File

@@ -22,10 +22,8 @@ from src.services.provider.adapters.kiro.token_manager import generate_machine_i
def _resolve_region(cfg: KiroAuthConfig) -> str:
from src.services.provider.adapters.kiro.constants import DEFAULT_REGION
region = str(cfg.region or "").strip()
return region or DEFAULT_REGION
"""解析 API 服务端点的 regionq.{region}.amazonaws.com"""
return cfg.effective_api_region()
def _is_thinking_enabled(request_body: dict[str, Any]) -> bool:

View File

@@ -29,6 +29,13 @@ def _get_str(raw: dict[str, Any], *keys: str) -> str | None:
return None
def _nonempty(s: str | None) -> str | None:
"""Return *s* if it's a non-empty stripped string, else ``None``."""
if isinstance(s, str) and s.strip():
return s.strip()
return None
def _parse_iso_to_epoch_seconds(value: object) -> int | None:
if not isinstance(value, str) or not value.strip():
return None
@@ -54,7 +61,13 @@ class KiroAuthConfig:
expires_at: int = 0
profile_arn: str | None = None
region: str | None = None
region: str | None = None # OIDC regionIdC token 刷新用)
# 独立的 auth / api region与 kiro.rs 对齐)
# auth_region: token 刷新端点,未设置时回退到 region
# api_region: q.{region} 服务端点,未设置时回退到 DEFAULT_REGION
auth_region: str | None = None
api_region: str | None = None
client_id: str | None = None
client_secret: str | None = None
@@ -69,6 +82,26 @@ class KiroAuthConfig:
# 缓存的 access_token可选用于避免频繁刷新
access_token: str | None = None
def effective_auth_region(self) -> str:
"""Token 刷新用的 region。
优先级: auth_region > region > DEFAULT_REGION
"""
from src.services.provider.adapters.kiro.constants import DEFAULT_REGION
return _nonempty(self.auth_region) or _nonempty(self.region) or DEFAULT_REGION
def effective_api_region(self) -> str:
"""API 服务端点q.{region})用的 region。
优先级: api_region > DEFAULT_REGION
注意: 不从 region 继承,因为 region 通常是 OIDC region如 eu-north-1
而 q.{region} 端点目前仅 us-east-1 可用。
"""
from src.services.provider.adapters.kiro.constants import DEFAULT_REGION
return _nonempty(self.api_region) or DEFAULT_REGION
@staticmethod
def infer_auth_method(raw: dict[str, Any]) -> str:
"""
@@ -140,6 +173,8 @@ class KiroAuthConfig:
expires_at=int(expires_at),
profile_arn=_get_str(raw, "profile_arn", "profileArn"),
region=_get_str(raw, "region"),
auth_region=_get_str(raw, "auth_region", "authRegion"),
api_region=_get_str(raw, "api_region", "apiRegion"),
client_id=_get_str(raw, "client_id", "clientId"),
client_secret=_get_str(raw, "client_secret", "clientSecret"),
machine_id=_get_str(raw, "machine_id", "machineId"),
@@ -164,6 +199,8 @@ class KiroAuthConfig:
"expires_at": self.expires_at,
"profile_arn": self.profile_arn,
"region": self.region,
"auth_region": self.auth_region,
"api_region": self.api_region,
"client_id": self.client_id,
"client_secret": self.client_secret,
"machine_id": self.machine_id,

View File

@@ -76,13 +76,13 @@ def is_token_expired(expires_at: int | None, *, skew_seconds: int = 120) -> bool
def _resolve_region(cfg: KiroAuthConfig) -> str:
region = str(cfg.region or "").strip()
if region and _REGION_RE.fullmatch(region):
"""解析 token 刷新端点的 region。"""
region = cfg.effective_auth_region()
if _REGION_RE.fullmatch(region):
return region
# Keep best-effort fallback; actual host parsing happens in transport hook.
from src.services.provider.adapters.kiro.constants import DEFAULT_REGION
return region or DEFAULT_REGION
return DEFAULT_REGION
def _try_extract_email_from_jwt(token: str) -> str | None:

View File

@@ -80,22 +80,23 @@ async def fetch_kiro_usage_limits(
raise RuntimeError("无法获取 Kiro access_token")
# 构建请求
from src.services.provider.adapters.kiro.constants import DEFAULT_REGION
region = (updated_cfg.region if updated_cfg else cfg.region) or DEFAULT_REGION
effective_cfg = updated_cfg or cfg
region = effective_cfg.effective_api_region()
host = f"q.{region}.amazonaws.com"
machine_id = generate_machine_id(updated_cfg or cfg)
kiro_version = (updated_cfg.kiro_version if updated_cfg else cfg.kiro_version) or "0.8.0"
machine_id = generate_machine_id(effective_cfg)
kiro_version = (effective_cfg.kiro_version or "0.8.0").strip() or "0.8.0"
# 构建 URL添加 isEmailRequired=true 获取邮箱)
url = f"https://{host}/getUsageLimits?origin=AI_EDITOR&resourceType=AGENTIC_REQUEST&isEmailRequired=true"
profile_arn = updated_cfg.profile_arn if updated_cfg else cfg.profile_arn
profile_arn = effective_cfg.profile_arn
if profile_arn:
from urllib.parse import quote
url += f"&profileArn={quote(profile_arn, safe='')}"
logger.debug("[KIRO_QUOTA] 请求 URL: {}", url)
# 构建 headers
headers = {
"x-amz-user-agent": build_x_amz_user_agent_usage(
@@ -141,9 +142,7 @@ async def fetch_kiro_usage_limits(
if response.status_code == 423:
ban_reason = response_text[:200] if response_text else "HTTP 423 Locked"
else:
ban_reason = (
response_text[:200] if response_text else "HTTP 403 权限被拒绝"
)
ban_reason = response_text[:200] if response_text else "HTTP 403 权限被拒绝"
error_msg = {
401: "认证失败Token 无效或已过期",