Python over the Terraform Plan JSON — the plan is an object, not a page

A policy engine is a good answer to the second rule an organization writes. It is a poor answer to the first.

Section IFrame

A policy engine is a good answer to the second rule an organization writes. It is a poor answer to the first.

The first rule is always small and always specific. No storage account in this subscription without a cost_center tag. No virtual machine outside the four approved SKUs. No resource whose name breaks the convention the platform team agreed on in March. Writing that rule in Rego means learning Rego, installing conftest in the runner, and deciding where the policy files live before anyone knows whether the rule survives contact with the team.

Writing it in Python means reading a JSON document.

The thing worth learning today is that terraform show -json produces a documented, versioned schema, not an engine-private blob. Once that is understood, the policy engine becomes an optional convenience rather than a prerequisite, and the checks an organization actually needs stop waiting on a tooling decision.

Section IILanguage Idiom

The plan is an object, not a page

terraform plan prints a page for a person. terraform show -json tfplan emits an object for a program. Same run, same data, two renderings, and the second one carries a format_version field so a script can refuse a shape it does not understand.

The array that matters is resource_changes. Each element carries the address, the type, the provider, and a change object:

{
  "address": "azurerm_storage_account.artifacts",
  "type": "azurerm_storage_account",
  "name": "artifacts",
  "mode": "managed",
  "change": {
    "actions": ["create"],
    "before": null,
    "after": { "name": "hedroniteartifacts", "min_tls_version": "TLS1_2" },
    "after_unknown": { "id": true, "primary_access_key": true }
  }
}

Read the actions list rather than comparing before to after. Terraform has already done the comparison and written the verdict down. The list holds one of ["create"], ["update"], ["delete"], ["no-op"], ["read"], or the two-element ["delete", "create"] that means replacement. A checker that diffs the dictionaries reimplements work Terraform finished, and gets the replacement case wrong.

after_unknown is the parallel structure that marks every attribute the provider cannot resolve until apply. Its keys mirror after; the values are true where the value is pending. This is the same trap the Ops lesson names from the Rego side, and Python hits it harder because dict.get("flag") returns None and None == True is quietly false.

Three idioms the checker leans on

subprocess.run with check=True. Gift and the Python for DevOps authors demonstrate the failure directly: subprocess.run(['ls', '/doesnotexist'], capture_output=True, universal_newlines=True, check=True) raises CalledProcessError rather than returning a result object nobody inspected (Ch. 3, pp. 117-118). A wrapper around a CLI that omits check=True will happily parse an empty string as JSON and report zero violations on a Terraform run that never completed. The most dangerous checker is the one that passes because it read nothing.

dict.get chained with an explicit default that is not None. Where a missing attribute and an unsafe attribute must be distinguished, a sentinel does the work. Where they must be treated the same, say so in one place rather than in each rule.

sys.exit with a code the runner reads. The 08-02 lesson set the exit-code convention for this arc: 1 for a run that did its job and found a problem, 2 for a run that could not start. A policy checker inherits it exactly. A violation is 1. A malformed plan file is 2, because a supervisor should page rather than retry.

Section IIICode Worked Example

The tool has one job: read a plan, apply a list of rules, report every violation, exit nonzero if any fired.

The rule shape first. A rule is a function from a resource-change dictionary to a list of complaint strings, registered by decorator so adding a rule means adding a function.

from typing import Callable

RuleFn = Callable[[dict], list[str]]
RULES: dict[str, RuleFn] = {}


def rule(name: str):
    def register(fn: RuleFn) -> RuleFn:
        RULES[name] = fn
        return fn
    return register

The registry uses the decorator-with-argument shape rather than a bare decorator so the rule name lives at the call site where a reader will look for it. The 08-08 lesson reached the same place through __set_name__, which lets a descriptor learn the name it was bound to; here the name is passed in, because a plain function has no owner class to learn it from.

Now the plan reader. Two subprocess calls, one temporary file, one refusal:

import json
import subprocess
import sys
import tempfile
from pathlib import Path

SUPPORTED_FORMATS = {"1.0", "1.1", "1.2"}


def read_plan(chdir: Path) -> dict:
    with tempfile.TemporaryDirectory() as tmp:
        binary = Path(tmp) / "tfplan"
        subprocess.run(
            ["terraform", f"-chdir={chdir}", "plan", "-out", str(binary), "-input=false"],
            check=True,
            capture_output=True,
            text=True,
        )
        shown = subprocess.run(
            ["terraform", f"-chdir={chdir}", "show", "-json", str(binary)],
            check=True,
            capture_output=True,
            text=True,
        )
    plan = json.loads(shown.stdout)
    version = plan.get("format_version", "")
    if version not in SUPPORTED_FORMATS:
        raise RuntimeError(f"unsupported plan format_version {version!r}")
    return plan

The binary plan file lives inside a TemporaryDirectory and dies with it. Brikman is blunt about why that matters: a plan file holds the same secrets the state file holds, in the clear (Ch. 6, Plan files, pp. 341-343). A checker that writes tfplan into the repository working directory has handed the secret to whatever runs next, including the archive step that uploads build artifacts. The with block is the whole cure.

