115 lines
3.3 KiB
Python
115 lines
3.3 KiB
Python
|
"""Config loading for OpsWarden."""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import os
|
||
|
|
from dataclasses import dataclass, field
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Dict, Optional
|
||
|
|
|
||
|
|
import yaml
|
||
|
|
|
||
|
|
|
||
|
|
class ConfigError(Exception):
|
||
|
|
"""Raised when config is invalid or missing."""
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class VaultConfig:
|
||
|
|
addr: str
|
||
|
|
role_map: Dict[str, str] # ActorType.value -> vault role name
|
||
|
|
token_env: str = "VAULT_TOKEN" # env var holding the Vault token
|
||
|
|
mount: str = "ssh" # Vault secrets engine mount path
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class WardenConfig:
|
||
|
|
backend: str # "local" or "vault"
|
||
|
|
ca_key: Optional[Path] = None # required for local backend
|
||
|
|
vault: Optional[VaultConfig] = None # required for vault backend
|
||
|
|
inventory_path: Path = field(
|
||
|
|
default_factory=lambda: Path.home() / ".config" / "warden" / "inventory.yaml"
|
||
|
|
)
|
||
|
|
state_dir: Path = field(
|
||
|
|
default_factory=lambda: Path.home() / ".local" / "state" / "warden"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _default_config_path() -> Path:
|
||
|
|
return Path.home() / ".config" / "warden" / "warden.yaml"
|
||
|
|
|
||
|
|
|
||
|
|
def load_config(path: Optional[Path] = None) -> WardenConfig:
|
||
|
|
"""Load and validate warden.yaml. Respects WARDEN_CONFIG env var."""
|
||
|
|
config_path = path or Path(
|
||
|
|
os.environ.get("WARDEN_CONFIG", str(_default_config_path()))
|
||
|
|
)
|
||
|
|
if not config_path.exists():
|
||
|
|
raise ConfigError(f"Config not found: {config_path}")
|
||
|
|
|
||
|
|
try:
|
||
|
|
with config_path.open() as f:
|
||
|
|
raw = yaml.safe_load(f)
|
||
|
|
except yaml.YAMLError as e:
|
||
|
|
raise ConfigError(f"Invalid YAML in {config_path}: {e}") from e
|
||
|
|
|
||
|
|
if not isinstance(raw, dict):
|
||
|
|
raise ConfigError("Config must be a YAML mapping")
|
||
|
|
|
||
|
|
backend = str(raw.get("backend", "local"))
|
||
|
|
if backend not in ("local", "vault"):
|
||
|
|
raise ConfigError(
|
||
|
|
f"backend must be 'local' or 'vault', got: {backend!r}"
|
||
|
|
)
|
||
|
|
|
||
|
|
ca_key = None
|
||
|
|
if "ca_key" in raw and raw["ca_key"]:
|
||
|
|
ca_key = Path(os.path.expanduser(str(raw["ca_key"])))
|
||
|
|
|
||
|
|
vault_cfg = None
|
||
|
|
if backend == "vault":
|
||
|
|
v = raw.get("vault") or {}
|
||
|
|
if "addr" not in v:
|
||
|
|
raise ConfigError("vault backend requires vault.addr")
|
||
|
|
role_map = v.get("role_map") or {
|
||
|
|
"adm": "adm-role",
|
||
|
|
"agt": "agt-role",
|
||
|
|
"atm": "atm-role",
|
||
|
|
}
|
||
|
|
vault_cfg = VaultConfig(
|
||
|
|
addr=str(v["addr"]),
|
||
|
|
role_map=dict(role_map),
|
||
|
|
token_env=str(v.get("token_env", "VAULT_TOKEN")),
|
||
|
|
mount=str(v.get("mount", "ssh")),
|
||
|
|
)
|
||
|
|
elif backend == "local" and ca_key is None:
|
||
|
|
raise ConfigError("local backend requires ca_key")
|
||
|
|
|
||
|
|
inventory_path = Path(
|
||
|
|
os.path.expanduser(
|
||
|
|
str(
|
||
|
|
raw.get(
|
||
|
|
"inventory_path",
|
||
|
|
str(Path.home() / ".config" / "warden" / "inventory.yaml"),
|
||
|
|
)
|
||
|
|
)
|
||
|
|
)
|
||
|
|
)
|
||
|
|
state_dir = Path(
|
||
|
|
os.path.expanduser(
|
||
|
|
str(
|
||
|
|
raw.get(
|
||
|
|
"state_dir",
|
||
|
|
str(Path.home() / ".local" / "state" / "warden"),
|
||
|
|
)
|
||
|
|
)
|
||
|
|
)
|
||
|
|
)
|
||
|
|
|
||
|
|
return WardenConfig(
|
||
|
|
backend=backend,
|
||
|
|
ca_key=ca_key,
|
||
|
|
vault=vault_cfg,
|
||
|
|
inventory_path=inventory_path,
|
||
|
|
state_dir=state_dir,
|
||
|
|
)
|