"""Catalog: the non-secret registry of secret lanes and grants. A catalog entry describes *where* a secret lives in OpenBao, *who* may consume it, *how* it is delivered, and *what* approval/verification/rotation it requires. It never contains a secret value. Loading and validation are strict: a malformed or under-specified lane is rejected rather than silently defaulted. """ from __future__ import annotations from dataclasses import dataclass, field from pathlib import Path from typing import Any import yaml from secrets_engine.errors import CatalogError from secrets_engine.redact import looks_secret VALID_KINDS = ("kv", "auth-capability") VALID_STAGES = ("build", "test", "prod") VALID_DELIVERY_MODES = ( "exec-env", "exec-file", "npm-config", "wrapped", "read-check", "approle-login", ) VALID_APPROVAL_MODELS = ("decision", "ccr", "dual-control", "bootstrap-only") VALID_MOUNT_MANAGEMENT = ("engine", "existing") VALID_DELIVERY_AUTH_MANAGEMENT = ("engine", "existing", "none") VALID_DELIVERY_AUTH_METHODS = ("approle", "none") VALID_RISK_CLASSIFICATIONS = ("standard", "high") REQUIRED_FIELDS = ( "id", "kind", "org", "repo", "stage", "mount", "path", "consumers", "delivery_modes", "approval", "verification", "rotation", "deactivation", "audit", ) @dataclass(frozen=True) class CatalogEntry: id: str kind: str # Gitea coordinates, kept explicit to avoid the overloaded word "project". # org = the Gitea organisation (e.g. "coulomb") # repo = the Gitea repository / product (e.g. "whynot-design") org: str repo: str stage: str mount: str path: str fields: list[str] consumers: list[dict[str, Any]] delivery_modes: list[str] approval: dict[str, Any] verification: dict[str, Any] rotation: dict[str, Any] deactivation: dict[str, Any] audit: dict[str, Any] mount_management: str = "engine" delivery_auth: dict[str, Any] = field(default_factory=dict) workload_delivery: list[dict[str, Any]] = field(default_factory=list) risk: dict[str, Any] = field(default_factory=dict) delivery_config: dict[str, Any] = field(default_factory=dict) auth_capability: dict[str, Any] = field(default_factory=dict) description: str = "" raw: dict[str, Any] = field(default_factory=dict) @property def owner(self) -> str: """The lane's owning repo as the explicit ``org/repo`` slug.""" return f"{self.org}/{self.repo}" @property def npm(self) -> dict[str, Any]: """npm delivery config (registry, scope, package) when present.""" return self.delivery_config.get("npm", {}) @property def kv_data_path(self) -> str: """Full KV v2 *data* path used for read/write of the value.""" return f"{self.mount}/data/{self.path}" @property def kv_logical_path(self) -> str: """KV v2 logical path (used inside ACL policy capabilities).""" return f"{self.mount}/data/{self.path}" @property def policy_name(self) -> str: if self.kind == "auth-capability": return self.auth_capability.get("policy_name") or self.id if self.delivery_auth.get("policy_name"): return str(self.delivery_auth["policy_name"]) return f"se-{self.stage}-{self.id}" @property def role_name(self) -> str: if self.kind == "auth-capability": return self.auth_capability.get("role_name") or self.id if self.delivery_auth.get("role_name"): return str(self.delivery_auth["role_name"]) return f"se-{self.stage}-{self.id}" @property def manages_mount(self) -> bool: return self.kind == "kv" and self.mount_management == "engine" @property def delivery_auth_method(self) -> str: if self.kind == "auth-capability": return "approle" return str(self.delivery_auth.get("method", "approle")) @property def delivery_auth_management(self) -> str: if self.kind == "auth-capability": return "engine" return str(self.delivery_auth.get("management", "engine")) @property def manages_delivery_auth(self) -> bool: return self.delivery_auth_management == "engine" @property def has_delivery_auth(self) -> bool: return self.delivery_auth_management != "none" @property def delivery_token_ttl(self) -> str: return str(self.delivery_auth.get("token_ttl", "30m")) @property def delivery_token_max_ttl(self) -> str: return str(self.delivery_auth.get("token_max_ttl", self.delivery_token_ttl)) @property def delivery_secret_id_ttl(self) -> str: return str(self.delivery_auth.get("secret_id_ttl", self.delivery_token_ttl)) @property def delivery_secret_id_num_uses(self) -> int: return int(self.delivery_auth.get("secret_id_num_uses", 0)) @property def delivery_token_num_uses(self) -> int: return int(self.delivery_auth.get("token_num_uses", 0)) @property def auth_allowed_paths(self) -> dict[str, list[str]]: """Allowed OpenBao paths/capabilities for an auth-capability lane.""" paths: dict[str, list[str]] = {} for item in self.auth_capability.get("allowed_paths", []): paths[item["path"]] = list(item["capabilities"]) return paths @property def auth_denied_probe_paths(self) -> list[str]: """Non-secret denial probes proving the lane did not get broader.""" return list(self.auth_capability.get("denied_probe_paths", [])) @property def token_ttl(self) -> str: return str(self.auth_capability.get("token_ttl", self.rotation.get("ttl", "30m"))) @property def token_max_ttl(self) -> str: return str(self.auth_capability.get("token_max_ttl", self.token_ttl)) @property def secret_id_ttl(self) -> str: return str(self.auth_capability.get("secret_id_ttl", self.token_ttl)) @property def secret_id_num_uses(self) -> int: return int(self.auth_capability.get("secret_id_num_uses", 0)) @property def token_num_uses(self) -> int: return int(self.auth_capability.get("token_num_uses", 0)) def approval_required(self) -> bool: return self.approval.get("model") != "bootstrap-only" def stores_kv_value(self) -> bool: return self.kind == "kv" def validate_entry(data: dict[str, Any], *, source: str = "") -> CatalogEntry: """Validate a raw mapping and return a CatalogEntry, or raise CatalogError.""" if not isinstance(data, dict): raise CatalogError(f"{source}: catalog entry must be a mapping") data = dict(data) data.setdefault("kind", "kv") if data["kind"] not in VALID_KINDS: raise CatalogError( f"{source}: kind '{data['kind']}' invalid; must be one of {VALID_KINDS}" ) if data["kind"] == "auth-capability": data.setdefault("fields", []) data.setdefault("mount_management", "existing") data.setdefault("delivery_auth", {}) data.setdefault("workload_delivery", []) else: data.setdefault("auth_capability", {}) data.setdefault("mount_management", "engine") data.setdefault( "delivery_auth", {"method": "approle", "management": "engine"} ) data.setdefault("workload_delivery", []) missing = [k for k in REQUIRED_FIELDS if k not in data or data[k] in (None, "", [], {})] if data["kind"] == "kv" and not data.get("fields"): missing.append("fields") if missing: raise CatalogError(f"{source}: missing required fields: {', '.join(missing)}") stage = data["stage"] if stage not in VALID_STAGES: raise CatalogError( f"{source}: stage '{stage}' invalid; must be one of {VALID_STAGES}" ) modes = data["delivery_modes"] if not isinstance(modes, list) or not modes: raise CatalogError(f"{source}: delivery_modes must be a non-empty list") bad_modes = [m for m in modes if m not in VALID_DELIVERY_MODES] if bad_modes: raise CatalogError( f"{source}: unknown delivery_modes {bad_modes}; allowed {VALID_DELIVERY_MODES}" ) fields = data["fields"] if not isinstance(fields, list) or not all(isinstance(f, str) for f in fields): raise CatalogError(f"{source}: fields must be a list of strings") if data["kind"] == "auth-capability" and fields: raise CatalogError(f"{source}: auth-capability lanes must not declare KV fields") consumers = data["consumers"] if not isinstance(consumers, list) or not consumers: raise CatalogError(f"{source}: consumers must be a non-empty list") for c in consumers: if not isinstance(c, dict) or "name" not in c or "auth" not in c: raise CatalogError( f"{source}: each consumer needs at least 'name' and 'auth'" ) if data["mount_management"] not in VALID_MOUNT_MANAGEMENT: raise CatalogError( f"{source}: mount_management '{data['mount_management']}' invalid; " f"must be one of {VALID_MOUNT_MANAGEMENT}" ) workload_delivery = data["workload_delivery"] if not isinstance(workload_delivery, list): raise CatalogError(f"{source}: workload_delivery must be a list") for item in workload_delivery: if ( not isinstance(item, dict) or not isinstance(item.get("mode"), str) or not item["mode"].strip() or not isinstance(item.get("owner"), str) or not item["owner"].strip() ): raise CatalogError( f"{source}: each workload_delivery item needs non-empty mode and owner" ) if data["kind"] == "kv": delivery_auth = data["delivery_auth"] if not isinstance(delivery_auth, dict): raise CatalogError(f"{source}: delivery_auth must be a mapping") auth_method = delivery_auth.get("method", "approle") auth_management = delivery_auth.get("management", "engine") if auth_method not in VALID_DELIVERY_AUTH_METHODS: raise CatalogError( f"{source}: delivery_auth.method '{auth_method}' invalid; " f"must be one of {VALID_DELIVERY_AUTH_METHODS}" ) if auth_management not in VALID_DELIVERY_AUTH_MANAGEMENT: raise CatalogError( f"{source}: delivery_auth.management '{auth_management}' invalid; " f"must be one of {VALID_DELIVERY_AUTH_MANAGEMENT}" ) if auth_management == "none" and auth_method != "none": raise CatalogError( f"{source}: delivery_auth.management none requires method none" ) if auth_management != "none" and auth_method != "approle": raise CatalogError( f"{source}: current native delivery auth must use approle" ) if auth_management == "existing" and not delivery_auth.get("role_name"): raise CatalogError( f"{source}: existing delivery auth requires delivery_auth.role_name" ) for ttl_field in ("token_ttl", "token_max_ttl", "secret_id_ttl"): if ttl_field in delivery_auth and ( not isinstance(delivery_auth[ttl_field], str) or not delivery_auth[ttl_field].strip() ): raise CatalogError( f"{source}: delivery_auth.{ttl_field} must be a non-empty string" ) for uses_field in ("secret_id_num_uses", "token_num_uses"): uses = delivery_auth.get(uses_field, 0) if isinstance(uses, bool) or not isinstance(uses, int) or uses < 0: raise CatalogError( f"{source}: delivery_auth.{uses_field} must be a non-negative integer" ) native_modes = {"exec-env", "exec-file", "npm-config", "read-check", "wrapped"} if native_modes.intersection(modes) and auth_management == "none": raise CatalogError( f"{source}: native delivery/verification modes require delivery_auth" ) approval = data["approval"] if not isinstance(approval, dict) or "model" not in approval: raise CatalogError(f"{source}: approval must include a 'model'") if approval["model"] not in VALID_APPROVAL_MODELS: raise CatalogError( f"{source}: approval.model '{approval['model']}' invalid; " f"allowed {VALID_APPROVAL_MODELS}" ) risk = data.get("risk", {}) if not isinstance(risk, dict): raise CatalogError(f"{source}: risk must be a mapping") classification = risk.get("classification", "standard") if classification not in VALID_RISK_CLASSIFICATIONS: raise CatalogError( f"{source}: risk.classification '{classification}' invalid; " f"must be one of {VALID_RISK_CLASSIFICATIONS}" ) if classification == "high": if approval["model"] == "bootstrap-only": raise CatalogError( f"{source}: high-risk lanes cannot use bootstrap-only approval" ) for lifecycle_name in ("rotation", "deactivation"): lifecycle = data[lifecycle_name] if not isinstance(lifecycle, dict) or not lifecycle.get("owner"): raise CatalogError( f"{source}: high-risk lanes require {lifecycle_name}.owner" ) # npm-config delivery must declare WHERE it publishes (registry + scope), so # the registry is catalog data, never hardcoded in the engine. if "npm-config" in modes: npm = (data.get("delivery_config") or {}).get("npm") if not isinstance(npm, dict) or not npm.get("registry") or not npm.get("scope"): raise CatalogError( f"{source}: npm-config delivery requires " "delivery_config.npm.registry and .scope" ) if not str(npm["registry"]).startswith(("http://", "https://")): raise CatalogError( f"{source}: delivery_config.npm.registry must be an http(s) URL" ) if data["kind"] == "auth-capability": cfg = data.get("auth_capability") if not isinstance(cfg, dict) or not cfg.get("allowed_paths"): raise CatalogError( f"{source}: auth-capability lanes require auth_capability.allowed_paths" ) if "approle-login" not in modes: raise CatalogError( f"{source}: auth-capability lanes require approle-login delivery mode" ) allowed = cfg["allowed_paths"] if not isinstance(allowed, list) or not allowed: raise CatalogError(f"{source}: auth_capability.allowed_paths must be a list") for item in allowed: if not isinstance(item, dict) or "path" not in item or "capabilities" not in item: raise CatalogError( f"{source}: each allowed path needs path and capabilities" ) caps = item["capabilities"] if not isinstance(caps, list) or not all(isinstance(c, str) for c in caps): raise CatalogError( f"{source}: auth_capability.allowed_paths capabilities must be strings" ) denied = cfg.get("denied_probe_paths", []) if not isinstance(denied, list) or not all(isinstance(p, str) for p in denied): raise CatalogError( f"{source}: auth_capability.denied_probe_paths must be a list of strings" ) # A path must never leak a value through a field name suggesting inline secrets. if any(looks_secret(k) and data.get(k) for k in ("value", "secret", "token", "password")): raise CatalogError(f"{source}: catalog entries must not contain secret values") # Guard against accidentally broad mount/path. path = data["path"] if "*" in path or "*" in data["mount"]: raise CatalogError(f"{source}: wildcard mount/path not allowed in catalog") known = {f.name for f in CatalogEntry.__dataclass_fields__.values()} - {"raw"} kwargs = {k: v for k, v in data.items() if k in known} return CatalogEntry(raw=data, **kwargs) def load_entry(path: Path) -> CatalogEntry: try: data = yaml.safe_load(path.read_text(encoding="utf-8")) except FileNotFoundError as e: raise CatalogError(f"catalog file not found: {path}") from e except yaml.YAMLError as e: raise CatalogError(f"{path}: YAML parse error: {e}") from e return validate_entry(data, source=str(path)) def load_catalog(catalog_dir: Path) -> dict[str, CatalogEntry]: """Load and validate every *.yaml in the catalog directory.""" catalog_dir = Path(catalog_dir) if not catalog_dir.exists(): return {} entries: dict[str, CatalogEntry] = {} for path in sorted(catalog_dir.glob("*.yaml")): entry = load_entry(path) if entry.id in entries: raise CatalogError(f"duplicate catalog id '{entry.id}' in {path}") entries[entry.id] = entry return entries def get_entry(catalog_dir: Path, catalog_id: str) -> CatalogEntry: entries = load_catalog(catalog_dir) if catalog_id not in entries: raise CatalogError( f"catalog id '{catalog_id}' not found in {catalog_dir} " f"(known: {', '.join(sorted(entries)) or 'none'})" ) return entries[catalog_id]