monky-deployd/monky_deployd/bundle.py

"""The rendered bundle: parse the tar, verify the sha, and the refusal checks that run
BEFORE any secret is read (so a refused bundle never causes a lease).

Files (monky-deploy `render_files`): `docker-compose.yml`, `.env.template`,
`secrets.manifest.json`, `bundle.json` (+ `.env.example`). The sha is monky-tenancy's
`bundle_sha`: sha256 over sorted (name, "\\0", content, "\\0")."""

from __future__ import annotations

import hashlib
import io
import json
import re
import tarfile
from dataclasses import dataclass, field

COMPOSE_NAMES = ("docker-compose.yml", "docker-compose.yaml", "compose.yaml", "compose.yml")
MANIFEST = "secrets.manifest.json"
ENV_TEMPLATE = ".env.template"
BUNDLE_JSON = "bundle.json"

# ${VAR}, ${VAR:-default}, ${VAR-default}, ${VAR:?err}, ${VAR?err}; `$$` is an escape
_VAR_RE = re.compile(r"(?<!\$)\$\{([A-Za-z_][A-Za-z0-9_]*)(?::?[-?+][^}]*)?\}")
_VAR_DEFAULTED_RE = re.compile(r"(?<!\$)\$\{([A-Za-z_][A-Za-z0-9_]*):?[-+][^}]*\}")
_PRIV_RE = re.compile(r"^\s*privileged\s*:\s*(true|yes|on)\s*$", re.I | re.M)
_HOSTNET_RE = re.compile(r"^\s*network_mode\s*:\s*[\"']?host[\"']?\s*$", re.I | re.M)
_PID_HOST_RE = re.compile(r"^\s*pid\s*:\s*[\"']?host[\"']?\s*$", re.I | re.M)
_CAP_SYSADMIN_RE = re.compile(r"^\s*-\s*[\"']?(ALL|SYS_ADMIN)[\"']?\s*$", re.M)


class BundleError(Exception):
    """A malformed or mismatching bundle (refusal: exit 1, report failed)."""

    def __init__(self, code: str, detail: str):
        super().__init__(f"{code}: {detail}")
        self.code, self.detail = code, detail


@dataclass
class Bundle:
    files: dict[str, str]
    sha: str
    compose_name: str
    meta: dict = field(default_factory=dict)
    manifest: dict = field(default_factory=dict)

    @property
    def compose(self) -> str:
        return self.files[self.compose_name]

    @property
    def env_template(self) -> str:
        return self.files.get(ENV_TEMPLATE, "")

    @property
    def env_id(self) -> str | None:
        return self.meta.get("env_id")

    @property
    def tier(self) -> str | None:
        return self.meta.get("tier")

    @property
    def agent_profile(self) -> dict:
        prof = self.meta.get("agent") or {}
        return prof if isinstance(prof, dict) else {}

    def flag(self, name: str) -> bool:
        """`bundle.json.<name>` or `bundle.json.agent.<name>` (either spelling wins)."""
        return bool(self.meta.get(name) or self.agent_profile.get(name))

    @property
    def disk_need_bytes(self) -> int:
        for src in (self.agent_profile, self.meta):
            for key in ("disk_need_bytes", "need_bytes"):
                v = src.get(key)
                if isinstance(v, (int, float)) and v > 0:
                    return int(v)
            d = src.get("disk")
            if isinstance(d, dict) and isinstance(d.get("need_bytes"), (int, float)):
                return int(d["need_bytes"])
        return 0


def bundle_sha(files: dict[str, str]) -> str:
    h = hashlib.sha256()
    for name in sorted(files):
        h.update(name.encode())
        h.update(b"\0")
        h.update(files[name].encode())
        h.update(b"\0")
    return h.hexdigest()


