scripts.check_infra_drift

Reference › scripts › check_infra_drift

The infra drift gate, the mechanism for ADR-0077 item 4. It detects drift between the infrastructure code and the live Incus infrastructure. No module or test imports it; a human or a scheduled gate runs it.

Invocation: main takes --tf-dir (the Terraform or OpenTofu directory), --state (the state file) and --report (write a JSON verdict). It copies the process environment and sets TF_IN_AUTOMATION to 1 when unset, then calls run_gate.

resolve_binary finds a usable tofu or terraform binary, checking versions with _version_tuple. plan_drift runs a plan in the given directory (default timeout 300 seconds) and returns the plan's exit code, its changes and the raw output. bridge_expected_config reads the network bridge settings the code expects, and bridge_drift compares them with the live values obtained by incus_network_get, which calls the Incus CLI for one network key. finish prints the verdict and returns the exit code, writing the JSON report when asked. The commands go through _run, which applies a timeout.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/check_infra_drift.py

#!/usr/bin/env python3
"""Infra drift gate — the mechanism for **ADR-0077 item 4**.

OpenTofu owns the Incus topology of all three environments (prod `default`,
`orchestrata-test`, `orchestrata-dev` — projects, bridges, ACLs, profiles and
every container), and the live fleets were IMPORTED into the host-owned state
at ``/var/lib/orchestrata/terraform/fleet.tfstate``. This gate is how drift is
detected instead of discovered:

1. ``tofu plan -detailed-exitcode`` over ``infra/terraform`` — exit 0 is
   zero-diff, exit 2 is drift (a container, bridge, ACL or profile no longer
   matches the declaration), anything else is a tool error.
2. The three bridge keys the lxc/incus provider never writes to state
   (``ipv4.address``, ``ipv4.nat``, ``ipv6.address`` — see networks.tf) are
   proven live with ``incus network get`` against the root's
   ``bridge_expected_config`` output, so the plan's silence on them is never
   mistaken for proof.

Verdicts (exit code):

* ``0`` — zero-diff AND every bridge key matches live.
* ``1`` — FAIL: drift, a MISSING state file, no usable binary, or an init /
  plan / output error. A missing state is a failure, never a skip: the
  September state was lost to a worktree gc and nothing noticed.
* ``0`` with a ``NOTE`` — no IaC root at ``--tf-dir``: a checkout without the
  IaC is not a fleet host (the health-check sandboxes), and the note is
  printed so the pass is never silent.

Binary resolution: ``$TOFU``, else ``tofu`` on PATH, else ``~/bin/tofu``, else
``terraform`` only when its version satisfies the root's ``required_version``
floor (1.6.0 — terraform 1.5.7 on this host cannot even init the root).

Stdlib only; runs from the deployed tree on a user timer
(deploy/systemd/orchestrata-infra-drift.timer) and from
``scripts/health_check.py``.
"""

from __future__ import annotations

import argparse
import json
import os
import shutil
import subprocess
import sys
from collections.abc import Callable, Mapping
from datetime import UTC, datetime
from pathlib import Path

RULE_ID = "infra-drift"
DEFAULT_TF_DIR = Path("infra/terraform")
DEFAULT_STATE = Path("/var/lib/orchestrata/terraform/fleet.tfstate")
REQUIRED_VERSION_FLOOR = (1, 6, 0)
BRIDGE_OUTPUT = "bridge_expected_config"
_DRIFT_MARKERS = (" will be ", " must be ", "forces replacement", "Plan:")

NetworkGet = Callable[[str, str], str | None]


def _run(
    cmd: list[str],
    cwd: Path | None,
    env: Mapping[str, str],
    timeout: int,
    stdin: str | None = None,
) -> tuple[int, str, str]:
    """Run ``cmd`` and return ``(returncode, stdout, stderr)``; never raises."""
    try:
        proc = subprocess.run(
            cmd,
            cwd=cwd,
            env=dict(env),
            input=stdin,
            capture_output=True,
            text=True,
            timeout=timeout,
        )
    except FileNotFoundError as exc:
        return 127, "", str(exc)
    except subprocess.TimeoutExpired:
        return 124, "", f"{cmd[0]} timed out after {timeout}s"
    return proc.returncode, proc.stdout, proc.stderr


def _version_tuple(binary: str, env: Mapping[str, str]) -> tuple[int, ...] | None:
    rc, out, _ = _run([binary, "version", "-json"], None, env, 30)
    if rc != 0:
        return None
    try:
        raw = json.loads(out)["terraform_version"]
    except ValueError, KeyError, TypeError:
        return None
    digits = raw.split("-", 1)[0].split(".")
    try:
        return tuple(int(d) for d in digits)
    except ValueError:
        return None


def resolve_binary(env: Mapping[str, str], home: Path | None = None) -> str | None:
    """``$TOFU`` > ``tofu`` on PATH > ``~/bin/tofu`` > a new-enough ``terraform``."""
    explicit = env.get("TOFU")
    if explicit:
        return explicit
    path_env = env.get("PATH", "")
    found = shutil.which("tofu", path=path_env)
    if found:
        return found
    home_dir = home if home is not None else Path(env.get("HOME", str(Path.home())))
    candidate = home_dir / "bin" / "tofu"
    if candidate.is_file() and os.access(candidate, os.X_OK):
        return str(candidate)
    terraform = shutil.which("terraform", path=path_env)
    if terraform is None:
        return None
    version = _version_tuple(terraform, env)
    if version is not None and version >= REQUIRED_VERSION_FLOOR:
        return terraform
    return None


