324 lines
10 KiB
Python
324 lines
10 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
Secret storage abstraction layer.
|
|
|
|
Supported storage backends:
|
|
- none : No secret stored (e.g., browser profile login state)
|
|
- env : Secret is read from environment variable at runtime
|
|
- windows_credential : Secret stored in Windows Credential Manager (ctypes)
|
|
|
|
Raw secret values are NEVER logged or written to SQLite.
|
|
"""
|
|
import os
|
|
import re
|
|
import sys
|
|
import uuid
|
|
from util.logging_config import get_skill_logger
|
|
|
|
|
|
def mask_secret(secret: str) -> str:
|
|
"""
|
|
Return a masked version of the secret for display/storage.
|
|
Never returns the full original secret or a reversible prefix.
|
|
|
|
Examples:
|
|
- Long API key -> \"****abcd\" (last 4 chars only, after separator)
|
|
- Short / empty -> \"****\"
|
|
"""
|
|
if secret is None:
|
|
return "****"
|
|
s = str(secret).strip()
|
|
if not s:
|
|
return "****"
|
|
if len(s) <= 4:
|
|
return "****"
|
|
return "****" + s[-4:]
|
|
|
|
|
|
def make_windows_credential_target(
|
|
platform_key: str,
|
|
credential_type: str,
|
|
credential_id_or_label: str,
|
|
) -> str:
|
|
"""
|
|
Generate a Windows Credential Manager target name for a credential.
|
|
|
|
Format: openclaw/account-manager/{user_id}/{platform_key}/{credential_type}/{uuid_short}
|
|
"""
|
|
user_id = (
|
|
(os.getenv("CLAW_USER_ID") or os.getenv("JIANGCHANG_USER_ID") or "").strip()
|
|
or "_anon"
|
|
)
|
|
safe_label = re.sub(r"[^a-zA-Z0-9_-]", "_", credential_id_or_label)[:32]
|
|
uid = uuid.uuid4().hex[:8]
|
|
return (
|
|
f"openclaw/account-manager/{user_id}/"
|
|
f"{platform_key}/{credential_type}/{safe_label}_{uid}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Write
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def write_secret(storage: str, secret_ref: str, secret_value: str) -> None:
|
|
"""
|
|
Write a secret to the specified storage backend.
|
|
Raises ValueError on unsupported storage or write failure.
|
|
"""
|
|
log = get_skill_logger()
|
|
log.info(
|
|
"secret_write_attempt storage=%s ref_prefix=%s",
|
|
storage,
|
|
secret_ref[:16] if secret_ref else "none",
|
|
)
|
|
storage_n = storage.strip().lower()
|
|
|
|
if storage_n == "none":
|
|
log.info("secret_write_success storage=none ref=%s", secret_ref)
|
|
return
|
|
|
|
elif storage_n == "env":
|
|
log.info("secret_write_success storage=env ref=%s", secret_ref)
|
|
return
|
|
|
|
elif storage_n == "windows_credential":
|
|
_win_cred_write_impl(secret_ref, secret_value)
|
|
log.info(
|
|
"secret_write_success storage=windows_credential ref=%s",
|
|
secret_ref[:48],
|
|
)
|
|
return
|
|
|
|
else:
|
|
log.error("secret_write_failed unsupported storage=%s", storage)
|
|
raise ValueError(f"Unsupported secret storage type: {storage}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Read
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def read_secret(storage: str, secret_ref: str) -> str:
|
|
"""
|
|
Read/retrieve a secret from the specified storage backend.
|
|
Returns the secret value as a string.
|
|
Raises ValueError on unsupported storage, missing ref, or read failure.
|
|
"""
|
|
log = get_skill_logger()
|
|
log.info(
|
|
"secret_resolve_attempt storage=%s ref_prefix=%s",
|
|
storage,
|
|
secret_ref[:16] if secret_ref else "none",
|
|
)
|
|
storage_n = storage.strip().lower()
|
|
|
|
if storage_n == "none":
|
|
log.info("secret_resolve_success storage=none")
|
|
return ""
|
|
|
|
elif storage_n == "env":
|
|
val = os.environ.get(secret_ref)
|
|
if val is None:
|
|
log.error("secret_resolve_failed env_missing ref=%s", secret_ref)
|
|
raise ValueError(
|
|
f"Environment variable '{secret_ref}' is not set (SECRET_ENV_MISSING)"
|
|
)
|
|
log.info("secret_resolve_success storage=env ref=%s", secret_ref)
|
|
return val
|
|
|
|
elif storage_n == "windows_credential":
|
|
val = _win_cred_read_impl(secret_ref)
|
|
log.info(
|
|
"secret_resolve_success storage=windows_credential ref=%s",
|
|
secret_ref[:48],
|
|
)
|
|
return val
|
|
|
|
elif storage_n == "local_encrypted":
|
|
raise ValueError(
|
|
"local_encrypted backend cannot be read via secret_ref. "
|
|
"Use credentials_repo.read_credential_plaintext() instead."
|
|
)
|
|
|
|
else:
|
|
log.error("secret_resolve_failed unsupported storage=%s", storage)
|
|
raise ValueError(f"Unsupported secret storage type: {storage}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Delete
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def delete_secret(storage: str, secret_ref: str) -> None:
|
|
"""Delete a secret from the specified storage backend."""
|
|
log = get_skill_logger()
|
|
log.info(
|
|
"secret_delete_attempt storage=%s ref_prefix=%s",
|
|
storage,
|
|
secret_ref[:16] if secret_ref else "none",
|
|
)
|
|
storage_n = storage.strip().lower()
|
|
|
|
if storage_n == "none":
|
|
return
|
|
elif storage_n == "env":
|
|
log.info("secret_delete_env_nop ref=%s", secret_ref)
|
|
return
|
|
elif storage_n == "windows_credential":
|
|
_win_cred_delete_impl(secret_ref)
|
|
log.info(
|
|
"secret_delete_success storage=windows_credential ref=%s",
|
|
secret_ref[:48],
|
|
)
|
|
return
|
|
else:
|
|
raise ValueError(f"Unsupported secret storage type: {storage}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Windows Credential Manager implementation (ctypes)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_WINCRED_TYPE_GENERIC = 1 # CRED_TYPE_GENERIC
|
|
_WINCRED_PERSIST_LOCAL_MACHINE = 2 # CRED_PERSIST_LOCAL_MACHINE
|
|
|
|
|
|
def _win_cred_available() -> bool:
|
|
"""Check if Windows Credential Manager API is available."""
|
|
if sys.platform != "win32":
|
|
return False
|
|
try:
|
|
import ctypes # noqa: F401
|
|
return True
|
|
except ImportError:
|
|
return False
|
|
|
|
|
|
def _win_cred_unavailable() -> OSError:
|
|
return OSError(
|
|
"Windows Credential Manager is not available on this platform "
|
|
"(WINDOWS_CREDENTIAL_UNAVAILABLE)"
|
|
)
|
|
|
|
|
|
def _win_cred_write_impl(target: str, secret_value: str) -> None:
|
|
if not _win_cred_available():
|
|
raise _win_cred_unavailable()
|
|
|
|
import ctypes
|
|
from ctypes import wintypes
|
|
|
|
class CREDENTIALW(ctypes.Structure):
|
|
_fields_ = [
|
|
("Flags", wintypes.DWORD),
|
|
("Type", wintypes.DWORD),
|
|
("TargetName", wintypes.LPCWSTR),
|
|
("Comment", wintypes.LPCWSTR),
|
|
("LastWritten", wintypes.FILETIME),
|
|
("CredentialBlobSize", wintypes.DWORD),
|
|
("CredentialBlob", ctypes.POINTER(ctypes.c_byte)),
|
|
("Persist", wintypes.DWORD),
|
|
("AttributeCount", wintypes.DWORD),
|
|
("Attributes", ctypes.c_void_p),
|
|
("TargetAlias", wintypes.LPCWSTR),
|
|
("UserName", wintypes.LPCWSTR),
|
|
]
|
|
|
|
secret_bytes = secret_value.encode("utf-16-le")
|
|
blob_arr = (ctypes.c_byte * len(secret_bytes)).from_buffer_copy(secret_bytes)
|
|
|
|
cred = CREDENTIALW()
|
|
cred.Type = _WINCRED_TYPE_GENERIC
|
|
cred.TargetName = target
|
|
cred.CredentialBlobSize = len(secret_bytes)
|
|
cred.CredentialBlob = ctypes.cast(blob_arr, ctypes.POINTER(ctypes.c_byte))
|
|
cred.Persist = _WINCRED_PERSIST_LOCAL_MACHINE
|
|
cred.UserName = "account-manager"
|
|
|
|
advapi32 = ctypes.windll.advapi32
|
|
if not advapi32.CredWriteW(ctypes.byref(cred), 0):
|
|
err = ctypes.get_last_error()
|
|
raise OSError(f"CredWriteW failed with error {err} (SECRET_WRITE_FAILED)")
|
|
del blob_arr, cred
|
|
|
|
|
|
def _win_cred_read_impl(target: str) -> str:
|
|
if not _win_cred_available():
|
|
raise _win_cred_unavailable()
|
|
|
|
import ctypes
|
|
from ctypes import wintypes
|
|
|
|
class CREDENTIALW(ctypes.Structure):
|
|
_fields_ = [
|
|
("Flags", wintypes.DWORD),
|
|
("Type", wintypes.DWORD),
|
|
("TargetName", wintypes.LPCWSTR),
|
|
("Comment", wintypes.LPCWSTR),
|
|
("LastWritten", wintypes.FILETIME),
|
|
("CredentialBlobSize", wintypes.DWORD),
|
|
("CredentialBlob", ctypes.POINTER(ctypes.c_byte)),
|
|
("Persist", wintypes.DWORD),
|
|
("AttributeCount", wintypes.DWORD),
|
|
("Attributes", ctypes.c_void_p),
|
|
("TargetAlias", wintypes.LPCWSTR),
|
|
("UserName", wintypes.LPCWSTR),
|
|
]
|
|
|
|
PCREDENTIALW = ctypes.POINTER(CREDENTIALW)
|
|
|
|
advapi32 = ctypes.windll.advapi32
|
|
advapi32.CredReadW.argtypes = [
|
|
wintypes.LPCWSTR,
|
|
wintypes.DWORD,
|
|
wintypes.DWORD,
|
|
ctypes.POINTER(PCREDENTIALW),
|
|
]
|
|
advapi32.CredReadW.restype = wintypes.BOOL
|
|
advapi32.CredFree.argtypes = [ctypes.c_void_p]
|
|
advapi32.CredFree.restype = None
|
|
|
|
ppcred = PCREDENTIALW()
|
|
if not advapi32.CredReadW(target, _WINCRED_TYPE_GENERIC, 0, ctypes.byref(ppcred)):
|
|
err = ctypes.get_last_error()
|
|
if err == 1168:
|
|
raise ValueError(
|
|
f"Credential not found in Windows Credential Manager: target={target}"
|
|
)
|
|
raise OSError(f"CredReadW failed with error {err} (SECRET_READ_FAILED)")
|
|
|
|
try:
|
|
cred = ppcred.contents
|
|
blob_size = int(cred.CredentialBlobSize)
|
|
if blob_size <= 0 or not cred.CredentialBlob:
|
|
return ""
|
|
raw = ctypes.string_at(cred.CredentialBlob, blob_size)
|
|
return raw.decode("utf-16-le")
|
|
finally:
|
|
advapi32.CredFree(ppcred)
|
|
|
|
|
|
def _win_cred_delete_impl(target: str) -> None:
|
|
if not _win_cred_available():
|
|
raise _win_cred_unavailable()
|
|
|
|
import ctypes
|
|
|
|
advapi32 = ctypes.windll.advapi32
|
|
if not advapi32.CredDeleteW(target, _WINCRED_TYPE_GENERIC, 0):
|
|
err = ctypes.get_last_error()
|
|
if err != 1168:
|
|
raise OSError(f"CredDeleteW failed with error {err}")
|
|
|
|
|
|
def encrypt_secret_for_storage(secret_value: str) -> str:
|
|
"""加密 secret 用于 secret_storage='local_encrypted'。
|
|
|
|
返回密文字符串，调用方负责写到 credentials.secret_ciphertext 列。
|
|
"""
|
|
from service.crypto import encrypt_secret
|
|
|
|
return encrypt_secret(secret_value)
|
|
|