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