def plan_drift(
    tf_dir: Path, binary: str, env: Mapping[str, str], timeout: int = 300
) -> tuple[int, list[str], str]:
    """``plan -detailed-exitcode``; returns ``(rc, action lines, stderr)``."""
    rc, out, err = _run(
        [
            binary,
            "plan",
            "-detailed-exitcode",
            "-input=false",
            "-no-color",
            "-lock=false",
        ],
        tf_dir,
        env,
        timeout,
    )
    lines = [
        line.strip()
        for line in out.splitlines()
        if any(marker in line for marker in _DRIFT_MARKERS)
    ]
    return rc, lines, err.strip()


def bridge_expected_config(
    tf_dir: Path, binary: str, env: Mapping[str, str]
) -> tuple[dict[str, dict[str, str]], str | None]:
    """The root's ``bridge_expected_config`` output, or ``({}, reason)``."""
    rc, out, err = _run(
        [binary, "output", "-no-color", "-json", BRIDGE_OUTPUT], tf_dir, env, 60
    )
    if rc != 0:
        return {}, (err or out).strip() or f"output {BRIDGE_OUTPUT} unavailable"
    try:
        value = json.loads(out)
    except ValueError as exc:
        return {}, f"output {BRIDGE_OUTPUT} is not JSON: {exc}"
    if not isinstance(value, dict):
        return {}, f"output {BRIDGE_OUTPUT} is not a map"
    return {
        str(name): {str(k): str(v) for k, v in keys.items()}
        for name, keys in value.items()
        if isinstance(keys, dict)
    }, None


def incus_network_get(network: str, key: str) -> str | None:
    """Live value of ``key`` on bridge ``network`` (project default), or None."""
    rc, out, _ = _run(
        ["incus", "network", "get", "--project", "default", network, key],
        None,
        os.environ,
        30,
    )
    if rc != 0:
        return None
    return out.strip()


def bridge_drift(
    expected: Mapping[str, Mapping[str, str]], network_get: NetworkGet
) -> list[str]:
    """One line per bridge key whose live value differs from the declaration."""
    findings: list[str] = []
    for network in sorted(expected):
        for key in sorted(expected[network]):
            want = expected[network][key]
            live = network_get(network, key)
            if live != want:
                shown = "<absent>" if live is None else repr(live)
                findings.append(
                    f"BRIDGE DRIFT: {network} {key} declared {want!r}, live {shown}"
                )
    return findings


def run_gate(
    tf_dir: Path,
    state_path: Path,
    env: Mapping[str, str],
    network_get: NetworkGet = incus_network_get,
    report_path: Path | None = None,
    out: Callable[[str], None] = print,
) -> int:
    """Run the whole gate; print the verdict lines; return the exit code."""
    report: dict[str, object] = {
        "rule": RULE_ID,
        "checked_at": datetime.now(UTC).isoformat(timespec="seconds"),
        "tf_dir": str(tf_dir),
        "state": str(state_path),
        "drift": [],
        "bridge_drift": [],
    }

    def finish(code: int, verdict: str, lines: list[str]) -> int:
        report["verdict"] = verdict
        report["exit"] = code
        for line in lines:
            out(line)
        if report_path is not None:
            report_path.parent.mkdir(parents=True, exist_ok=True)
            report_path.write_text(json.dumps(report, indent=2) + "\n")
        return code

    if not tf_dir.is_dir():
        return finish(0, "no-root", [f"NOTE: no IaC root at {tf_dir}; nothing to gate"])
    if not state_path.is_file():
        return finish(
            1,
            "missing-state",
            [
                f"MISSING STATE: {state_path} does not exist — the fleet is not "
                "imported (run infra/terraform/import.sh; ADR-0077 items 3-4)"
            ],
        )
    binary = resolve_binary(env)
    report["binary"] = binary
    if binary is None:
        return finish(
            1,
            "no-binary",
            [
                "no OpenTofu: set $TOFU, put `tofu` on PATH, or install a "
                f"terraform >= {'.'.join(map(str, REQUIRED_VERSION_FLOOR))}"
            ],
        )
    if not (tf_dir / ".terraform").is_dir():
        rc, _, err = _run(
            [binary, "init", "-input=false", "-no-color"], tf_dir, env, 300
        )
        if rc != 0:
            return finish(1, "init-error", [f"init failed: {err.strip()[:400]}"])

    rc, lines, err = plan_drift(tf_dir, binary, env)
    report["plan_exit"] = rc
    if rc == 2:
        report["drift"] = lines
        return finish(1, "drift", ["DRIFT: " + line for line in lines] or ["DRIFT"])
    if rc != 0:
        return finish(1, "plan-error", [f"plan failed (exit {rc}): {err[:400]}"])

    expected, reason = bridge_expected_config(tf_dir, binary, env)
    if reason is not None:
        report["bridge_note"] = reason
    findings = bridge_drift(expected, network_get)
    report["bridge_drift"] = findings
    if findings:
        return finish(1, "bridge-drift", findings)
    note = f" ({reason})" if reason else ""
    return finish(
        0,
        "zero-diff",
        [f"OK: zero-diff; {len(expected)} bridge(s) verified live{note}"],
    )


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n", 1)[0])
    parser.add_argument("--tf-dir", type=Path, default=DEFAULT_TF_DIR)
    parser.add_argument("--state", type=Path, default=DEFAULT_STATE)
    parser.add_argument("--report", type=Path, default=None, help="write JSON verdict")
    args = parser.parse_args(argv)
    env = dict(os.environ)
    env.setdefault("TF_IN_AUTOMATION", "1")
    return run_gate(args.tf_dir, args.state, env, report_path=args.report)


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