ops-mason/src/ops_mason/executor.py

"""Phase 4: execute an approved construction plan against OpenBao.

Narrow on purpose (MASON-WP-0001-T04) — implements only the operations
the first real plan (rein-openweights-openrouter-approle,
credential_type: openbao-approle-kv) needs: a read-only policy scoped to
one KV path, an AppRole bound to it, and role_id/secret_id delivery.

Never writes to the KV path itself and never touches the downstream
secret value (the OpenRouter API key). The KV path a policy references
comes into existence later, when the founder does a paste-once-provision
through ops-warden's desk (`bao kv put`) — not through this executor.
role_id/secret_id are the AppRole's own access credential, not the
downstream secret; generating and delivering them is squarely
ops-mason's job (INTENT.md: "create ... credentials, tokens"), but they
are still handled write-only — never logged, never returned in a
printable form beyond confirmation that delivery happened.
"""

from __future__ import annotations

import hashlib
import json
import os
import re
import stat
import subprocess
from contextlib import contextmanager
from dataclasses import asdict, dataclass, replace
from pathlib import Path

from ops_mason.audit import record_build
from ops_mason.plan import ConstructionPlan, PlanError


class BuildRefused(RuntimeError):
    """The plan is not approved. This is the one place a bug is a security incident."""


class BuildError(RuntimeError):
    """The plan was approved but a bao operation failed."""


@dataclass
class AppRoleKVSpec:
    policy_name: str
    kv_path: str
    approle_name: str
    # token_num_uses has no default on purpose: OpenBao's own default (0)
    # means *unlimited* uses, the opposite of "bounded" -- a plan that says
    # "bounded token_num_uses" (the agent-harness-binky-mail shape, which
    # actually uses 8) must have that number chosen explicitly, not
    # inherited silently from whatever OpenBao considers a sane default.
    token_num_uses: int
    kv_capabilities: tuple[str, ...] = ("read",)
    token_ttl: str = "15m"
    token_max_ttl: str = "30m"
    secret_id_ttl: str = "0"
    delivery_dir: Path | None = None
    bao_bin: str = "bao"
    audit_log_path: Path | None = None
    # When True the named policy already exists (and may grant more than one
    # KV path). Do not rewrite it — AppRole bind only.
    reuse_policy: bool = False
    reuse_policy_sha256: str | None = None


@dataclass
class KubernetesKVSpec:
    policy_name: str
    role_name: str
    service_account_names: tuple[str, ...]
    service_account_namespaces: tuple[str, ...]
    policy_sha256: str = ""
    token_ttl: str = "15m"
    audit_log_path: Path | None = None
    bao_bin: str = "bao"


def _run(bao_bin: str, args: list[str], input_text: str | None = None) -> str:
    try:
        proc = subprocess.run(
            [bao_bin, *args], input=input_text, capture_output=True, text=True, timeout=30,
        )
    except (OSError, UnicodeError, subprocess.TimeoutExpired):
        raise BuildError("OpenBao command unavailable or timed out; output suppressed") from None
    if proc.returncode != 0:
        raise BuildError(f"OpenBao command failed (exit {proc.returncode}); output suppressed")
    return proc.stdout


def _policy_hcl(kv_path: str, capabilities: tuple[str, ...]) -> str:
    """Policy HCL for a KV v2 path — grants on both data/ and metadata/ sub-paths.

    KV v2 routes actual secret reads/writes through `<mount>/data/<path>`;
    a policy written against the bare `<mount>/<path>` (the KV v1 shape)
    silently denies everything on a v2 mount. Caught live during
    MASON-WP-0001-T05's build — `bao token capabilities` on the bare path
    reported full access, but the actual `kv get` still 403'd, because
    OpenBao evaluates policy against the real `data/`-prefixed path, not
    the one a caller might naively check capabilities against.
    """
    if not isinstance(kv_path, str) or len(kv_path.split("/")) < 2:
        raise BuildRefused("KV path must contain a mount and a literal entry")
    for part in kv_path.split("/"):
        _name(part, "KV path segment")
    if not isinstance(capabilities, tuple) or capabilities != ("read",):
        raise BuildRefused("this KV read-lane engine permits only the read capability")
    mount, _, rest = kv_path.partition("/")
    caps = ", ".join(f'"{c}"' for c in capabilities)
    return (
        f'path "{mount}/data/{rest}" {{\n  capabilities = [{caps}]\n}}\n\n'
        f'path "{mount}/metadata/{rest}" {{\n  capabilities = [{caps}]\n}}\n'
    )