def parse(data: bytes, *, max_bytes: int = 4 * 1024 * 1024) -> Bundle:
    if len(data) > max_bytes:
        raise BundleError("BUNDLE_TOO_LARGE", f"{len(data)} bytes")
    files: dict[str, str] = {}
    try:
        with tarfile.open(fileobj=io.BytesIO(data), mode="r:*") as tar:
            for m in tar.getmembers():
                if not m.isfile():
                    if m.isdir():
                        continue
                    raise BundleError("BUNDLE_MEMBER", f"{m.name!r} is not a regular file")
                name = m.name
                while name.startswith("./"):
                    name = name[2:]
                if not name or name.startswith("/") or ".." in name.split("/"):
                    raise BundleError("BUNDLE_MEMBER", f"unsafe member name {m.name!r}")
                fh = tar.extractfile(m)
                raw = fh.read() if fh else b""
                try:
                    files[name] = raw.decode("utf-8")
                except UnicodeDecodeError as exc:
                    raise BundleError("BUNDLE_MEMBER", f"{name} is not UTF-8 text") from exc
    except tarfile.TarError as exc:
        raise BundleError("BUNDLE_TAR", str(exc)) from exc
    compose_name = next((n for n in COMPOSE_NAMES if n in files), None)
    if compose_name is None:
        raise BundleError("BUNDLE_INCOMPLETE", f"no compose file among {list(files)}")
    meta: dict = {}
    if BUNDLE_JSON in files:
        try:
            meta = json.loads(files[BUNDLE_JSON])
        except json.JSONDecodeError as exc:
            raise BundleError("BUNDLE_JSON", str(exc)) from exc
    manifest: dict = {"vault_mode": "openbao", "entries": []}
    if MANIFEST in files:
        try:
            manifest = json.loads(files[MANIFEST])
        except json.JSONDecodeError as exc:
            raise BundleError("BUNDLE_MANIFEST", str(exc)) from exc
        if not isinstance(manifest.get("entries"), list):
            raise BundleError("BUNDLE_MANIFEST", "entries must be a list")
        for e in manifest["entries"]:
            if not isinstance(e, dict) or not e.get("var") or not e.get("path"):
                raise BundleError("BUNDLE_MANIFEST", f"bad entry {e!r}")
            if "value" in e:
                raise BundleError("BUNDLE_MANIFEST", f"entry {e['var']} carries a value (refused)")
    return Bundle(files=files, sha=bundle_sha(files), compose_name=compose_name, meta=meta, manifest=manifest)


# --- refusal checks (pure; names only, never values) -----------------------------------------


def referenced_vars(text: str) -> set[str]:
    return {m.group(1) for m in _VAR_RE.finditer(text)}


def defaulted_vars(text: str) -> set[str]:
    return {m.group(1) for m in _VAR_DEFAULTED_RE.finditer(text)}


def unresolved_vars(bundle: Bundle, provided: set[str]) -> list[str]:
    """Variables the compose file / .env.template need that neither the manifest nor a compose
    default supplies. NAMES only."""
    need = referenced_vars(bundle.compose) | referenced_vars(bundle.env_template)
    have = provided | defaulted_vars(bundle.compose)
    return sorted(need - have)


def privileged_findings(bundle: Bundle) -> list[str]:
    text = bundle.compose
    out = []
    if _PRIV_RE.search(text):
        out.append("privileged: true")
    if _HOSTNET_RE.search(text):
        out.append("network_mode: host")
    if _PID_HOST_RE.search(text):
        out.append("pid: host")
    if re.search(r"^\s*cap_add\s*:", text, re.M) and _CAP_SYSADMIN_RE.search(text):
        out.append("cap_add SYS_ADMIN/ALL")
    return out


def manifest_vars(bundle: Bundle) -> set[str]:
    return {e["var"] for e in bundle.manifest.get("entries", [])}


def render_env(template: str, values: dict[str, str]) -> str:
    """Fill `VAR=${VAR}` lines; every other line is copied. Values are quoted the compose way
    (double quotes, `\\`/`"` escaped, `$` doubled) unless plain."""
    out = []
    for line in template.splitlines():
        m = re.match(r"^([A-Za-z_][A-Za-z0-9_]*)=\$\{\1\}\s*$", line)
        if m and m.group(1) in values:
            out.append(f"{m.group(1)}={_quote(values[m.group(1)])}")
        else:
            out.append(line)
    return "\n".join(out) + "\n"


def _quote(v: str) -> str:
    if re.fullmatch(r"[A-Za-z0-9_./:@+=,%~-]*", v):
        return v
    esc = v.replace("\\", "\\\\").replace('"', '\\"').replace("$", "$$").replace("\n", "\\n")
    return f'"{esc}"'