From 3ce713eb9ea8ae15c2343d6e070b59a24e72183d Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Thu, 21 May 2026 10:32:14 +0800 Subject: [PATCH] =?UTF-8?q?feat(secrets):=20=E6=96=B0=E5=A2=9E=E5=AE=8C?= =?UTF-8?q?=E6=95=B4=E7=9A=84=E5=AF=86=E9=92=A5=E7=AE=A1=E7=90=86=E5=B7=A5?= =?UTF-8?q?=E5=85=B7=E9=93=BE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增了包括密钥加解密、配置健康检查、原子写入、审计、解析器以及 scrubber 在内的完整 secrets 模块，实现了明文密钥检测替换、密钥引用解析和配置安全校验能力 --- .../package/yuxi/channel/secrets/__init__.py | 117 +++++ .../package/yuxi/channel/secrets/atomic.py | 137 +++++ backend/package/yuxi/channel/secrets/audit.py | 284 +++++++++++ .../package/yuxi/channel/secrets/crypto.py | 114 +++++ .../package/yuxi/channel/secrets/health.py | 104 ++++ .../package/yuxi/channel/secrets/models.py | 233 +++++++++ .../package/yuxi/channel/secrets/providers.py | 470 ++++++++++++++++++ .../package/yuxi/channel/secrets/resolver.py | 326 ++++++++++++ .../package/yuxi/channel/secrets/scrubber.py | 66 +++ 9 files changed, 1851 insertions(+) create mode 100644 backend/package/yuxi/channel/secrets/__init__.py create mode 100644 backend/package/yuxi/channel/secrets/atomic.py create mode 100644 backend/package/yuxi/channel/secrets/audit.py create mode 100644 backend/package/yuxi/channel/secrets/crypto.py create mode 100644 backend/package/yuxi/channel/secrets/health.py create mode 100644 backend/package/yuxi/channel/secrets/models.py create mode 100644 backend/package/yuxi/channel/secrets/providers.py create mode 100644 backend/package/yuxi/channel/secrets/resolver.py create mode 100644 backend/package/yuxi/channel/secrets/scrubber.py diff --git a/backend/package/yuxi/channel/secrets/__init__.py b/backend/package/yuxi/channel/secrets/__init__.py new file mode 100644 index 00000000..1cfafca3 --- /dev/null +++ b/backend/package/yuxi/channel/secrets/__init__.py @@ -0,0 +1,117 @@ +from yuxi.channel.secrets.atomic import ( + AtomicWritePlan, + AtomicWriteResult, + atomic_write_config, +) +from yuxi.channel.secrets.crypto import ( + CryptoError, + decrypt_secret, + encrypt_secret, + is_encrypted, +) +from yuxi.channel.secrets.audit import ( + SecretsAuditResult, + run_secrets_audit, + run_secrets_audit_strict, +) +from yuxi.channel.secrets.health import ( + ConfigHealth, + ConfigHealthStatus, + track_config_health, +) +from yuxi.channel.secrets.models import ( + DEFAULT_SECRET_PROVIDER_ALIAS, + SINGLE_VALUE_FILE_REF_ID, + ExecSecretRefIdValidationReason, + ExecSecretRefIdValidationResult, + SecretIssue, + SecretRef, + SecretResolveResult, + SecretsAuditFinding, + SecretsAuditReport, + SecretSource, + assert_expected_resolved_secret_value, + format_exec_secret_ref_id_validation_message, + has_configured_plaintext_secret_value, + is_expected_resolved_secret_value, + is_non_empty_string, + is_record, + is_secret_field_name, + is_secret_ref, + is_valid_exec_secret_ref_id, + is_valid_provider_alias, + parse_secret_ref, + secret_ref_key, + validate_exec_secret_ref_id, + validate_ref_id, +) +from yuxi.channel.secrets.providers import ( + EnvProvider, + ExecProvider, + FileProvider, + SecretsProvider, +) +from yuxi.channel.secrets.resolver import ( + SecretProviderResolutionError, + SecretRefResolutionError, + SecretResolver, + create_exec_provider, + create_resolver_with_exec, + resolve_secrets, +) +from yuxi.channel.secrets.scrubber import ( + scrub_env, + scrub_plaintext_secrets, +) + +__all__ = [ + "CryptoError", + "decrypt_secret", + "encrypt_secret", + "is_encrypted", + "DEFAULT_SECRET_PROVIDER_ALIAS", + "SINGLE_VALUE_FILE_REF_ID", + "ExecSecretRefIdValidationReason", + "ExecSecretRefIdValidationResult", + "SecretRef", + "SecretSource", + "SecretResolveResult", + "SecretIssue", + "SecretsAuditFinding", + "SecretsAuditReport", + "assert_expected_resolved_secret_value", + "format_exec_secret_ref_id_validation_message", + "has_configured_plaintext_secret_value", + "is_expected_resolved_secret_value", + "is_non_empty_string", + "is_record", + "is_secret_field_name", + "is_secret_ref", + "is_valid_exec_secret_ref_id", + "is_valid_provider_alias", + "parse_secret_ref", + "secret_ref_key", + "validate_exec_secret_ref_id", + "validate_ref_id", + "SecretsProvider", + "EnvProvider", + "FileProvider", + "ExecProvider", + "SecretResolver", + "SecretProviderResolutionError", + "SecretRefResolutionError", + "resolve_secrets", + "create_exec_provider", + "create_resolver_with_exec", + "AtomicWritePlan", + "AtomicWriteResult", + "atomic_write_config", + "ConfigHealth", + "ConfigHealthStatus", + "track_config_health", + "scrub_env", + "scrub_plaintext_secrets", + "run_secrets_audit", + "run_secrets_audit_strict", + "SecretsAuditResult", +] diff --git a/backend/package/yuxi/channel/secrets/atomic.py b/backend/package/yuxi/channel/secrets/atomic.py new file mode 100644 index 00000000..079382c8 --- /dev/null +++ b/backend/package/yuxi/channel/secrets/atomic.py @@ -0,0 +1,137 @@ +from __future__ import annotations + +import hashlib +import json +import logging +from collections.abc import Callable +from dataclasses import dataclass, field +from enum import StrEnum +from typing import Any + +logger = logging.getLogger(__name__) + + +class AtomicWritePhase(StrEnum): + PLAN = "plan" + PROJECT = "project" + VALIDATE = "validate" + APPLY = "apply" + + +@dataclass +class ConfigFileSnapshot: + path: str + raw: str + config_hash: str + + @classmethod + def capture(cls, path: str, raw: str) -> ConfigFileSnapshot: + config_hash = hashlib.sha256(raw.encode()).hexdigest()[:12] + return cls(path=path, raw=raw, config_hash=config_hash) + + +@dataclass +class AtomicWritePlan: + phase: AtomicWritePhase = AtomicWritePhase.PLAN + target_state: dict[str, Any] = field(default_factory=dict) + changes: list[dict[str, Any]] = field(default_factory=list) + issues: list[str] = field(default_factory=list) + warnings: list[str] = field(default_factory=list) + snapshots: dict[str, ConfigFileSnapshot] = field(default_factory=dict) + dry_run: bool = True + + @property + def ok(self) -> bool: + return len(self.issues) == 0 + + +@dataclass +class AtomicWriteResult: + success: bool + phase: AtomicWritePhase + issues: list[str] = field(default_factory=list) + warnings: list[str] = field(default_factory=list) + applied_changes: int = 0 + + @classmethod + def ok(cls, phase: AtomicWritePhase, applied: int = 0, warnings: list[str] | None = None): + return cls(success=True, phase=phase, applied_changes=applied, warnings=warnings or []) + + @classmethod + def failed(cls, phase: AtomicWritePhase, issues: list[str], warnings: list[str] | None = None): + return cls(success=False, phase=phase, issues=issues, warnings=warnings or []) + + +async def atomic_write_config( + plan: AtomicWritePlan, + apply_fn: Callable[[dict[str, Any]], Any], + validate_fn: Callable[[dict[str, Any]], list[str]] | None = None, + capture_fn: Callable[[], dict[str, ConfigFileSnapshot]] | None = None, + restore_fn: Callable[[dict[str, ConfigFileSnapshot]], Any] | None = None, +) -> AtomicWriteResult: + """Plan → Project → Validate → Apply with snapshot/rollback.""" + + # Phase 1: Plan — generate target state + plan.phase = AtomicWritePhase.PLAN + if not plan.target_state: + return AtomicWriteResult.failed(AtomicWritePhase.PLAN, ["No target state provided"]) + + # Phase 2: Project — simulate changes, capture snapshots + plan.phase = AtomicWritePhase.PROJECT + if capture_fn: + plan.snapshots = capture_fn() + + # Phase 3: Validate — pre-check target state + plan.phase = AtomicWritePhase.VALIDATE + if validate_fn: + validation_issues = validate_fn(plan.target_state) + if validation_issues: + plan.issues.extend(validation_issues) + return AtomicWriteResult.failed( + AtomicWritePhase.VALIDATE, + validation_issues, + plan.warnings, + ) + + if plan.dry_run: + logger.info( + "AtomicWrite: dry-run complete, planned %d changes, %d issues, %d warnings", + len(plan.changes), + len(plan.issues), + len(plan.warnings), + ) + return AtomicWriteResult.ok(AtomicWritePhase.VALIDATE, warnings=plan.warnings) + + # Phase 4: Apply — execute with rollback on failure + plan.phase = AtomicWritePhase.APPLY + try: + await _maybe_await(apply_fn, plan.target_state) + applied = len(plan.changes) + logger.info("AtomicWrite: applied %d changes successfully", applied) + return AtomicWriteResult.ok(AtomicWritePhase.APPLY, applied=applied, warnings=plan.warnings) + except Exception as e: + logger.exception("AtomicWrite: apply failed, rolling back") + apply_issues: list[str] = [f"Apply failed: {e}"] + if restore_fn and plan.snapshots: + try: + await _maybe_await(restore_fn, plan.snapshots) + logger.info("AtomicWrite: rollback successful") + except Exception as rollback_err: + logger.exception("AtomicWrite: rollback failed: %s", rollback_err) + apply_issues.insert(0, f"Rollback failed: {rollback_err}") + return AtomicWriteResult.failed(AtomicWritePhase.APPLY, apply_issues, plan.warnings) + + +async def _maybe_await(fn: Callable, *args, **kwargs) -> Any: + import inspect + + result = fn(*args, **kwargs) + if inspect.isawaitable(result): + return await result + return result + + +def capture_db_snapshot(config: dict[str, Any]) -> ConfigFileSnapshot: + raw = json.dumps(config, sort_keys=True, default=str, ensure_ascii=False) + config_hash = hashlib.sha256(raw.encode()).hexdigest()[:12] + return ConfigFileSnapshot(path="db:channel_config", raw=raw, config_hash=config_hash) diff --git a/backend/package/yuxi/channel/secrets/audit.py b/backend/package/yuxi/channel/secrets/audit.py new file mode 100644 index 00000000..73e6aa00 --- /dev/null +++ b/backend/package/yuxi/channel/secrets/audit.py @@ -0,0 +1,284 @@ +from __future__ import annotations + +import logging +import re +from pathlib import Path +from typing import Any + +from yuxi.channel.secrets.models import ( + SecretRef, + SecretsAuditFinding, + SecretsAuditReport, + SecretSource, + is_expected_resolved_secret_value, + is_non_empty_string, + is_secret_field_name, + is_secret_ref, + parse_secret_ref, + secret_ref_key, +) +from yuxi.channel.secrets.providers import ( + SecretsProvider, +) +from yuxi.channel.secrets.resolver import SecretResolver + +logger = logging.getLogger(__name__) + + +class SecretsAuditResult: + def __init__( + self, + report: SecretsAuditReport, + exit_code: int = 0, + ): + self.report = report + self.exit_code = exit_code + + +def _resolve_json_path(path: str, key: str) -> str: + if not path: + return key + return f"{path}.{key}" + + +def _find_plaintext_secrets( + config: dict, + path: str = "", + file_label: str = "config", +) -> list[SecretsAuditFinding]: + findings: list[SecretsAuditFinding] = [] + + def _walk(current: Any, current_path: str): + if not isinstance(current, dict): + return + + for key, value in current.items(): + entry_path = _resolve_json_path(current_path, key) + + if is_secret_ref(value): + continue + + if isinstance(value, dict): + _walk(value, entry_path) + elif isinstance(value, list): + for i, item in enumerate(value): + if isinstance(item, dict) and not is_secret_ref(item): + _walk(item, f"{entry_path}[{i}]") + elif is_secret_field_name(key) and is_non_empty_string(value) and value != "***": + findings.append( + SecretsAuditFinding( + code="PLAINTEXT_FOUND", + severity="warn", + file=file_label, + json_path=entry_path, + message=f"{entry_path} is stored as plaintext.", + ) + ) + + _walk(config, path) + return findings + + +async def _check_ref_resolvability( + findings: list[SecretsAuditFinding], + config: dict, + providers: dict[SecretSource, SecretsProvider] | None = None, + allow_exec: bool = True, +) -> dict[str, Any]: + resolver = SecretResolver(providers) + refs_collected: list[tuple[SecretRef, str, str]] = [] + + def _collect_refs(current: Any, current_path: str, file_label: str): + if not isinstance(current, dict): + return + + for key, value in current.items(): + entry_path = _resolve_json_path(current_path, key) + + if is_secret_ref(value): + ref = parse_secret_ref(value) + if ref: + refs_collected.append((ref, entry_path, file_label)) + elif isinstance(value, dict): + _collect_refs(value, entry_path, file_label) + elif isinstance(value, list): + for i, item in enumerate(value): + if isinstance(item, dict): + _collect_refs(item, f"{entry_path}[{i}]", file_label) + + _collect_refs(config, "", "config") + + if not allow_exec: + exec_refs = [r for r, _, _ in refs_collected if r.source == SecretSource.EXEC] + skipped = len(exec_refs) + refs_collected = [(r, p, f) for r, p, f in refs_collected if r.source != SecretSource.EXEC] + else: + skipped = 0 + + refs = [r for r, _, _ in refs_collected] + if not refs: + return {"refs_checked": 0, "skipped_exec_refs": skipped, "resolvability_complete": True} + + try: + resolved = await resolver.resolve_secret_ref_values(refs) + except Exception as e: + for ref, entry_path, file_label in refs_collected: + findings.append( + SecretsAuditFinding( + code="REF_UNRESOLVED", + severity="error", + file=file_label, + json_path=entry_path, + message=f"Failed to resolve {secret_ref_key(ref)} ({e}).", + ) + ) + return { + "refs_checked": len(refs), + "skipped_exec_refs": skipped, + "resolvability_complete": skipped == 0, + } + + for ref, entry_path, file_label in refs_collected: + key = secret_ref_key(ref) + if key not in resolved or resolved[key] is None: + findings.append( + SecretsAuditFinding( + code="REF_UNRESOLVED", + severity="error", + file=file_label, + json_path=entry_path, + message=f"Failed to resolve {key} (resolved value is missing).", + ) + ) + elif not is_expected_resolved_secret_value(resolved[key], "string"): + findings.append( + SecretsAuditFinding( + code="REF_UNRESOLVED", + severity="error", + file=file_label, + json_path=entry_path, + message=f"Failed to resolve {key} (not a non-empty string).", + ) + ) + + return { + "refs_checked": len(refs), + "skipped_exec_refs": skipped, + "resolvability_complete": skipped == 0, + } + + +def _unquote_value(value: str) -> str: + """Remove matching outer quotes and handle escaped quotes.""" + value = value.strip() + if len(value) >= 2: + if value[0] == '"' and value[-1] == '"': + return value[1:-1].replace('\\"', '"') + if value[0] == "'" and value[-1] == "'": + return value[1:-1].replace("\\'", "'") + return value + + +def _scan_env_file(env_path: str | Path, known_env_vars: set[str]) -> list[SecretsAuditFinding]: + findings: list[SecretsAuditFinding] = [] + env_path = Path(env_path) + if not env_path.exists(): + return findings + + try: + content = env_path.read_text(encoding="utf-8") + except OSError: + return findings + + pattern = re.compile(r"^\s*(?:export\s+)?([A-Za-z_][A-Za-z0-9_]*)\s*=\s*(.+)$") + for line in content.splitlines(): + stripped = line.strip() + if not stripped or stripped.startswith("#"): + continue + match = pattern.match(line) + if not match: + continue + key = match.group(1) + if key not in known_env_vars: + continue + value = _unquote_value(match.group(2)) + if value: + findings.append( + SecretsAuditFinding( + code="PLAINTEXT_FOUND", + severity="warn", + file=str(env_path), + json_path=f"$env.{key}", + message=f"Potential secret found in .env ({key}).", + ) + ) + return findings + + +async def run_secrets_audit( + config: dict[str, Any], + providers: dict[SecretSource, SecretsProvider] | None = None, + env_path: str | Path | None = None, + known_env_vars: set[str] | None = None, + allow_exec: bool = True, +) -> SecretsAuditResult: + files_scanned: list[str] = ["config"] + + findings = _find_plaintext_secrets(config) + + resolution = await _check_ref_resolvability( + findings, + config, + providers=providers, + allow_exec=allow_exec, + ) + + if env_path and known_env_vars: + env_findings = _scan_env_file(env_path, known_env_vars) + findings.extend(env_findings) + files_scanned.append(str(env_path)) + + plaintext_count = sum(1 for f in findings if f.code == "PLAINTEXT_FOUND") + unresolved_count = sum(1 for f in findings if f.code == "REF_UNRESOLVED") + + if unresolved_count > 0: + status = "unresolved" + elif findings: + status = "findings" + else: + status = "clean" + + report = SecretsAuditReport( + version=1, + status=status, + resolution=resolution, + files_scanned=files_scanned, + summary={ + "plaintextCount": plaintext_count, + "unresolvedRefCount": unresolved_count, + }, + findings=findings, + ) + + exit_code = 2 if unresolved_count > 0 else 0 + + return SecretsAuditResult(report=report, exit_code=exit_code) + + +async def run_secrets_audit_strict( + config: dict[str, Any], + providers: dict[SecretSource, SecretsProvider] | None = None, + env_path: str | Path | None = None, + known_env_vars: set[str] | None = None, + allow_exec: bool = True, +) -> SecretsAuditResult: + result = await run_secrets_audit( + config=config, + providers=providers, + env_path=env_path, + known_env_vars=known_env_vars, + allow_exec=allow_exec, + ) + if result.report.status == "findings" and result.exit_code == 0: + result.exit_code = 1 + return result diff --git a/backend/package/yuxi/channel/secrets/crypto.py b/backend/package/yuxi/channel/secrets/crypto.py new file mode 100644 index 00000000..3a734fab --- /dev/null +++ b/backend/package/yuxi/channel/secrets/crypto.py @@ -0,0 +1,114 @@ +import base64 +import hashlib +import logging +import os + +from cryptography.hazmat.primitives.ciphers.aead import AESGCM +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC + +logger = logging.getLogger(__name__) + +_CIPHER_VERSION = b"\x01" +_KEY_BYTES = 32 +_MIN_KEY_LENGTH = 32 +_PBKDF2_ITERATIONS = 480_000 + + +class CryptoError(Exception): + pass + + +def _derive_key() -> bytes: + key = os.getenv("FORCEPILOT_SECRET_ENCRYPTION_KEY", "") + if not key: + raise CryptoError("FORCEPILOT_SECRET_ENCRYPTION_KEY must be set and at least 32 characters long") + raw = key.encode("utf-8") + if len(raw) < _MIN_KEY_LENGTH: + raise CryptoError(f"FORCEPILOT_SECRET_ENCRYPTION_KEY must be at least {_MIN_KEY_LENGTH} bytes, got {len(raw)}") + + deterministic_salt = hashlib.sha256(b"forcepilot:secret:salt:" + raw).digest()[:16] + kdf = PBKDF2HMAC( + algorithm=hashes.SHA256(), + length=_KEY_BYTES, + salt=deterministic_salt, + iterations=_PBKDF2_ITERATIONS, + ) + derived = kdf.derive(raw) + return deterministic_salt + derived + + +_ENCRYPTION_KEY: bytes | None = None + + +def get_encryption_key() -> bytes: + global _ENCRYPTION_KEY + if _ENCRYPTION_KEY is None: + _ENCRYPTION_KEY = _derive_key() + return _ENCRYPTION_KEY + + +def refresh_encryption_key() -> None: + global _ENCRYPTION_KEY + _ENCRYPTION_KEY = _derive_key() + logger.info("Encryption key refreshed") + + +def encrypt_secret(plaintext: str) -> str: + if not plaintext: + return plaintext + try: + key_material = get_encryption_key() + salt = key_material[:16] + key = key_material[16:] + aesgcm = AESGCM(key) + nonce = os.urandom(12) + ciphertext = aesgcm.encrypt(nonce, plaintext.encode("utf-8"), None) + payload = _CIPHER_VERSION + salt + nonce + ciphertext + return base64.urlsafe_b64encode(payload).decode("ascii") + except CryptoError: + raise + except Exception as e: + raise CryptoError(f"Failed to encrypt secret: {e}") from e + + +def decrypt_secret(ciphertext_b64: str) -> str: + if not ciphertext_b64: + return ciphertext_b64 + try: + payload = base64.urlsafe_b64decode(ciphertext_b64) + except Exception as e: + raise CryptoError(f"Failed to base64-decode ciphertext: {e}") from e + + if len(payload) < 1 + 16 + 12 + 16: + raise CryptoError("Ciphertext too short for valid AES-GCM payload") + + version = payload[0:1] + if version != _CIPHER_VERSION: + raise CryptoError(f"Unknown cipher version: {version!r}") + + salt = payload[1:17] + nonce = payload[17:29] + encrypted = payload[29:] + try: + key_material = get_encryption_key() + stored_salt = key_material[:16] + if salt != stored_salt: + logger.warning("Salt mismatch in ciphertext: encryption key may have been rotated") + key = key_material[16:] + aesgcm = AESGCM(key) + return aesgcm.decrypt(nonce, encrypted, None).decode("utf-8") + except CryptoError: + raise + except Exception as e: + raise CryptoError(f"Failed to decrypt secret (key mismatch or tampering): {e}") from e + + +def is_encrypted(value: str) -> bool: + if not value: + return False + try: + payload = base64.urlsafe_b64decode(value) + return len(payload) > 29 and payload[0:1] == _CIPHER_VERSION + except Exception: + return False diff --git a/backend/package/yuxi/channel/secrets/health.py b/backend/package/yuxi/channel/secrets/health.py new file mode 100644 index 00000000..09d22308 --- /dev/null +++ b/backend/package/yuxi/channel/secrets/health.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +import copy +import hashlib +import json +import logging +from collections.abc import Callable +from dataclasses import dataclass, field +from datetime import UTC, datetime +from enum import StrEnum +from typing import Any + +logger = logging.getLogger(__name__) + + +class ConfigHealthStatus(StrEnum): + HEALTHY = "healthy" + SUSPECT = "suspect" + BROKEN = "broken" + + +@dataclass +class ConfigHealth: + status: ConfigHealthStatus = ConfigHealthStatus.HEALTHY + last_known_good: str | None = None + last_promoted_good: str | None = None + last_checked_at: datetime | None = None + issues: list[str] = field(default_factory=list) + revision: int = 0 + last_known_good_config: dict[str, Any] | None = None + + @staticmethod + def compute_fingerprint(config: dict[str, Any]) -> str: + raw = json.dumps(config, sort_keys=True, default=str, ensure_ascii=False) + return hashlib.sha256(raw.encode()).hexdigest()[:12] + + def promote(self, config: dict[str, Any]) -> None: + fingerprint = self.compute_fingerprint(config) + self.last_known_good = fingerprint + self.last_promoted_good = fingerprint + self.last_known_good_config = copy.deepcopy(config) + self.revision += 1 + self.last_checked_at = datetime.now(UTC) + self.status = ConfigHealthStatus.HEALTHY + self.issues.clear() + + def mark_suspect(self, reason: str) -> None: + self.status = ConfigHealthStatus.SUSPECT + self.issues.append(reason) + self.last_checked_at = datetime.now(UTC) + + def mark_broken(self, reason: str) -> None: + self.status = ConfigHealthStatus.BROKEN + self.issues.append(reason) + self.last_checked_at = datetime.now(UTC) + + def can_auto_recover(self) -> bool: + return ( + self.last_known_good is not None + and self.last_known_good_config is not None + and self.status + in ( + ConfigHealthStatus.SUSPECT, + ConfigHealthStatus.BROKEN, + ) + ) + + def to_dict(self) -> dict[str, Any]: + return { + "status": self.status.value, + "last_known_good": self.last_known_good, + "last_promoted_good": self.last_promoted_good, + "last_checked_at": self.last_checked_at.isoformat() if self.last_checked_at else None, + "issues": self.issues, + "revision": self.revision, + "has_last_known_good_config": self.last_known_good_config is not None, + } + + +def track_config_health( + health: ConfigHealth, + config: dict[str, Any], + validate_fn: Callable[[dict[str, Any]], list[str]] | None = None, +) -> ConfigHealth: + fingerprint = ConfigHealth.compute_fingerprint(config) + + if fingerprint == health.last_known_good: + return health + + health.last_checked_at = datetime.now(UTC) + + issues = validate_fn(config) if validate_fn else [] + if issues: + health.mark_suspect(f"Validation issues: {'; '.join(issues)}") + else: + if health.status in (ConfigHealthStatus.SUSPECT, ConfigHealthStatus.BROKEN): + health.promote(config) + else: + health.status = ConfigHealthStatus.HEALTHY + health.last_known_good = fingerprint + health.last_known_good_config = copy.deepcopy(config) + health.revision += 1 + + return health diff --git a/backend/package/yuxi/channel/secrets/models.py b/backend/package/yuxi/channel/secrets/models.py new file mode 100644 index 00000000..8c687cca --- /dev/null +++ b/backend/package/yuxi/channel/secrets/models.py @@ -0,0 +1,233 @@ +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from enum import StrEnum +from typing import Any + +DEFAULT_SECRET_PROVIDER_ALIAS = "default" +SINGLE_VALUE_FILE_REF_ID = "value" + +PROVIDER_ALIAS_PATTERN = re.compile(r"^[a-z][a-z0-9_-]{0,63}$") +EXEC_SECRET_REF_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,255}$") +EXEC_SECRET_REF_TRAVERSAL_SEGMENTS = frozenset({".", ".."}) + +_SECRET_FIELD_NAMES = frozenset( + { + "app_secret", + "bot_token", + "signing_secret", + "encrypt_key", + "verification_token", + "api_key", + "api_secret", + "access_token", + "refresh_token", + "webhook_token", + "key", + "secret", + "token", + "password", + } +) + + +def is_secret_field_name(key: str) -> bool: + return key.lower() in _SECRET_FIELD_NAMES + + +class SecretSource(StrEnum): + ENV = "env" + FILE = "file" + EXEC = "exec" + + +SecretExpectedResolvedValue = str + + +class ExecSecretRefIdValidationReason(StrEnum): + PATTERN = "pattern" + TRAVERSAL_SEGMENT = "traversal-segment" + + +@dataclass +class ExecSecretRefIdValidationResult: + ok: bool + reason: ExecSecretRefIdValidationReason | None = None + + @classmethod + def valid(cls) -> ExecSecretRefIdValidationResult: + return cls(ok=True) + + @classmethod + def invalid(cls, reason: ExecSecretRefIdValidationReason) -> ExecSecretRefIdValidationResult: + return cls(ok=False, reason=reason) + + +@dataclass +class SecretRef: + source: SecretSource + ref: str + provider: str | None = None + + @classmethod + def from_dict(cls, data: dict) -> SecretRef: + provider = data.get("provider") + if provider is not None and not isinstance(provider, str): + raise TypeError(f"SecretRef.provider must be a string or None, got {type(provider).__name__}") + ref = data.get("ref") + if not isinstance(ref, str): + raise TypeError(f"SecretRef.ref must be a string, got {type(ref).__name__}") + source_str = data.get("source", "env") + if not isinstance(source_str, str): + raise TypeError(f"SecretRef.source must be a string, got {type(source_str).__name__}") + return cls( + source=SecretSource(source_str), + ref=ref, + provider=provider if provider else None, + ) + + @property + def key(self) -> str: + return secret_ref_key(self) + + def to_dict(self) -> dict[str, str | None]: + d: dict[str, str | None] = { + "source": self.source.value, + "ref": self.ref, + } + if self.provider: + d["provider"] = self.provider + return d + + +def secret_ref_key(ref: SecretRef) -> str: + provider = ref.provider or DEFAULT_SECRET_PROVIDER_ALIAS + return f"{ref.source.value}:{provider}:{ref.ref}" + + +def is_valid_provider_alias(value: str) -> bool: + return bool(PROVIDER_ALIAS_PATTERN.match(value)) + + +def is_valid_exec_secret_ref_id(value: str) -> bool: + return validate_exec_secret_ref_id(value).ok + + +def validate_exec_secret_ref_id(value: str) -> ExecSecretRefIdValidationResult: + if not EXEC_SECRET_REF_ID_PATTERN.match(value): + return ExecSecretRefIdValidationResult.invalid(ExecSecretRefIdValidationReason.PATTERN) + for segment in value.split("/"): + if segment in EXEC_SECRET_REF_TRAVERSAL_SEGMENTS: + return ExecSecretRefIdValidationResult.invalid(ExecSecretRefIdValidationReason.TRAVERSAL_SEGMENT) + return ExecSecretRefIdValidationResult.valid() + + +def format_exec_secret_ref_id_validation_message() -> str: + pattern = EXEC_SECRET_REF_ID_PATTERN.pattern + return ( + f"Exec secret reference id must match {pattern} " + 'and must not include "." or ".." path segments ' + '(example: "vault/openai/api-key").' + ) + + +@dataclass +class SecretIssue: + path: str + message: str + ref: SecretRef | None = None + severity: str = "error" + + def __str__(self) -> str: + return f"[{self.severity}] {self.path}: {self.message}" + + +@dataclass +class SecretResolveResult: + resolved: dict[str, Any] + issues: list[SecretIssue] = field(default_factory=list) + warnings: list[SecretIssue] = field(default_factory=list) + + @property + def ok(self) -> bool: + return not any(i.severity == "error" for i in self.issues) + + +@dataclass +class SecretsAuditFinding: + code: str + severity: str + file: str + json_path: str + message: str + provider: str | None = None + profile_id: str | None = None + + +@dataclass +class SecretsAuditReport: + version: int = 1 + status: str = "clean" + resolution: dict[str, Any] = field(default_factory=dict) + files_scanned: list[str] = field(default_factory=list) + summary: dict[str, int] = field(default_factory=dict) + findings: list[SecretsAuditFinding] = field(default_factory=list) + + +_SECRET_MARKER = "$secret" + +_ENV_REF_PATTERN = re.compile(r"^[A-Z][A-Z0-9_]{0,127}$") +_FILE_REF_SEGMENT_PATTERN = re.compile(r"^(?:[^~]|~0|~1)*$") + + +def is_non_empty_string(value: Any) -> bool: + return isinstance(value, str) and len(value.strip()) > 0 + + +def is_record(value: Any) -> bool: + return isinstance(value, dict) + + +def is_expected_resolved_secret_value(value: Any, expected: str) -> bool: + if expected == "string": + return is_non_empty_string(value) + return is_non_empty_string(value) or is_record(value) + + +def assert_expected_resolved_secret_value(value: Any, expected: str, error_message: str) -> None: + if not is_expected_resolved_secret_value(value, expected): + raise ValueError(error_message) + + +def has_configured_plaintext_secret_value(value: Any, expected: str) -> bool: + if expected == "string": + return is_non_empty_string(value) + return is_non_empty_string(value) or (is_record(value) and len(value) > 0) + + +def is_secret_ref(value: Any) -> bool: + return isinstance(value, dict) and _SECRET_MARKER in value + + +def parse_secret_ref(value: dict) -> SecretRef | None: + try: + inner = value[_SECRET_MARKER] + return SecretRef.from_dict(inner) + except (KeyError, TypeError, ValueError): + return None + + +def validate_ref_id(source: SecretSource, ref: str) -> str | None: + if source == SecretSource.ENV: + if not _ENV_REF_PATTERN.match(ref): + return f"Invalid env ref '{ref}': must match /^[A-Z][A-Z0-9_]{{0,127}}$/" + elif source == SecretSource.FILE: + if ref == SINGLE_VALUE_FILE_REF_ID: + return None + if not ref.startswith("/"): + return f"Invalid JSON pointer ref '{ref}': must start with '/'" + segments = ref[1:].split("/") + if not all(_FILE_REF_SEGMENT_PATTERN.match(s) for s in segments): + return f"Invalid JSON pointer ref '{ref}': contains invalid escape sequence" + return None diff --git a/backend/package/yuxi/channel/secrets/providers.py b/backend/package/yuxi/channel/secrets/providers.py new file mode 100644 index 00000000..fd9342d7 --- /dev/null +++ b/backend/package/yuxi/channel/secrets/providers.py @@ -0,0 +1,470 @@ +from __future__ import annotations + +import asyncio +import json +import logging +import os +import time +from abc import ABC, abstractmethod +from pathlib import Path +from typing import Any + +from yuxi.channel.secrets.models import ( + SINGLE_VALUE_FILE_REF_ID, + SecretRef, + SecretSource, + is_record, +) + +logger = logging.getLogger(__name__) + +DEFAULT_FILE_MAX_BYTES = 1024 * 1024 +DEFAULT_FILE_TIMEOUT_MS = 5_000 +DEFAULT_EXEC_TIMEOUT_MS = 5_000 +DEFAULT_EXEC_MAX_OUTPUT_BYTES = 1024 * 1024 +DEFAULT_EXEC_MAX_BATCH_BYTES = 256 * 1024 + + +class SecretsProvider(ABC): + """Abstract base for secret resolution providers.""" + + @abstractmethod + async def resolve(self, ref: SecretRef) -> str | None: ... + + @abstractmethod + async def resolve_batch(self, refs: list[SecretRef]) -> dict[str, str | None]: ... + + @property + @abstractmethod + def source_type(self) -> SecretSource: ... + + +class EnvProvider(SecretsProvider): + def __init__(self, allowlist: set[str] | None = None): + self._allowlist = allowlist + + @property + def source_type(self) -> SecretSource: + return SecretSource.ENV + + async def resolve(self, ref: SecretRef) -> str | None: + resolved = await self.resolve_batch([ref]) + return resolved.get(ref.ref) + + async def resolve_batch(self, refs: list[SecretRef]) -> dict[str, str | None]: + result: dict[str, str | None] = {} + for ref in refs: + if self._allowlist is not None and ref.ref not in self._allowlist: + logger.warning("EnvProvider: ref '%s' not in allowlist", ref.ref) + result[ref.ref] = None + continue + value = os.getenv(ref.ref) + if value is None: + logger.warning("EnvProvider: env var '%s' not set", ref.ref) + result[ref.ref] = value + return result + + +class FileProvider(SecretsProvider): + def __init__( + self, + base_dir: str | Path | None = None, + mode: str = "json", + max_bytes: int = DEFAULT_FILE_MAX_BYTES, + timeout_ms: int = DEFAULT_FILE_TIMEOUT_MS, + allow_insecure_path: bool = False, + filename: str = "secrets.json", + ): + self._base_dir = Path(base_dir) if base_dir else Path.cwd() + self._mode = mode + self._max_bytes = max_bytes + self._timeout_ms = timeout_ms + self._allow_insecure_path = allow_insecure_path + self._filename = filename + self._payload_cache: Any = None + self._payload_cache_mtime: float | None = None + + @property + def source_type(self) -> SecretSource: + return SecretSource.FILE + + async def resolve(self, ref: SecretRef) -> str | None: + resolved = await self.resolve_batch([ref]) + return resolved.get(ref.ref) + + async def resolve_batch(self, refs: list[SecretRef]) -> dict[str, str | None]: + payload = await self._read_payload() + if payload is None: + return {ref.ref: None for ref in refs} + + if self._mode == "singleValue": + result: dict[str, str | None] = {} + payload_str = payload if isinstance(payload, str) else json.dumps(payload) + for ref in refs: + if ref.ref != SINGLE_VALUE_FILE_REF_ID: + logger.warning( + "FileProvider: singleValue mode expects ref id '%s', got '%s'", + SINGLE_VALUE_FILE_REF_ID, + ref.ref, + ) + result[ref.ref] = None + else: + result[ref.ref] = payload_str + return result + + result = {} + for ref in refs: + try: + result[ref.ref] = self._resolve_json_pointer(payload, ref.ref) + except (KeyError, TypeError) as e: + logger.warning("FileProvider: JSON pointer '%s' not found: %s", ref.ref, e) + result[ref.ref] = None + return result + + def _resolve_json_pointer(self, payload: Any, pointer: str) -> str: + if not pointer.startswith("/"): + raise KeyError(f"JSON pointer must start with '/': {pointer}") + tokens = pointer[1:].split("/") + current = payload + for token in tokens: + token = token.replace("~1", "/").replace("~0", "~") + if isinstance(current, dict): + current = current[token] + elif isinstance(current, list): + try: + idx = int(token) + except ValueError: + raise KeyError(f"Invalid array index '{token}' in JSON pointer") + current = current[idx] + else: + raise KeyError(f"Cannot index into non-container at token '{token}'") + if isinstance(current, str): + return current + return json.dumps(current) + + def _validate_path(self, file_path: Path) -> None: + if self._allow_insecure_path: + return + resolved = file_path.resolve() + base_resolved = self._base_dir.resolve() + try: + resolved.relative_to(base_resolved) + except ValueError: + raise PermissionError( + f"FileProvider: path '{resolved}' is outside allowed base directory '{base_resolved}'" + ) + + async def _read_payload(self) -> Any: + file_path = self._base_dir / self._filename + self._validate_path(file_path) + + try: + current_mtime = os.path.getmtime(file_path) + except OSError: + current_mtime = None + + if self._payload_cache_mtime is not None and current_mtime == self._payload_cache_mtime: + return self._payload_cache + + try: + content = await asyncio.wait_for( + asyncio.to_thread(file_path.read_text, encoding="utf-8"), + timeout=self._timeout_ms / 1000, + ) + content = content.strip() + except FileNotFoundError: + logger.warning("FileProvider: file not found: %s", file_path) + self._payload_cache_mtime = current_mtime + self._payload_cache = None + return None + except OSError as e: + logger.warning("FileProvider: read error for %s: %s", file_path, e) + self._payload_cache_mtime = current_mtime + self._payload_cache = None + return None + except TimeoutError: + logger.warning("FileProvider: read timeout for %s", file_path) + self._payload_cache = None + return None + + self._payload_cache_mtime = current_mtime + try: + self._payload_cache = json.loads(content) + except json.JSONDecodeError: + if self._mode == "singleValue": + self._payload_cache = content + else: + logger.warning("FileProvider: invalid JSON in %s", file_path) + self._payload_cache = None + return self._payload_cache + + def invalidate_cache(self) -> None: + self._payload_cache_mtime = None + self._payload_cache = None + + +class ExecProvider(SecretsProvider): + EXEC_PROTOCOL_VERSION = 1 + + def __init__( + self, + command: str | None = None, + args: list[str] | None = None, + trusted_dirs: set[str] | None = None, + timeout_ms: int = DEFAULT_EXEC_TIMEOUT_MS, + no_output_timeout_ms: int | None = None, + max_output_bytes: int = DEFAULT_EXEC_MAX_OUTPUT_BYTES, + max_batch_bytes: int = DEFAULT_EXEC_MAX_BATCH_BYTES, + stderr_max_bytes: int = 4096, + json_only: bool = True, + pass_env: list[str] | None = None, + extra_env: dict[str, str] | None = None, + allow_insecure_path: bool = False, + allow_symlink_command: bool = False, + ): + self._command = command + self._args = args or [] + self._trusted_dirs = trusted_dirs or set() + self._timeout_ms = timeout_ms + self._no_output_timeout_ms = no_output_timeout_ms or timeout_ms + self._max_output_bytes = max_output_bytes + self._max_batch_bytes = max_batch_bytes + self._stderr_max_bytes = stderr_max_bytes + self._json_only = json_only + self._pass_env = pass_env or [] + self._extra_env = extra_env or {} + self._allow_insecure_path = allow_insecure_path + self._allow_symlink_command = allow_symlink_command + + @property + def source_type(self) -> SecretSource: + return SecretSource.EXEC + + async def resolve(self, ref: SecretRef) -> str | None: + resolved = await self.resolve_batch([ref]) + return resolved.get(ref.ref) + + def _validate_command(self, cmd: str) -> str: + if not cmd: + raise ValueError("ExecProvider: command is empty") + + cmd_path = Path(cmd) + if not self._allow_insecure_path and not cmd_path.is_absolute(): + raise ValueError(f"ExecProvider: command must be an absolute path, got '{cmd}'") + + resolved_cmd = cmd_path.resolve() + + if not self._allow_symlink_command: + real_cmd = resolved_cmd.resolve() + if real_cmd != resolved_cmd: + raise ValueError(f"ExecProvider: command '{cmd}' is a symlink and allow_symlink_command is False") + + if self._trusted_dirs: + cmd_dir = resolved_cmd.parent + in_trusted = any( + cmd_dir == Path(td).resolve() or cmd_dir.is_relative_to(Path(td).resolve()) for td in self._trusted_dirs + ) + if not in_trusted: + raise ValueError(f"ExecProvider: command '{resolved_cmd}' is not in trusted directories") + + return str(resolved_cmd) + + async def resolve_batch(self, refs: list[SecretRef]) -> dict[str, str | None]: + ids = list(dict.fromkeys(r.ref for r in refs)) + + if not self._command: + if not refs: + return {} + raise ValueError("ExecProvider: no command configured") + + cmd = self._command + cmd = self._validate_command(cmd) + + request = { + "protocolVersion": self.EXEC_PROTOCOL_VERSION, + "provider": "exec", + "ids": ids, + } + input_data = json.dumps(request) + + if len(input_data.encode("utf-8")) > self._max_batch_bytes: + logger.warning( + "ExecProvider: request exceeds max batch bytes (%d > %d)", + len(input_data.encode("utf-8")), + self._max_batch_bytes, + ) + return {ref.ref: None for ref in refs} + + child_env: dict[str, str] = {} + for key in self._pass_env: + value = os.getenv(key) + if value is not None: + child_env[key] = value + child_env.update(self._extra_env) + + proc = None + try: + proc = await asyncio.create_subprocess_exec( + cmd, + *self._args, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + env=child_env, + ) + + if proc.stdin is None or proc.stdout is None: + logger.warning("ExecProvider: failed to open process streams") + await self._kill_process(proc) + return {ref.ref: None for ref in refs} + + proc.stdin.write(input_data.encode()) + await proc.stdin.drain() + proc.stdin.close() + await proc.stdin.wait_closed() + + stdout_bytes, stderr_bytes = await self._read_with_limits(proc) + except asyncio.CancelledError: + if proc is not None: + await self._kill_process(proc) + raise + except TimeoutError: + if proc is not None: + await self._kill_process(proc) + logger.warning("ExecProvider: command timed out: %s", cmd[:80]) + return {ref.ref: None for ref in refs} + except Exception as e: + if proc is not None: + await self._kill_process(proc) + logger.warning("ExecProvider: command failed: %s: %s", cmd[:80], e) + return {ref.ref: None for ref in refs} + + if proc.returncode != 0: + stderr_text = stderr_bytes.decode("utf-8", errors="replace")[:200] if stderr_bytes else "" + logger.warning( + "ExecProvider: command exit code %d: %s", + proc.returncode, + stderr_text, + ) + return {ref.ref: None for ref in refs} + + stdout = stdout_bytes.decode("utf-8").strip() if stdout_bytes else "" + return self._parse_exec_values(ids, stdout) + + async def _read_with_limits(self, proc) -> tuple[bytes, bytes]: + stdout_chunks: list[bytes] = [] + stderr_chunks: list[bytes] = [] + global_deadline = time.monotonic() + self._timeout_ms / 1000 + + async def _read_stream( + stream: asyncio.StreamReader | None, + chunks: list[bytes], + byte_counter: list[int], + label: str, + max_bytes: int, + ) -> None: + if stream is None: + return + while True: + remaining = global_deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError(f"ExecProvider: global timeout ({label})") + + chunk_timeout = min(remaining, self._no_output_timeout_ms / 1000) + try: + chunk = await asyncio.wait_for(stream.read(8192), timeout=chunk_timeout) + except TimeoutError: + raise TimeoutError(f"ExecProvider: no output timeout ({label})") + + if not chunk: + return + + byte_counter[0] += len(chunk) + if byte_counter[0] > max_bytes: + raise ValueError( + f"ExecProvider: {label} output exceeded max bytes ({byte_counter[0]} > {max_bytes})" + ) + + chunks.append(chunk) + + try: + stdout_counter = [0] + stderr_counter = [0] + stderr_limit = min(self._max_output_bytes, self._stderr_max_bytes) + await asyncio.gather( + _read_stream(proc.stdout, stdout_chunks, stdout_counter, "stdout", self._max_output_bytes), + _read_stream(proc.stderr, stderr_chunks, stderr_counter, "stderr", stderr_limit), + ) + except Exception: + await self._kill_process(proc) + raise + + await proc.wait() + return b"".join(stdout_chunks), b"".join(stderr_chunks) + + async def _kill_process(self, proc) -> None: + try: + proc.kill() + await asyncio.wait_for(proc.wait(), timeout=5.0) + except TimeoutError: + logger.warning("ExecProvider: process did not exit after kill") + except Exception as e: + logger.warning("ExecProvider: failed to kill process: %s", e) + + def _parse_exec_values(self, ids: list[str], stdout: str) -> dict[str, str | None]: + trimmed = stdout.strip() + if not trimmed: + logger.warning("ExecProvider: empty stdout") + return {ref_id: None for ref_id in ids} + + if not self._json_only and len(ids) == 1: + try: + parsed = json.loads(trimmed) + except json.JSONDecodeError: + return {ids[0]: trimmed} + else: + try: + parsed = json.loads(trimmed) + except json.JSONDecodeError: + logger.warning("ExecProvider: invalid JSON response") + return {ref_id: None for ref_id in ids} + + if not is_record(parsed): + if not self._json_only and len(ids) == 1 and isinstance(parsed, str): + return {ids[0]: parsed} + logger.warning("ExecProvider: response must be an object") + return {ref_id: None for ref_id in ids} + + protocol_version = parsed.get("protocolVersion") + if protocol_version != self.EXEC_PROTOCOL_VERSION: + logger.warning("ExecProvider: protocolVersion must be %d", self.EXEC_PROTOCOL_VERSION) + return {ref_id: None for ref_id in ids} + + response_values = parsed.get("values") + if not is_record(response_values): + logger.warning('ExecProvider: response missing "values"') + return {ref_id: None for ref_id in ids} + + response_errors = parsed.get("errors") if is_record(parsed.get("errors")) else None + + result: dict[str, str | None] = {} + for ref_id in ids: + if response_errors and ref_id in response_errors: + error_entry = response_errors[ref_id] + if is_record(error_entry) and isinstance(error_entry.get("message"), str): + msg = error_entry["message"].strip() + logger.warning("ExecProvider: error for id '%s': %s", ref_id, msg) + else: + logger.warning("ExecProvider: error for id '%s'", ref_id) + result[ref_id] = None + continue + + if ref_id not in response_values: + logger.warning("ExecProvider: response missing id '%s'", ref_id) + result[ref_id] = None + continue + + value = response_values[ref_id] + result[ref_id] = str(value) if not isinstance(value, str) else value + + return result diff --git a/backend/package/yuxi/channel/secrets/resolver.py b/backend/package/yuxi/channel/secrets/resolver.py new file mode 100644 index 00000000..ef44e5d3 --- /dev/null +++ b/backend/package/yuxi/channel/secrets/resolver.py @@ -0,0 +1,326 @@ +from __future__ import annotations + +import asyncio +import logging +from collections import defaultdict +from typing import Any + +from yuxi.channel.secrets.models import ( + DEFAULT_SECRET_PROVIDER_ALIAS, + SecretIssue, + SecretRef, + SecretResolveResult, + SecretSource, + is_secret_ref, + is_valid_exec_secret_ref_id, + parse_secret_ref, + secret_ref_key, + validate_ref_id, +) +from yuxi.channel.secrets.providers import ( + EnvProvider, + ExecProvider, + FileProvider, + SecretsProvider, +) + +logger = logging.getLogger(__name__) + +DEFAULT_PROVIDER_CONCURRENCY = 4 +DEFAULT_MAX_REFS_PER_PROVIDER = 512 + + +def _provider_key(source: SecretSource, alias: str | None = None) -> str: + return f"{source.value}:{alias or DEFAULT_SECRET_PROVIDER_ALIAS}" + + +def _parse_provider_key(key: str) -> tuple[SecretSource, str] | None: + parts = key.split(":", 1) + if len(parts) != 2: + return None + try: + return SecretSource(parts[0]), parts[1] + except ValueError: + return None + + +class SecretProviderResolutionError(Exception): + def __init__(self, source: SecretSource, provider: str, message: str): + self.source = source + self.provider = provider + super().__init__(message) + + +class SecretRefResolutionError(Exception): + def __init__(self, source: SecretSource, provider: str, ref_id: str, message: str): + self.source = source + self.provider = provider + self.ref_id = ref_id + super().__init__(message) + + +class SecretResolver: + def __init__( + self, + providers: dict[SecretSource, SecretsProvider] | dict[str, SecretsProvider] | None = None, + max_provider_concurrency: int = DEFAULT_PROVIDER_CONCURRENCY, + max_refs_per_provider: int = DEFAULT_MAX_REFS_PER_PROVIDER, + ): + self._providers: dict[str, SecretsProvider] = {} + if providers: + for key, provider in providers.items(): + if isinstance(key, SecretSource): + self._providers[_provider_key(key)] = provider + else: + self._providers[key] = provider + else: + self._providers = { + _provider_key(SecretSource.ENV): EnvProvider(), + _provider_key(SecretSource.FILE): FileProvider(), + _provider_key(SecretSource.EXEC): ExecProvider(), + } + self._max_provider_concurrency = max_provider_concurrency + self._max_refs_per_provider = max_refs_per_provider + + def register_provider( + self, + source: SecretSource, + provider: SecretsProvider, + alias: str | None = None, + ) -> None: + self._providers[_provider_key(source, alias)] = provider + + async def resolve(self, config: dict[str, Any], path: str = "") -> SecretResolveResult: + result = SecretResolveResult(resolved={}, issues=[], warnings=[]) + await self._walk(config, result.resolved, path, result) + return result + + async def resolve_secret_ref_values(self, refs: list[SecretRef]) -> dict[str, str | None]: + if not refs: + return {} + + unique_refs: dict[str, SecretRef] = {} + for ref in refs: + ref_id = ref.ref.strip() + if not ref_id: + raise ValueError("Secret reference id is empty.") + if ref.source == SecretSource.EXEC and not is_valid_exec_secret_ref_id(ref_id): + raise ValueError( + f"Invalid exec secret ref id '{ref_id}': must match /^[A-Za-z0-9][A-Za-z0-9._:/-]{{0,255}}$/" + ) + normalized = SecretRef(source=ref.source, ref=ref_id, provider=ref.provider) + unique_refs[secret_ref_key(normalized)] = normalized + + grouped: dict[str, list[SecretRef]] = defaultdict(list) + for ref in unique_refs.values(): + provider_alias = ref.provider or DEFAULT_SECRET_PROVIDER_ALIAS + group_key = _provider_key(ref.source, provider_alias) + grouped[group_key].append(ref) + + for group_key, group_refs in grouped.items(): + if len(group_refs) > self._max_refs_per_provider: + raise SecretProviderResolutionError( + source=group_refs[0].source, + provider=group_refs[0].provider or DEFAULT_SECRET_PROVIDER_ALIAS, + message=f"Provider exceeded maxRefsPerProvider ({self._max_refs_per_provider}).", + ) + + semaphore = asyncio.Semaphore(self._max_provider_concurrency) + + async def resolve_group(group_refs: list[SecretRef]) -> dict[str, str | None]: + async with semaphore: + first = group_refs[0] + provider = self._providers.get(_provider_key(first.source, first.provider)) + if provider is None: + raise SecretProviderResolutionError( + source=first.source, + provider=first.provider or DEFAULT_SECRET_PROVIDER_ALIAS, + message=( + f"No provider registered for source '{first.source.value}'" + f" with alias '{first.provider or DEFAULT_SECRET_PROVIDER_ALIAS}'" + ), + ) + batch_result = await provider.resolve_batch(group_refs) + return {secret_ref_key(ref): batch_result.get(ref.ref) for ref in group_refs} + + tasks = [resolve_group(g) for g in grouped.values()] + results = await asyncio.gather(*tasks, return_exceptions=True) + + resolved: dict[str, str | None] = {} + errors: list[Exception] = [] + for r in results: + if isinstance(r, Exception): + errors.append(r) + else: + resolved.update(r) + + if errors: + if len(errors) == 1: + raise errors[0] + raise ExceptionGroup("Secret resolution failed for one or more providers", errors) + return resolved + + async def resolve_secret_ref_value(self, ref: SecretRef) -> str | None: + ref_id = ref.ref.strip() + if not ref_id: + raise ValueError("Secret reference id is empty.") + normalized = SecretRef(source=ref.source, ref=ref_id, provider=ref.provider) + resolved = await self.resolve_secret_ref_values([normalized]) + return resolved.get(secret_ref_key(normalized)) + + async def _walk( + self, + src: Any, + dst: dict, + path: str, + result: SecretResolveResult, + ) -> None: + if not isinstance(src, dict): + return + + for key, value in src.items(): + current_path = f"{path}.{key}" if path else key + + if is_secret_ref(value): + resolved = await self._resolve_ref(value, current_path, result) + dst[key] = resolved if resolved is not None else "" + elif isinstance(value, dict): + dst[key] = {} + await self._walk(value, dst[key], current_path, result) + elif isinstance(value, list): + dst[key] = await self._walk_list(value, current_path, result) + else: + dst[key] = value + + async def _walk_list( + self, + src: list, + path: str, + result: SecretResolveResult, + ) -> list: + resolved_list = [] + for i, item in enumerate(src): + item_path = f"{path}[{i}]" + if is_secret_ref(item): + resolved = await self._resolve_ref(item, item_path, result) + resolved_list.append(resolved if resolved is not None else "") + elif isinstance(item, dict): + item_resolved: dict = {} + await self._walk(item, item_resolved, item_path, result) + resolved_list.append(item_resolved) + elif isinstance(item, list): + resolved_list.append(await self._walk_list(item, item_path, result)) + else: + resolved_list.append(item) + return resolved_list + + async def _resolve_ref( + self, + value: dict, + path: str, + result: SecretResolveResult, + ) -> str | None: + ref = parse_secret_ref(value) + if ref is None: + issue = SecretIssue(path=path, message="Invalid $secret format", severity="error") + result.issues.append(issue) + return None + + validation_error = validate_ref_id(ref.source, ref.ref) + if validation_error: + result.issues.append(SecretIssue(path=path, message=validation_error, ref=ref)) + return None + + provider = self._providers.get(_provider_key(ref.source, ref.provider)) + if provider is None: + result.issues.append( + SecretIssue( + path=path, + message=( + f"No provider registered for source '{ref.source.value}'" + f" with alias '{ref.provider or DEFAULT_SECRET_PROVIDER_ALIAS}'" + ), + ref=ref, + ) + ) + return None + + try: + resolved = await provider.resolve(ref) + except asyncio.CancelledError: + raise + except Exception as e: + result.issues.append(SecretIssue(path=path, message=f"Resolution failed: {e}", ref=ref)) + return None + + if resolved is None: + result.warnings.append( + SecretIssue( + path=path, + message=f"Secret not found: {ref.source.value}:{ref.ref}", + ref=ref, + severity="warning", + ) + ) + return resolved + + +async def resolve_secrets( + config: dict[str, Any], + providers: dict[SecretSource, SecretsProvider] | None = None, +) -> SecretResolveResult: + resolver = SecretResolver(providers) + return await resolver.resolve(config) + + +def create_exec_provider( + command: str, + args: list[str] | None = None, + timeout_ms: int = 5000, + max_output_bytes: int = 1024 * 1024, + pass_env: list[str] | None = None, + extra_env: dict[str, str] | None = None, + trusted_dirs: set[str] | None = None, +) -> ExecProvider: + return ExecProvider( + command=command, + args=args or [], + trusted_dirs=trusted_dirs or set(), + timeout_ms=timeout_ms, + max_output_bytes=max_output_bytes, + pass_env=pass_env or [], + extra_env=extra_env or {}, + ) + + +def create_resolver_with_exec( + command: str, + args: list[str] | None = None, + alias: str = "default", + timeout_ms: int = 5000, + max_output_bytes: int = 1024 * 1024, + pass_env: list[str] | None = None, + extra_env: dict[str, str] | None = None, + env_allowlist: set[str] | None = None, + file_base_dir: str | None = None, + max_provider_concurrency: int = DEFAULT_PROVIDER_CONCURRENCY, + max_refs_per_provider: int = DEFAULT_MAX_REFS_PER_PROVIDER, +) -> SecretResolver: + provider = create_exec_provider( + command=command, + args=args, + timeout_ms=timeout_ms, + max_output_bytes=max_output_bytes, + pass_env=pass_env, + extra_env=extra_env, + ) + env_provider = EnvProvider(allowlist=env_allowlist) if env_allowlist else EnvProvider() + file_provider = FileProvider(base_dir=file_base_dir) if file_base_dir else FileProvider() + resolver = SecretResolver( + max_provider_concurrency=max_provider_concurrency, + max_refs_per_provider=max_refs_per_provider, + ) + resolver.register_provider(SecretSource.ENV, env_provider, alias=None) + resolver.register_provider(SecretSource.FILE, file_provider, alias=None) + resolver.register_provider(SecretSource.EXEC, provider, alias=alias) + return resolver diff --git a/backend/package/yuxi/channel/secrets/scrubber.py b/backend/package/yuxi/channel/secrets/scrubber.py new file mode 100644 index 00000000..c7c8826b --- /dev/null +++ b/backend/package/yuxi/channel/secrets/scrubber.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +import copy +import logging +import os + +from yuxi.channel.secrets.models import is_secret_field_name, is_secret_ref + +logger = logging.getLogger(__name__) + + +def scrub_env(config: dict, env_prefix_map: dict[str, str]) -> list[str]: + """Detect and report plaintext secrets that should be migrated to SecretRef. + + Returns a list of suggestions like 'config.app_secret → $secret{env:FEISHU_APP_SECRET}'. + """ + suggestions: list[str] = [] + + def _walk(prefix: str, d: dict): + for key, value in d.items(): + path = f"{prefix}.{key}" if prefix else key + if isinstance(value, dict): + if is_secret_ref(value): + continue + _walk(path, value) + elif is_secret_field_name(key) and isinstance(value, str) and value and value != "***": + env_var = env_prefix_map.get(path) + if env_var is None: + env_var = _derive_env_name(path) + logger.debug("Derived env name '%s' for path '%s' via heuristic", env_var, path) + suggestions.append(f"{path} → $secret{{env:{env_var}}}") + + _walk("", config) + return suggestions + + +def _derive_env_name(path: str) -> str: + return path.replace(".", "_").replace("-", "_").upper() + + +def scrub_plaintext_secrets(config: dict) -> dict: + """Remove plaintext secret values from a config dict that have corresponding SecretRef entries. + + Walks the config and for any secret field that has a plaintext value AND an adjacent + SecretRef entry for the same env var, replaces the plaintext with empty string. + + Returns a deep copy; the original config dict is not modified. + """ + cleaned = copy.deepcopy(config) + + def _walk(d: dict, path_prefix: str = ""): + for key, value in list(d.items()): + current_path = f"{path_prefix}.{key}" if path_prefix else key + if isinstance(value, dict): + if is_secret_ref(value): + continue + _walk(value, current_path) + elif is_secret_field_name(key): + env_key = _derive_env_name(current_path) + env_val = os.getenv(env_key) + if env_val and value == env_val: + d[key] = "" + logger.info("Scrubbed plaintext secret at '%s' (matched env %s)", current_path, env_key) + + _walk(cleaned) + return cleaned