def _name(value: str, label: str) -> None:
    if not isinstance(value, str) or not re.fullmatch(r"[a-zA-Z0-9][a-zA-Z0-9_.-]*", value):
        raise BuildRefused(f"{label} must be one literal name, without wildcards or separators")


def _duration(value: str, label: str, *, allow_zero: bool = False) -> int:
    if allow_zero and value == "0":
        return 0
    if not isinstance(value, str) or not re.fullmatch(r"[1-9][0-9]*[smh]", value):
        raise BuildRefused(f"{label} must be a positive duration in s, m or h")
    return int(value[:-1]) * {"s": 1, "m": 60, "h": 3600}[value[-1]]


def _hash(value: str) -> None:
    if not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{64}", value):
        raise BuildRefused("reused policy requires an exact SHA-256 content pin")


def _validate_spec(spec: AppRoleKVSpec | KubernetesKVSpec) -> None:
    _name(spec.policy_name, "policy name")
    _duration(spec.token_ttl, "token_ttl")
    if not isinstance(spec.bao_bin, str) or not spec.bao_bin.strip():
        raise BuildRefused("bao_bin must name an executable")
    if spec.audit_log_path is not None and not isinstance(spec.audit_log_path, Path):
        raise BuildRefused("audit_log_path must be a Path")
    if isinstance(spec, AppRoleKVSpec):
        _name(spec.approle_name, "AppRole name")
        _policy_hcl(spec.kv_path, spec.kv_capabilities)
        if type(spec.token_num_uses) is not int or spec.token_num_uses < 0:
            raise BuildRefused("token_num_uses must be an explicit nonnegative integer")
        if _duration(spec.token_max_ttl, "token_max_ttl") < _duration(spec.token_ttl, "token_ttl"):
            raise BuildRefused("token_max_ttl must not be shorter than token_ttl")
        _duration(spec.secret_id_ttl, "secret_id_ttl", allow_zero=True)
        if type(spec.reuse_policy) is not bool:
            raise BuildRefused("reuse_policy must be boolean")
        if spec.reuse_policy:
            _hash(spec.reuse_policy_sha256)
        elif spec.reuse_policy_sha256 is not None:
            raise BuildRefused("policy pin is only used with reuse_policy")
        if not isinstance(spec.delivery_dir, Path) or not spec.delivery_dir.is_absolute():
            raise BuildRefused("credential delivery requires an explicit absolute directory")
        if ".." in spec.delivery_dir.parts:
            raise BuildRefused("credential delivery path must not contain parent traversal")
        if spec.audit_log_path is not None and spec.audit_log_path.absolute() in {
            spec.delivery_dir, spec.delivery_dir / "role_id", spec.delivery_dir / "secret_id"
        }:
            raise BuildRefused("audit path must not overlap credential delivery")
    else:
        _name(spec.role_name, "Kubernetes role name")
        _hash(spec.policy_sha256)
        for values in (spec.service_account_names, spec.service_account_namespaces):
            if not isinstance(values, tuple) or not values:
                raise BuildRefused("Kubernetes role requires explicit service account bindings")
            for value in values:
                if not isinstance(value, str) or not re.fullmatch(r"[a-z0-9](?:[a-z0-9-]*[a-z0-9])?", value) or len(value) > 63:
                    raise BuildRefused("service account bindings must be literal DNS labels")
            if len(set(values)) != len(values):
                raise BuildRefused("duplicate service account binding")


