256 lines
7.8 KiB
Python
256 lines
7.8 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Shared helpers for account service command modules."""
|
||
import json
|
||
import os
|
||
import shutil
|
||
import sys
|
||
from dataclasses import dataclass, field
|
||
from typing import Optional
|
||
|
||
from db.accounts_repo import (
|
||
insert_account,
|
||
resolve_auth_strategy_for_platform,
|
||
ensure_platform_from_caller_meta,
|
||
)
|
||
from db.auth_strategy import (
|
||
ALL_AUTH_STRATEGIES,
|
||
AUTH_STRATEGY_CLIENT_CERTIFICATE,
|
||
AUTH_STRATEGY_DEVICE_BOUND,
|
||
AUTH_STRATEGY_PASSWORD_PLUS_2FA,
|
||
AUTH_STRATEGY_PER_SESSION_MANUAL,
|
||
AUTH_STRATEGY_QR_CODE_MANUAL,
|
||
AUTH_STRATEGY_SSO_REDIRECT,
|
||
)
|
||
from db.connection import get_conn, init_db
|
||
from util.logging_config import get_skill_logger
|
||
from util.platforms import (
|
||
DEFAULT_CAPABILITIES_JSON,
|
||
PlatformResolveResult,
|
||
_platform_list_cn_for_help,
|
||
is_valid_platform_key_format,
|
||
resolve_platform_key_dynamic,
|
||
seed_platforms,
|
||
)
|
||
|
||
|
||
def _map_secret_read_error(exc: BaseException) -> tuple[str, str]:
|
||
"""Map secret read/write exceptions to ERROR:* codes (no secret text in code)."""
|
||
msg = str(exc)
|
||
if "SECRET_ENV_MISSING" in msg:
|
||
return "ERROR:SECRET_ENV_MISSING", msg
|
||
if "WINDOWS_CREDENTIAL_UNAVAILABLE" in msg:
|
||
return "ERROR:WINDOWS_CREDENTIAL_UNAVAILABLE", msg
|
||
return "ERROR:SECRET_READ_FAILED", msg
|
||
|
||
|
||
def _credential_ok_line(payload: dict) -> str:
|
||
"""Single-line JSON for credential commands (machine integration)."""
|
||
return json.dumps({"success": True, **payload}, ensure_ascii=False)
|
||
|
||
|
||
def _derive_session_persistent(auth_strategy: str) -> int:
|
||
"""Derive default session_persistent from auth_strategy when CLI omits it."""
|
||
if auth_strategy in (
|
||
AUTH_STRATEGY_QR_CODE_MANUAL,
|
||
AUTH_STRATEGY_PASSWORD_PLUS_2FA,
|
||
AUTH_STRATEGY_DEVICE_BOUND,
|
||
AUTH_STRATEGY_SSO_REDIRECT,
|
||
):
|
||
return 1
|
||
if auth_strategy in (AUTH_STRATEGY_PER_SESSION_MANUAL, AUTH_STRATEGY_CLIENT_CERTIFICATE):
|
||
return 0
|
||
return 1
|
||
|
||
|
||
def _insert_secret_holder_account(
|
||
conn,
|
||
*,
|
||
platform_key: str,
|
||
account_label: str,
|
||
environment: str,
|
||
role: str,
|
||
provider_code: str,
|
||
url: Optional[str],
|
||
now: int,
|
||
) -> int:
|
||
"""Create a minimal accounts row so credentials can join on environment/role."""
|
||
return insert_account(
|
||
conn=conn,
|
||
platform_key=platform_key,
|
||
account_label=account_label,
|
||
login_id=None,
|
||
login_id_type="api_account",
|
||
tenant_id=None,
|
||
environment=environment,
|
||
role=role,
|
||
provider_code=provider_code,
|
||
profile_dir=None,
|
||
url=url,
|
||
extra_json="{}",
|
||
now=now,
|
||
auth_strategy=resolve_auth_strategy_for_platform(conn, platform_key),
|
||
)
|
||
|
||
|
||
def _err_json(code: str, message: str) -> str:
|
||
"""Return a JSON error line."""
|
||
return json.dumps(
|
||
{"success": False, "error": {"code": code, "message": message}},
|
||
ensure_ascii=False,
|
||
)
|
||
|
||
|
||
def _ok_json(data) -> str:
|
||
"""Return a JSON success line."""
|
||
return json.dumps(
|
||
{"success": True, "data": data},
|
||
ensure_ascii=False,
|
||
)
|
||
|
||
|
||
def _say(msg: str) -> None:
|
||
"""Print to stdout."""
|
||
print(msg)
|
||
|
||
|
||
def _say_err(msg: str) -> None:
|
||
"""Print to stderr."""
|
||
print(msg, file=sys.stderr)
|
||
|
||
|
||
def _fail_human(code: str, hint_lines: list[str]) -> None:
|
||
"""Print human-readable error block."""
|
||
_say(f"ERROR:{code}")
|
||
for line in hint_lines:
|
||
_say(line)
|
||
|
||
|
||
def _remove_profile_dir(path: str) -> None:
|
||
p = (path or "").strip()
|
||
if not p or not os.path.isdir(p):
|
||
return
|
||
try:
|
||
shutil.rmtree(p)
|
||
except OSError as e:
|
||
_say_err(f"WARNING: 未能删除用户数据目录（可手工删）：{p}\n {e}")
|
||
|
||
|
||
@dataclass
|
||
class PlatformCallerMeta:
|
||
"""调用方为未注册自定义平台提供的自动注册元数据（pick-web / add-web）。"""
|
||
|
||
display_name: Optional[str] = None
|
||
url: Optional[str] = None
|
||
domain: Optional[str] = None
|
||
auth_strategy: Optional[str] = None
|
||
aliases: list[str] = field(default_factory=list)
|
||
|
||
def is_sufficient_for_auto_register(self) -> bool:
|
||
return bool((self.display_name or "").strip() and (self.url or "").strip())
|
||
|
||
|
||
def _emit_platform_resolve_error(result: PlatformResolveResult, hint: str = "") -> None:
|
||
code = result.error_code or "INVALID_PLATFORM"
|
||
if not code.startswith("ERROR:"):
|
||
code = f"ERROR:{code}"
|
||
_say(_err_json(code, result.message or "无法识别平台。"))
|
||
if hint:
|
||
_say(hint)
|
||
|
||
|
||
def _resolve_or_auto_register(
|
||
platform_input: str,
|
||
meta: Optional[PlatformCallerMeta] = None,
|
||
*,
|
||
hint: str = "",
|
||
) -> Optional[str]:
|
||
"""解析平台；未注册时可在元数据充足时自动 ensure（幂等、仅填空字段）。"""
|
||
init_db()
|
||
conn = get_conn()
|
||
try:
|
||
result = resolve_platform_key_dynamic(platform_input, conn=conn)
|
||
if result.ok:
|
||
return result.platform_key
|
||
|
||
if result.error_code in ("PLATFORM_DISABLED", "INVALID_PLATFORM_KEY", "INVALID_PLATFORM"):
|
||
_emit_platform_resolve_error(result, hint=hint)
|
||
return None
|
||
|
||
if result.error_code != "PLATFORM_NOT_REGISTERED":
|
||
_emit_platform_resolve_error(result, hint=hint)
|
||
if result.error_code not in (
|
||
"PLATFORM_NOT_REGISTERED",
|
||
"INVALID_PLATFORM_KEY",
|
||
"PLATFORM_DISABLED",
|
||
):
|
||
_say("支持：" + _platform_list_cn_for_help())
|
||
return None
|
||
|
||
raw_key = (platform_input or "").strip().lower()
|
||
if not is_valid_platform_key_format(raw_key):
|
||
_emit_platform_resolve_error(
|
||
PlatformResolveResult(
|
||
error_code="INVALID_PLATFORM_KEY",
|
||
message=result.message or f"平台 key 格式非法：「{platform_input}」。",
|
||
),
|
||
hint=hint,
|
||
)
|
||
return None
|
||
|
||
if not meta or not meta.is_sufficient_for_auto_register():
|
||
_emit_platform_resolve_error(
|
||
PlatformResolveResult(
|
||
error_code="PLATFORM_NOT_REGISTERED",
|
||
message=(
|
||
f"平台 {raw_key} 尚未注册。"
|
||
"请传入 --platform-display-name 与 --platform-url，"
|
||
"或先执行 account platform ensure。"
|
||
),
|
||
),
|
||
hint=hint,
|
||
)
|
||
return None
|
||
|
||
if meta.auth_strategy and meta.auth_strategy not in ALL_AUTH_STRATEGIES:
|
||
_say(_err_json(
|
||
"ERR_AUTH_STRATEGY_NOT_SUPPORTED",
|
||
f"不支持的 --platform-auth-strategy：{meta.auth_strategy}",
|
||
))
|
||
return None
|
||
|
||
seed_platforms(conn)
|
||
action = ensure_platform_from_caller_meta(
|
||
conn,
|
||
raw_key,
|
||
display_name=(meta.display_name or "").strip(),
|
||
domain=(meta.domain or "generic").strip() or "generic",
|
||
default_url=(meta.url or "").strip(),
|
||
aliases=meta.aliases,
|
||
default_auth_strategy=meta.auth_strategy,
|
||
capabilities_json=DEFAULT_CAPABILITIES_JSON,
|
||
)
|
||
get_skill_logger().info(
|
||
"platform_auto_registered key=%s action=%s", raw_key, action
|
||
)
|
||
return raw_key
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _resolve_or_fail(input_name: str, hint: str = "") -> Optional[str]:
|
||
"""Resolve platform key or emit structured JSON error and return None."""
|
||
init_db()
|
||
conn = get_conn()
|
||
try:
|
||
result = resolve_platform_key_dynamic(input_name, conn=conn)
|
||
finally:
|
||
conn.close()
|
||
|
||
if result.ok:
|
||
return result.platform_key
|
||
|
||
_emit_platform_resolve_error(result, hint=hint)
|
||
if result.error_code not in ("PLATFORM_NOT_REGISTERED", "INVALID_PLATFORM_KEY", "PLATFORM_DISABLED"):
|
||
_say("支持：" + _platform_list_cn_for_help())
|
||
return None
|