monky-deployd/monky_deployd/config.py

"""`/etc/monky-deployd/config.yaml` — parsed with a small YAML *subset* reader (stdlib only).

Supported: nested block maps by indentation, `key: value` scalars (str / int / float / bool /
null, quoted strings), `- item` lists of scalars, `#` comments. That is exactly what the
install kit and the ansible role write. Anything fancier (anchors, flow style, multi-line
scalars) is a config error, not a silent misread."""

from __future__ import annotations

import os
import re
from dataclasses import dataclass, field
from pathlib import Path

DEFAULT_CONFIG_PATH = "/etc/monky-deployd/config.yaml"
TRANSPORTS = ("sdk", "proxy", "system")
ENV_ID_RE = re.compile(r"^env-(dev|qa|stage|prod)-[0-9]{2,3}$|^(dev-env-2|prod-cedar)$")
SITES = ("cbs", "pdx")


class ConfigError(Exception):
    pass


# --- YAML subset -----------------------------------------------------------------------------


def _scalar(raw: str):
    s = raw.strip()
    if s == "" or s in ("~", "null", "Null", "NULL"):
        return None
    if len(s) >= 2 and s[0] == s[-1] and s[0] in "\"'":
        inner = s[1:-1]
        if s[0] == '"':
            inner = inner.encode().decode("unicode_escape")
        return inner
    low = s.lower()
    if low in ("true", "yes", "on"):
        return True
    if low in ("false", "no", "off"):
        return False
    if re.fullmatch(r"[+-]?[0-9]+", s):
        return int(s)
    if re.fullmatch(r"[+-]?[0-9]*\.[0-9]+", s):
        return float(s)
    return s


def _strip_comment(line: str) -> str:
    out, quote = [], None
    for i, ch in enumerate(line):
        if quote:
            if ch == quote:
                quote = None
        elif ch in "\"'":
            quote = ch
        elif ch == "#" and (i == 0 or line[i - 1] in " \t"):
            break
        out.append(ch)
    return "".join(out).rstrip()


def parse_yaml_subset(text: str) -> dict:
    lines: list[tuple[int, str]] = []
    for n, raw in enumerate(text.splitlines(), 1):
        if raw.strip().startswith("#") or not raw.strip():
            continue
        if raw.strip() in ("---", "..."):
            continue
        if "\t" in raw[: len(raw) - len(raw.lstrip())]:
            raise ConfigError(f"line {n}: tabs are not allowed for indentation")
        body = _strip_comment(raw)
        if not body.strip():
            continue
        lines.append((len(body) - len(body.lstrip(" ")), body.strip()))
    pos = 0

    def block(indent: int):
        nonlocal pos
        if pos < len(lines) and lines[pos][1].startswith("- "):
            return seq(indent)
        return mapping(indent)

    def seq(indent: int) -> list:
        nonlocal pos
        items = []
        while pos < len(lines) and lines[pos][0] == indent and lines[pos][1].startswith("- "):
            items.append(_scalar(lines[pos][1][2:]))
            pos += 1
        return items

    def mapping(indent: int) -> dict:
        nonlocal pos
        out: dict = {}
        while pos < len(lines) and lines[pos][0] == indent:
            ind, body = lines[pos]
            m = re.match(r"^([A-Za-z0-9_.\-]+)\s*:(?:\s+(.*)|$)", body)
            if not m:
                raise ConfigError(f"cannot parse: {body!r}")
            key, val = m.group(1), m.group(2)
            pos += 1
            if val is None or val == "":
                if pos < len(lines) and lines[pos][0] > ind:
                    out[key] = block(lines[pos][0])
                else:
                    out[key] = None
            else:
                if val.startswith("[") or val.startswith("{"):
                    raise ConfigError(f"flow style is not supported ({key})")
                out[key] = _scalar(val)
        if pos < len(lines) and lines[pos][0] > indent:
            raise ConfigError(f"unexpected indentation near {lines[pos][1]!r}")
        return out

    if not lines:
        return {}
    result = block(lines[0][0])
    if pos != len(lines):
        raise ConfigError(f"unexpected content near {lines[pos][1]!r}")
    if not isinstance(result, dict):
        raise ConfigError("top level must be a map")
    return result


# --- the config -------------------------------------------------------------------------------


@dataclass
class TenancyCfg:
    service: str = "monky.tenancy.deploy"
    host: str = ""  # intercept host; defaults to `service`
    port: int = 443  # the service's INTERCEPT port (openziti); the pod-side listener is 127.0.0.1:8081
    scheme: str = "http"
    proxy_addr: str = "127.0.0.1:18443"
    timeout_s: int = 30


@dataclass
class BaoCfg:
    service: str = "openbao"
    addr: str = "https://bao.cbs.tikali.net:8200"
    proxy_addr: str = "127.0.0.1:18200"
    ca_bundle: str | None = "/etc/monky-deployd/openbao-ca.pem"
    mount: str = "jwt-tenancy"  # the AUTH mount (Gate 1 v2, ADR-0028 amendment)
    role: str = "see-env"
    kv_mount: str = "monky"
    token_max_ttl_s: int = 30 * 86400
    renew_below_s: int = 12 * 3600
    release_before_s: int = 2 * 86400
    timeout_s: int = 30


@dataclass
class DiskCfg:
    factor: float = 1.5
    headroom_bytes: int = 2 * 1024**3


