homelab-codex-ws/jobs/deploy-runner/action.py

#!/usr/bin/env python3
"""Validation and result-reporting helper for the host-side deploy runner.

The runner (deploy-runner.sh) is bash — it does flock, rsync and timeout, which
bash does well. Everything that needs to parse untrusted JSON, read
hosts/<node>/services.yaml or write an event in the exact node-agent format
lives here, where it is testable (jobs/deploy-runner/tests/).

Two subcommands, both designed to be consumed from bash:

  validate --file <action.json> --node <name> --repo <path>
      Prints shell assignments for `eval`:
          OK='true'|'false'
          ACTION_ID='...'      (best effort — set even when OK=false, so the
                                runner can still report a result for it)
          SERVICE='...'
          ERROR='...'
      Always exits 0 unless the arguments themselves are unusable; the verdict
      is in OK. Nothing from the action file is ever executed or interpolated
      into a command — only a validated service NAME is handed back.

  emit-result --events-dir <dir> --node <n> --action-id <id> --service <svc>
              --success true|false [--error <msg>]
      Writes evt-<node>-<ts>-action_result-<slug>.json into the node's event
      directory, byte-compatible with node_agent.emit_event() so the control
      plane executor's _find_action_result() picks it up unchanged. Prints the
      path it wrote.

Validation rules (all must hold):
  1. the file is JSON with type == "redeploy"
  2. action["node"] == this node — defense in depth; the inbox is already
     node-scoped, but a mis-delivered file must never deploy here
  3. action_id and service match strict name patterns (no path traversal)
  4. the service is listed in hosts/<node>/services.yaml — the repo's desired
     state is the authority on what this node may run, so a rogue or stale
     dispatch cannot pull an arbitrary stack onto the node
  5. services/<service>/docker-compose.yml exists in the repo
"""
from __future__ import annotations

import argparse
import json
import re
import shlex
import sys
import time
from datetime import datetime, timezone
from pathlib import Path

ACTION_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
SERVICE_RE = re.compile(r"^[a-z0-9][a-z0-9._-]{0,63}$")
ACTION_TYPE = "redeploy"


class Invalid(Exception):
    """Validation failure carrying a human-readable reason."""


def _load_desired_services(repo: Path, node: str) -> set[str]:
    """Service names listed in hosts/<node>/services.yaml."""
    try:
        import yaml
    except ImportError as exc:  # pragma: no cover - environment problem
        raise Invalid(f"PyYAML not available on this node: {exc}") from exc

    path = repo / "hosts" / node / "services.yaml"
    if not path.is_file():
        raise Invalid(f"no desired-state manifest for node '{node}' ({path})")
    try:
        data = yaml.safe_load(path.read_text()) or {}
    except Exception as exc:
        raise Invalid(f"unreadable {path}: {exc}") from exc

    services = data.get("services") or {}
    if isinstance(services, dict):
        return {str(k) for k in services.keys()}
    if isinstance(services, list):
        return {str(s) for s in services}
    raise Invalid(f"unexpected 'services' shape in {path}: {type(services).__name__}")


def validate(action_file: Path, node: str, repo: Path) -> tuple[str, str]:
    """Return (action_id, service) for a valid redeploy action, else raise Invalid.

    action_id is also returned inside the Invalid path via the caller's
    best-effort parse, so a rejected action can still be reported back.
    """
    try:
        action = json.loads(action_file.read_text())
    except Exception as exc:
        raise Invalid(f"unparseable action file: {exc}") from exc
    if not isinstance(action, dict):
        raise Invalid("action file is not a JSON object")

    action_id = str(action.get("action_id") or action_file.stem)
    if not ACTION_ID_RE.match(action_id):
        raise Invalid(f"invalid action_id: {action_id!r}")

    action_type = action.get("type")
    if action_type != ACTION_TYPE:
        raise Invalid(
            f"action type {action_type!r} is not executable by the deploy runner "
            f"(only {ACTION_TYPE!r})"
        )

    target_node = action.get("node")
    if target_node != node:
        raise Invalid(f"action addressed to node {target_node!r}, not {node!r} — refused")

    service = action.get("service")
    if not isinstance(service, str) or not SERVICE_RE.match(service):
        raise Invalid(f"invalid service name: {service!r}")

    desired = _load_desired_services(repo, node)
    if service not in desired:
        raise Invalid(
            f"service '{service}' is not in hosts/{node}/services.yaml — "
            f"refusing to deploy a service this node is not declared to run"
        )

    compose = repo / "services" / service / "docker-compose.yml"
    if not compose.is_file():
        raise Invalid(f"no compose file in repo for '{service}' ({compose})")

    return action_id, service


def _best_effort_action_id(action_file: Path) -> str:
    try:
        data = json.loads(action_file.read_text())
        candidate = str(data.get("action_id") or action_file.stem)
    except Exception:
        candidate = action_file.stem
    return candidate if ACTION_ID_RE.match(candidate) else ""


def _emit_assignments(ok: bool, action_id: str, service: str, error: str) -> str:
    return "\n".join(
        f"{key}={shlex.quote(value)}"
        for key, value in (
            ("OK", "true" if ok else "false"),
            ("ACTION_ID", action_id),
            ("SERVICE", service),
            ("ERROR", error),
        )
    )


def emit_result(events_dir: Path, node: str, action_id: str, service: str,
                success: bool, error: str) -> Path:
    """Write an action_result event in node_agent.emit_event()'s exact format.

    The control-plane executor globs evt-*-action_result-*.json in the node's
    event directory and matches payload.action_id, comparing the unix timestamp
    embedded in the filename against the action's started_at — so both the name
    and the payload shape matter here (see executor._find_action_result).
    """
    ts = int(time.time())
    slug = re.sub(r"[^a-z0-9]", "-", (service or "node").lower())[:32].strip("-")
    event_id = f"evt-{node}-{ts}-action_result-{slug}"
    event = {
        "id": event_id,
        "timestamp": ts,
        "date": datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z"),
        "type": "action_result",
        "severity": "info" if success else "high",
        "node": node,
        "service": service or "",
        "message": f"Action {action_id} {'succeeded' if success else 'failed'}",
        "payload": {
            "action_id": action_id,
            "success": success,
            "error": error,
            "node": node,
            "source": "deploy-runner",
        },
    }
    events_dir.mkdir(parents=True, exist_ok=True)
    path = events_dir / f"{event_id}.json"
    path.write_text(json.dumps(event, indent=2))
    return path


def main(argv=None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    sub = parser.add_subparsers(dest="command", required=True)

    p_val = sub.add_parser("validate")
    p_val.add_argument("--file", required=True, type=Path)
    p_val.add_argument("--node", required=True)
    p_val.add_argument("--repo", required=True, type=Path)

    p_emit = sub.add_parser("emit-result")
    p_emit.add_argument("--events-dir", required=True, type=Path)
    p_emit.add_argument("--node", required=True)
    p_emit.add_argument("--action-id", required=True)
    p_emit.add_argument("--service", default="")
    p_emit.add_argument("--success", required=True, choices=["true", "false"])
    p_emit.add_argument("--error", default="")

    args = parser.parse_args(argv)

    if args.command == "validate":
        try:
            action_id, service = validate(args.file, args.node, args.repo)
        except Invalid as exc:
            print(_emit_assignments(False, _best_effort_action_id(args.file), "", str(exc)))
            return 0
        print(_emit_assignments(True, action_id, service, ""))
        return 0

    path = emit_result(
        args.events_dir, args.node, args.action_id, args.service,
        args.success == "true", args.error,
    )
    print(path)
    return 0


if __name__ == "__main__":
    sys.exit(main())