The version check earns its four lines the first time HashiCorp bumps the schema. Without it the script keeps running and reports zero violations against keys that moved.

Two rules, written against Azure resources to match the day's Ops lesson:

REQUIRED_TAGS = {"cost_center", "env", "owner"}
APPROVED_VM_SIZES = {"Standard_D2s_v5", "Standard_D4s_v5", "Standard_E4s_v5"}
MUTATING = {"create", "update"}


@rule("required-tags")
def required_tags(rc: dict) -> list[str]:
    if not MUTATING & set(rc["change"]["actions"]):
        return []
    after = rc["change"].get("after") or {}
    tags = after.get("tags") or {}
    missing = REQUIRED_TAGS - set(tags)
    if not missing:
        return []
    return [f"{rc['address']} missing tags: {', '.join(sorted(missing))}"]


@rule("approved-vm-size")
def approved_vm_size(rc: dict) -> list[str]:
    if rc["type"] != "azurerm_linux_virtual_machine":
        return []
    if not MUTATING & set(rc["change"]["actions"]):
        return []
    after = rc["change"].get("after") or {}
    unknown = rc["change"].get("after_unknown") or {}
    if unknown.get("size"):
        return [f"{rc['address']} size is unknown at plan time; refusing"]
    size = after.get("size")
    if size in APPROVED_VM_SIZES:
        return []
    return [f"{rc['address']} uses unapproved size {size!r}"]

The second rule fails closed on the unknown. It refuses rather than skipping, and it says which resource and why. A rule that returns [] on an unknown value is a rule that lets a computed SKU through, and computed SKUs are exactly how a module author routes around a policy without meaning to.

The runner ties it together and speaks the exit-code convention:

def main(chdir: Path) -> int:
    try:
        plan = read_plan(chdir)
    except (subprocess.CalledProcessError, json.JSONDecodeError, RuntimeError) as exc:
        print(f"policy-check could not run: {exc}", file=sys.stderr)
        return 2

    violations: list[str] = []
    for rc in plan.get("resource_changes", []):
        if rc.get("mode") != "managed":
            continue
        for name, fn in RULES.items():
            violations.extend(f"[{name}] {v}" for v in fn(rc))

    for line in violations:
        print(line, file=sys.stderr)
    print(f"policy-check: {len(RULES)} rules, {len(violations)} violations")
    return 1 if violations else 0


if __name__ == "__main__":
    sys.exit(main(Path(sys.argv[1] if len(sys.argv) > 1 else ".")))

The mode != "managed" filter drops data sources, which show up in resource_changes with a read action and no tags anybody can control. Skipping them by mode rather than by action keeps the rule honest when a future Terraform version adds a mode.

Section IVConnection to Today's Ops Lesson

The Ops lesson mounts Sentinel inside the HCP Terraform run pipeline and Rego beside it, and names the plan as the correct input for both. The rule this script implements as required_tags is the same rule the Ops lesson writes in Sentinel as a filter plus an all over tfplan.resource_changes. Identical logic, identical input, three syntaxes.

The trade is worth stating plainly. Sentinel gives typed imports, a run integration, and three enforcement levels the platform team configures without touching code. Python gives everything else the runner can reach: an internal service that owns the cost-center list, a naming API, yesterday's Azure Advisor export. Rules about the world outside the plan are hard in Rego and easy in Python.

Run both. The engine holds the rules that are about the plan alone. The script holds the rules that need a second source.

Section VPrior-Lesson Reach

07-28, the HCP Terraform API via terrasnek. That lesson wrote to Terraform from Python: creating workspaces, triggering runs, reading state versions. This one reads Terraform's output. The Python-around-TF slot now has both directions, and they compose. A script that triggers a run through the API can pull the plan JSON from the run's plan endpoint and check it with the rules above, which turns the local subprocess version into the CI version without changing a single rule function.

08-08, the attribute protocol. Yesterday's Python lesson built a policy object that refuses a bad value at assignment through a data descriptor. Today's checker refuses a bad value at plan time through a function in a registry. Same intent, two layers: one protects the process holding the value, the other protects the world the value is about to reach.

08-02, exit-code discipline. 1 for found-a-problem, 2 for could-not-run. Held here without modification.

Section VIClosing

The plan file is a document. terraform show -json gives it to you in a schema HashiCorp versions and documents. Every governance rule an organization wants, in the first year, is a walk over one array in that document.

Write the version check before the first rule. Keep the plan file inside a TemporaryDirectory, because it carries the secrets the state file carries. Read change.actions and never diff before against after yourself. Where after_unknown marks an attribute you were about to judge, refuse rather than skip, and say which resource you refused.

Then add the rule you have been meaning to write for three months. It will take twenty lines.

Examine well.

Related

• Prior arc: Python for the Terraform Cloud API — Workspace CRUD, Run Triggering, and State-Version Reads via terrasnek
• Language hub: Python
• Grounding tome: Python for DevOps (Ch. 3, Creating Command-Line Tools, pp. 117-118)
• Paired Ops lesson: Policy as Code for Terraform on Azure
• Paired Cert lesson: TF Associate 003 — Sensitive Variables, Secrets in State, and HCP Terraform Policy Enforcement