def build_spec_document(spec: AppRoleKVSpec | KubernetesKVSpec) -> dict:
    """Render the complete, value-free execution input for phase-3 review."""
    _validate_spec(spec)
    fields = asdict(spec)
    for key, value in fields.items():
        if isinstance(value, Path):
            fields[key] = str(value.absolute())
        elif isinstance(value, tuple):
            fields[key] = list(value)
    return {
        "engine": "openbao-approle-kv" if isinstance(spec, AppRoleKVSpec) else "openbao-kubernetes-kv",
        **fields,
    }


def build_spec_digest(document: dict) -> str:
    """Hash canonical JSON; this is an integrity marker, not a signature."""
    try:
        encoded = json.dumps(document, sort_keys=True, separators=(",", ":"), allow_nan=False).encode()
    except (TypeError, ValueError) as exc:
        raise BuildRefused("build specification is not canonical JSON") from exc
    return hashlib.sha256(encoded).hexdigest()


def _approved_build(plan: ConstructionPlan, spec) -> ConstructionPlan:
    # The file is authoritative, so revoking approval after load takes effect.
    try:
        current = ConstructionPlan.load(plan.path)
    except (OSError, PlanError) as exc:
        raise BuildRefused("cannot reload construction-plan approval") from exc
    if current.id != plan.id or not current.is_approved():
        raise BuildRefused("construction plan is not approved; refusing to build")
    document = build_spec_document(spec)
    if current.credential_type != document["engine"]:
        raise BuildRefused("credential_type does not match the build engine")
    if not isinstance(current.build_spec, dict) or not current.approved_spec_sha256:
        raise BuildRefused("plan requires build_spec and approved_spec_sha256")
    expected = build_spec_digest(document)
    if build_spec_digest(current.build_spec) != expected or current.approved_spec_sha256 != expected:
        raise BuildRefused("build specification does not match the approved specification digest")
    return current


def _verify_policy(bao_bin: str, name: str, expected: str) -> None:
    content = _run(bao_bin, ["policy", "read", name])
    if hashlib.sha256(content.encode()).hexdigest() != expected:
        raise BuildRefused("live reused policy does not match the approved content pin")


@contextmanager
def _delivery_files(path: Path):
    """Reserve private files before Bao writes, without following any symlink.

    Hold directory/file descriptors throughout delivery. Existing files are
    never opened or overwritten. Failed issuance leaves private partial files
    for explicit recovery; retries must use a newly reviewed destination.
    """
    directory = os.open("/", os.O_RDONLY | os.O_DIRECTORY)
    files = {}
    try:
        for component in path.parts[1:]:
            try:
                os.mkdir(component, 0o700, dir_fd=directory)
            except FileExistsError:
                pass
            child = os.open(component, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=directory)
            os.close(directory)
            directory = child
        info = os.fstat(directory)
        if info.st_uid != os.geteuid() or stat.S_IMODE(info.st_mode) != 0o700:
            raise BuildRefused("delivery directory must be owned by the caller with mode 0700")
        # Check both first so a pre-existing destination never causes a write.
        for name in ("role_id", "secret_id"):
            try:
                os.stat(name, dir_fd=directory, follow_symlinks=False)
            except FileNotFoundError:
                continue
            raise BuildRefused("credential destination already exists; refusing overwrite")
        for name in ("role_id", "secret_id"):
            files[name] = os.open(name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
                                  0o600, dir_fd=directory)
            os.fchmod(files[name], 0o600)
    except (OSError, BuildRefused) as exc:
        for fd in files.values():
            os.close(fd)
        os.close(directory)
        if isinstance(exc, BuildRefused):
            raise
        raise BuildRefused("cannot reserve safe credential delivery files") from None
    try:
        yield files
    finally:
        for fd in files.values():
            os.close(fd)
        os.close(directory)