@dataclass
class Config:
    env_id: str
    site: str
    transport: str = "sdk"
    identity: str = ""
    tenancy: TenancyCfg = field(default_factory=TenancyCfg)
    bao: BaoCfg = field(default_factory=BaoCfg)
    disk: DiskCfg = field(default_factory=DiskCfg)
    state_dir: str = "/var/lib/monky-deployd"
    deploy_dir: str = ""
    bootstrap_path: str = "/etc/monky-deployd/bootstrap.jwt"
    interval_s: int = 60
    volumes_on_absent: str = "keep"
    laptop_mode: bool = False
    prod: bool | None = None
    healthy_timeout_s: int = 300
    compose_project: str = ""
    docker_bin: str = "docker"
    # doc 24 §4a: the registry the bundle pulls from, used when the seeded credential is a bare
    # `username:password` (a JSON credential names its own registry).
    registry_host: str = "harbor.tikali.net"
    log_level: str = "INFO"
    path: str = DEFAULT_CONFIG_PATH

    # derived
    @property
    def token_path(self) -> Path:
        return Path(self.state_dir) / "bao.token"

    @property
    def state_path(self) -> Path:
        return Path(self.state_dir) / "state.json"

    @property
    def lock_path(self) -> Path:
        return Path(self.state_dir) / "lock"

    @property
    def docker_config_dir(self) -> Path:
        """Where the agent keeps its OWN registry credentials (`DOCKER_CONFIG`). Not `$HOME`: the
        unit runs as `monky-deployd`, and a human's or root's `docker login` must not be what the
        agent depends on (doc 24 §4a)."""
        return Path(self.state_dir) / "docker"

    @property
    def is_prod(self) -> bool:
        if self.prod is not None:
            return self.prod
        return self.env_id.startswith("env-prod-") or self.env_id == "prod-cedar"

    @property
    def bao_url(self) -> tuple[str, str, int]:
        scheme, rest = self.bao.addr.split("://", 1)
        hostport = rest.split("/", 1)[0]
        if hostport.startswith("["):
            host, _, port = hostport[1:].partition("]")
            port = port.lstrip(":")
        else:
            host, _, port = hostport.partition(":")
        return scheme, host, int(port or (443 if scheme == "https" else 80))


def _apply(obj, data: dict, section: str) -> None:
    for k, v in (data or {}).items():
        key = k.replace("-", "_")
        if not hasattr(obj, key):
            raise ConfigError(f"unknown key {section}.{k}")
        setattr(obj, key, v)


def from_dict(data: dict, path: str = DEFAULT_CONFIG_PATH) -> Config:
    data = dict(data or {})
    if not data.get("env_id"):
        raise ConfigError("env_id is required")
    if not data.get("site"):
        raise ConfigError("site is required")
    cfg = Config(env_id=str(data.pop("env_id")), site=str(data.pop("site")), path=path)
    for section, obj in (("tenancy", cfg.tenancy), ("bao", cfg.bao), ("disk", cfg.disk)):
        sub = data.pop(section, None)
        if sub is None:
            continue
        if not isinstance(sub, dict):
            raise ConfigError(f"{section} must be a map")
        # tolerated aliases
        if section == "bao":
            sub = dict(sub)
            if "auth_mount" in sub:
                sub["mount"] = sub.pop("auth_mount")
            if "approle" in sub:
                raise ConfigError(
                    "bao.approle is not supported: monky-deployd logs in to the jwt-tenancy "
                    "mount with a deploy grant (ADR-0028 amendment 2026-09-05)"
                )
            if "base_url" in sub:
                sub.pop("base_url")
        if section == "tenancy":
            sub = dict(sub)
            base = sub.pop("base_url", None)
            if base:
                scheme, rest = base.split("://", 1)
                host, _, port = rest.rstrip("/").partition(":")
                sub.setdefault("scheme", scheme)
                sub.setdefault("host", host)
                if port:
                    sub.setdefault("port", int(port))
        _apply(obj, sub, section)
    _apply(cfg, data, "")
    return validate(cfg)


def validate(cfg: Config) -> Config:
    if not ENV_ID_RE.match(cfg.env_id):
        raise ConfigError(f"env_id {cfg.env_id!r} is not env-<tier>-<nn> (or a grandfathered id)")
    cfg.site = cfg.site.lower()
    if cfg.site not in SITES:
        raise ConfigError(f"site must be one of {SITES}")
    if cfg.transport not in TRANSPORTS:
        raise ConfigError(f"transport must be one of {TRANSPORTS}")
    if cfg.transport == "sdk" and not cfg.identity:
        cfg.identity = f"/opt/openziti/etc/identities/monky-host.{cfg.env_id}.json"
    if cfg.volumes_on_absent not in ("keep", "purge"):
        raise ConfigError("volumes_on_absent must be keep|purge")
    if not cfg.tenancy.host:
        cfg.tenancy.host = cfg.tenancy.service
    if not cfg.deploy_dir:
        cfg.deploy_dir = os.path.join(cfg.state_dir, cfg.env_id)
    if not cfg.compose_project:
        cfg.compose_project = f"monky-{cfg.env_id}"
    if cfg.interval_s < 10:
        raise ConfigError("interval_s must be >= 10")
    if cfg.disk.factor < 1.0:
        raise ConfigError("disk.factor must be >= 1.0")
    if cfg.bao.ca_bundle in ("", "none"):
        cfg.bao.ca_bundle = None
    cfg.log_level = str(cfg.log_level).upper()
    return cfg


def load(path: str | os.PathLike = DEFAULT_CONFIG_PATH) -> Config:
    p = Path(path)
    try:
        text = p.read_text()
    except FileNotFoundError as exc:
        raise ConfigError(f"config {p} not found") from exc
    return from_dict(parse_yaml_subset(text), str(p))