generated from coulomb/repo-seed
tegwick
8e9383a33a
Add policy.py client that calls flex-auth /v1/check before sign/issue when policy.enabled is true. Record policy_decision_id in signatures.log. Default off preserves existing inventory-only behavior. Document production OpenBao health probe and update config/wiki references.
137 lines
4.1 KiB
Python
137 lines
4.1 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 PolicyConfig:
|
|
enabled: bool = False
|
|
flex_auth_url: str = "http://127.0.0.1:8080"
|
|
fail_closed: bool = True
|
|
tenant: str = "tenant:platform"
|
|
subject_env: str = "WARDEN_POLICY_SUBJECT"
|
|
system: str = "ops-warden"
|
|
|
|
|
|
@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"
|
|
)
|
|
policy: PolicyConfig = field(default_factory=PolicyConfig)
|
|
|
|
|
|
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"),
|
|
)
|
|
)
|
|
)
|
|
)
|
|
|
|
policy_raw = raw.get("policy") or {}
|
|
policy_cfg = PolicyConfig(
|
|
enabled=bool(policy_raw.get("enabled", False)),
|
|
flex_auth_url=str(policy_raw.get("flex_auth_url", "http://127.0.0.1:8080")),
|
|
fail_closed=bool(policy_raw.get("fail_closed", True)),
|
|
tenant=str(policy_raw.get("tenant", "tenant:platform")),
|
|
subject_env=str(policy_raw.get("subject_env", "WARDEN_POLICY_SUBJECT")),
|
|
system=str(policy_raw.get("system", "ops-warden")),
|
|
)
|
|
|
|
return WardenConfig(
|
|
backend=backend,
|
|
ca_key=ca_key,
|
|
vault=vault_cfg,
|
|
inventory_path=inventory_path,
|
|
state_dir=state_dir,
|
|
policy=policy_cfg,
|
|
)
|