def _write_credential(fd: int, value: str) -> None:
    if not value or "\n" in value or "\r" in value or "\x00" in value:
        raise BuildError("credential command returned an invalid response")
    with os.fdopen(os.dup(fd), "w") as stream:
        stream.write(value + "\n")
        stream.flush()
        os.fsync(stream.fileno())


def build_approle_kv_lane(plan: ConstructionPlan, spec: AppRoleKVSpec) -> dict[str, str]:
    """Create the policy + AppRole for an `openbao-approle-kv` plan, deliver role_id/secret_id.

    Refuses unless plan.is_approved(). Returns object names only (no
    secret material) and appends a metadata-only audit record.
    """
    spec = replace(spec)
    plan = _approved_build(plan, spec)
    if spec.reuse_policy:
        _verify_policy(spec.bao_bin, spec.policy_name, spec.reuse_policy_sha256)
    with _delivery_files(spec.delivery_dir) as files:
        if not spec.reuse_policy:
            policy_hcl = _policy_hcl(spec.kv_path, spec.kv_capabilities)
            _run(spec.bao_bin, ["policy", "write", spec.policy_name, "-"], input_text=policy_hcl)

        _run(
            spec.bao_bin,
            [
                "write",
                f"auth/approle/role/{spec.approle_name}",
                f"token_policies={spec.policy_name}",
                f"token_ttl={spec.token_ttl}",
                f"token_max_ttl={spec.token_max_ttl}",
                f"token_num_uses={spec.token_num_uses}",
                f"secret_id_ttl={spec.secret_id_ttl}",
            ],
        )

        role_id = _run(
            spec.bao_bin, ["read", "-field=role_id", f"auth/approle/role/{spec.approle_name}/role-id"]
        ).strip()
        secret_id = _run(
            spec.bao_bin,
            ["write", "-field=secret_id", "-f", f"auth/approle/role/{spec.approle_name}/secret-id"],
        ).strip()

        _write_credential(files["role_id"], role_id)
        _write_credential(files["secret_id"], secret_id)
    delivered_to = str(spec.delivery_dir)

    objects = {
        "approved_spec_sha256": plan.approved_spec_sha256,
        "policy_name": spec.policy_name,
        "approle_name": spec.approle_name,
        "kv_path": spec.kv_path,
        "delivered_to": delivered_to,
    }
    record_build(
        plan_id=plan.id,
        plan_path=str(plan.path),
        approved_by=plan.approved_by or "",
        approved_at=plan.approved_at or "",
        objects=objects,
        log_path=spec.audit_log_path,
    )
    return objects


def build_kubernetes_kv_lane(
    plan: ConstructionPlan, spec: KubernetesKVSpec
) -> dict[str, str]:
    """Create a policy-bound Kubernetes auth role without handling secret values."""
    spec = replace(spec)
    plan = _approved_build(plan, spec)
    _verify_policy(spec.bao_bin, spec.policy_name, spec.policy_sha256)

    _run(
        spec.bao_bin,
        [
            "write",
            f"auth/kubernetes/role/{spec.role_name}",
            f"bound_service_account_names={','.join(spec.service_account_names)}",
            f"bound_service_account_namespaces={','.join(spec.service_account_namespaces)}",
            f"policies={spec.policy_name}",
            f"ttl={spec.token_ttl}",
        ],
    )
    objects = {
        "approved_spec_sha256": plan.approved_spec_sha256,
        "policy_name": spec.policy_name,
        "kubernetes_role_name": spec.role_name,
        "service_account_names": ",".join(spec.service_account_names),
        "service_account_namespaces": ",".join(spec.service_account_namespaces),
    }
    record_build(
        plan_id=plan.id,
        plan_path=str(plan.path),
        approved_by=plan.approved_by or "",
        approved_at=plan.approved_at or "",
        objects=objects,
        log_path=spec.audit_log_path,
    )
    return objects