122 lines
4.0 KiB
Python
122 lines
4.0 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Credential pick and resolve CLI commands."""
|
|
from typing import Optional
|
|
|
|
from db.accounts_repo import find_credentials, get_credential_by_id, update_credential_last_used
|
|
from db.connection import init_db
|
|
from service._svc_common import (
|
|
_credential_ok_line,
|
|
_err_json,
|
|
_map_secret_read_error,
|
|
_resolve_or_fail,
|
|
_say,
|
|
)
|
|
from service.secret_store import read_secret
|
|
from util.logging_config import get_skill_logger
|
|
|
|
|
|
def cmd_credential_pick(
|
|
platform_input: str,
|
|
credential_type: Optional[str] = None,
|
|
environment: Optional[str] = None,
|
|
role: Optional[str] = None,
|
|
reveal: bool = False,
|
|
) -> None:
|
|
"""Pick a credential for a platform, optionally with reveal."""
|
|
log = get_skill_logger()
|
|
log.info("cli_start subcommand=credential_pick platform=%s type=%s env=%s role=%s reveal=%s",
|
|
platform_input, credential_type, environment, role, reveal)
|
|
|
|
key = _resolve_or_fail(platform_input)
|
|
if not key:
|
|
return
|
|
|
|
init_db()
|
|
creds = find_credentials(
|
|
platform_key=key,
|
|
credential_type=credential_type,
|
|
environment=environment,
|
|
role=role,
|
|
status="active",
|
|
limit=1,
|
|
)
|
|
if not creds:
|
|
_say(_err_json("ERROR:CREDENTIAL_NOT_FOUND",
|
|
"没有符合条件的可用凭据。"))
|
|
return
|
|
|
|
cred = creds[0]
|
|
update_credential_last_used(cred["id"])
|
|
|
|
result = {
|
|
"id": cred["id"],
|
|
"platform_key": cred["platform_key"],
|
|
"credential_label": cred["credential_label"],
|
|
"credential_type": cred["credential_type"],
|
|
"secret_storage": cred["secret_storage"],
|
|
"secret_ref": cred["secret_ref"],
|
|
"secret_mask": cred["secret_mask"],
|
|
"status": cred["status"],
|
|
}
|
|
|
|
if reveal:
|
|
try:
|
|
secret_value = read_secret(cred["secret_storage"], cred["secret_ref"])
|
|
result["secret"] = secret_value
|
|
log.info(
|
|
"credential_pick_reveal_success id=%s storage=%s",
|
|
cred["id"],
|
|
cred["secret_storage"],
|
|
)
|
|
except (ValueError, OSError) as e:
|
|
code, msg = _map_secret_read_error(e)
|
|
log.error("credential_pick_reveal_failure id=%s error=%s", cred["id"], msg)
|
|
_say(_err_json(code, msg))
|
|
return
|
|
else:
|
|
log.info("credential_pick_success id=%s (masked)", cred["id"])
|
|
|
|
_say(_credential_ok_line(result))
|
|
|
|
|
|
def cmd_credential_resolve(credential_id: int, reveal: bool = False) -> None:
|
|
"""Resolve a credential by id."""
|
|
log = get_skill_logger()
|
|
log.info("cli_start subcommand=credential_resolve id=%s reveal=%s", credential_id, reveal)
|
|
|
|
init_db()
|
|
cred = get_credential_by_id(credential_id)
|
|
if not cred:
|
|
_say(_err_json("ERROR:CREDENTIAL_NOT_FOUND", f"凭据 id={credential_id} 不存在。"))
|
|
return
|
|
if cred["status"] != "active":
|
|
_say(_err_json("ERROR:CREDENTIAL_DISABLED", f"凭据 id={credential_id} 状态为 {cred['status']}，不可用。"))
|
|
return
|
|
|
|
result = {
|
|
"id": cred["id"],
|
|
"account_id": cred["account_id"],
|
|
"platform_key": cred["platform_key"],
|
|
"credential_label": cred["credential_label"],
|
|
"credential_type": cred["credential_type"],
|
|
"secret_storage": cred["secret_storage"],
|
|
"secret_ref": cred["secret_ref"],
|
|
"secret_mask": cred["secret_mask"],
|
|
"status": cred["status"],
|
|
}
|
|
|
|
if reveal:
|
|
try:
|
|
secret_value = read_secret(cred["secret_storage"], cred["secret_ref"])
|
|
result["secret"] = secret_value
|
|
log.info("credential_resolve_reveal_success id=%s", credential_id)
|
|
except (ValueError, OSError) as e:
|
|
code, msg = _map_secret_read_error(e)
|
|
log.error("credential_resolve_reveal_failure id=%s error=%s", credential_id, msg)
|
|
_say(_err_json(code, msg))
|
|
return
|
|
else:
|
|
log.info("credential_resolve_success id=%s (masked)", credential_id)
|
|
|
|
_say(_credential_ok_line(result))
|