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 os
import subprocess
from dataclasses import dataclass, field
from pathlib import Path

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


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
    kv_capabilities: tuple[str, ...] = ("read",)
    token_ttl: str = "15m"
    token_max_ttl: str = "30m"
    token_num_uses: int = 0
    secret_id_ttl: str = "0"
    delivery_dir: Path | None = None
    bao_bin: str = "bao"
    audit_log_path: Path | None = None


def _run(bao_bin: str, args: list[str], input_text: str | None = None) -> str:
    proc = subprocess.run(
        [bao_bin, *args],
        input=input_text,
        capture_output=True,
        text=True,
        timeout=30,
    )
    if proc.returncode != 0:
        raise BuildError(f"`{bao_bin} {' '.join(args)}` failed: {proc.stderr.strip()[:300]}")
    return proc.stdout


def _policy_hcl(kv_path: str, capabilities: tuple[str, ...]) -> str:
    caps = ", ".join(f'"{c}"' for c in capabilities)
    return f'path "{kv_path}" {{\n  capabilities = [{caps}]\n}}\n'


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.
    """
    if not plan.is_approved():
        raise BuildRefused(
            f"plan {plan.id!r} is not approved "
            f"(status={plan.status!r}, approved_by={plan.approved_by!r}, "
            f"approved_at={plan.approved_at!r}) — refusing to build"
        )

    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()

    delivered_to = ""
    if spec.delivery_dir is not None:
        spec.delivery_dir.mkdir(parents=True, exist_ok=True)
        os.chmod(spec.delivery_dir, 0o700)
        role_id_path = spec.delivery_dir / "role_id"
        secret_id_path = spec.delivery_dir / "secret_id"
        role_id_path.write_text(role_id + "\n")
        secret_id_path.write_text(secret_id + "\n")
        os.chmod(role_id_path, 0o600)
        os.chmod(secret_id_path, 0o600)
        delivered_to = str(spec.delivery_dir)

    objects = {
        "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