How to Govern AI-Generated Infrastructure with Policy as Code and OPA [Full Handbook]

<p>Modern models generate syntactically correct code nearly 100% of the time. Veracode's 2026 report puts it plainly: "Syntax is effectively solved."</p> <p>That reads like a milestone, but it's the reason you have a problem.</p> <p>The <a href="https://www.veracode.com/blog/2026-genai-code-security-report-ai-risk/">same report</a> tested more than a hundred models and found the average security pass rate at 56%, "barely changed from 55% in the first report", with roughly 44% of generation tasks introducing a risky vulnerability.</p> <p>Functional correctness and security turn out to be separate problems, and only one of them is close to solved.</p> <p>That result is neither an outlier nor new. At IEEE Security and Privacy in 2022, a team at NYU Tandon ran GitHub Copilot through 89 security-relevant scenarios, generated 1,689 programs, and found <a href="https://arxiv.org/abs/2108.09293">roughly 40% of them vulnerable</a> to something on MITRE's CWE Top 25. The paper was later selected as a <em>Communications of the ACM</em> research highlight.</p> <p>In November 2024, Georgetown's Center for Security and Emerging Technology <a href="https://cset.georgetown.edu/publication/cybersecurity-risks-of-ai-generated-code/">evaluated five LLMs</a> and reported that almost half the snippets they produced contained bugs that could lead to exploitation. Four years, four independent teams, four methodologies, and the same answer each time.</p> <p>At ACM CCS in 2023, Neil Perry, Megha Srivastava, Deepak Kumar, and Dan Boneh at Stanford <a href="https://arxiv.org/abs/2211.03622">put the developers into the experiment</a>: 47 participants, five security-related programming tasks, three languages, with 33 given an AI assistant and 14 not. The assisted group wrote significantly less secure code, and was <em>more</em> likely to believe the code it wrote was secure.</p> <p>It's a small study, and it explains why the problem doesn't correct itself: the mechanism that would normally catch this (a developer looking harder at code that worries them) is the exact mechanism the tooling switches off.</p> <p>Those studies all measure application code. But infrastructure code is the harder case, because a bad security group never fails: it works exactly as written, serving traffic to whoever asks, and the only thing that objects is a person reading a diff.</p> <p>I can't review that volume by reading it, and neither can anybody else. What I can do is write the rules down in a form a computer checks on every change, which is what Policy as Code means.</p> <p>In this handbook, I walk you through building that check. We'll point it at a real vulnerable repository, watch the obvious version of it clear five of the nine violations sitting in front of it, and then fix it.</p> <p>By the end, you'll know how to:</p> <ul> <li><p>Write a Rego policy against the JSON that <code>terraform show -json</code> produces.</p> </li> <li><p>Test a policy the way you test application code, with fixtures and a coverage report.</p> </li> <li><p>Build a command-line gate with an exit-code contract that a CI pipeline can trust.</p> </li> <li><p>Block non-compliant workloads at Kubernetes admission time using CEL.</p> </li> <li><p>Have a model write a policy and let <code>opa check</code> and your own tests decide whether to keep it.</p> </li> <li><p>Authorise an AI agent's tool calls from the same policy engine.</p> </li> </ul> <h2 id="heading-table-of-contents">Table of Contents</h2> <ul> <li><p><a href="#heading-prerequisites">Prerequisites</a></p> </li> <li><p><a href="#heading-key-terms-in-plain-english">Key Terms in Plain English</a></p> </li> <li><p><a href="#heading-step-1-fetch-real-infrastructure-to-test-against">Step 1: Fetch Real Infrastructure to Test Against</a></p> </li> <li><p><a href="#heading-step-2-write-the-tests-before-the-policy">Step 2: Write the Tests Before the Policy</a></p> </li> <li><p><a href="#heading-step-3-write-the-policy-until-the-tests-pass">Step 3: Write the Policy Until the Tests Pass</a></p> </li> <li><p><a href="#heading-step-4-point-it-at-the-real-plan">Step 4: Point It at the Real Plan</a></p> </li> <li><p><a href="#heading-step-5-turn-the-verdict-into-an-exit-code">Step 5: Turn the Verdict into an Exit Code</a></p> </li> <li><p><a href="#heading-step-6-enforce-at-admission-time">Step 6: Enforce at Admission Time</a></p> </li> <li><p><a href="#heading-step-7-let-a-model-write-the-policy">Step 7: Let a Model Write the Policy</a></p> </li> <li><p><a href="#heading-step-8-govern-the-agent-itself">Step 8: Govern the Agent Itself</a></p> </li> <li><p><a href="#heading-step-9-what-i-got-wrong">Step 9: What I Got Wrong</a></p> </li> <li><p><a href="#heading-limits-of-the-check">Limits of the Check</a></p> </li> <li><p><a href="#heading-conclusion">Conclusion</a></p> </li> </ul> <h2 id="heading-prerequisites">Prerequisites</h2> <p>You need:</p> <ul> <li><p>A terminal and a working <code>python3</code> (3.10 or newer).</p> </li> <li><p><code>jq</code>, for reading JSON at the command line.</p> </li> <li><p>About 700 MB of disk, because the AWS Terraform provider is large.</p> </li> <li><p>An Anthropic API key, but only for Step 7. Every other step runs offline.</p> </li> </ul> <pre><code class="language-bash">mkdir policy-lab &amp;amp;&amp;amp; cd policy-lab python3 -m venv .venv source .venv/bin/activate pip install anthropic curl -L -o opa https://openpolicyagent.org/downloads/v1.20.2/opa_darwin_arm64_static chmod +x opa &amp;amp;&amp;amp; sudo mv opa /usr/local/bin/ curl -L -o tf.zip https://releases.hashicorp.com/terraform/1.14.2/terraform_1.14.2_darwin_arm64.zip unzip tf.zip &amp;amp;&amp;amp; sudo mv terraform /usr/local/bin/ </code></pre> <p>On Windows, activate the environment with <code>.venv\Scripts\activate</code>, and swap the two download URLs for <code>opa_windows_amd64.exe</code> and <code>terraform_1.14.2_windows_amd64.zip</code>.</p> <p>I ran everything below on <strong>OPA 1.20.2</strong>, <strong>Terraform 1.14.2,</strong> and <strong>AWS provider 6.x</strong>, on macOS. The policy syntax is stable across OPA 1.x.</p> <p>If you're on OPA 0.x, every rule here needs <code>import rego.v1</code> added at the top, and I would upgrade instead. The violation counts depend on the AWS provider version only through the shape of the plan JSON, which has been stable since provider 5.</p> <h2 id="heading-key-terms-in-plain-english">Key Terms in Plain English</h2> <ul> <li><p><strong>Policy as Code</strong>: a rule your organisation has already agreed on, written as a program that takes a proposed change and returns a decision.</p> </li> <li><p><strong>Rego</strong>: the query language Open Policy Agent evaluates. It's declarative: a rule body is a list of conditions that must all hold.</p> </li> <li><p><strong>Plan JSON</strong>: the machine-readable description of what Terraform is about to do, produced by <code>terraform show -json</code>. This is what the policy reads, so your <code>.tf</code> files never reach it.</p> </li> <li><p><strong>Admission control</strong>: the point inside the Kubernetes API server where an object can be rejected before it's stored.</p> </li> <li><p><strong>CEL</strong>: Common Expression Language, the small expression language Kubernetes evaluates natively inside the API server, with no webhook to deploy.</p> </li> <li><p><strong>False clearance</strong>: a resource the policy passed that it should have failed. Nobody ever notices one, so it goes unmeasured unless you go looking for it.</p> </li> </ul> <img src="https://cdn.hashnode.com/uploads/covers/5f3a74bfc4d5973f55c91c8c/70640afe-46a8-4e6c-8953-3f97ed98254a.png" alt="Diagram titled &quot;Three decision points, three chances to say no&quot;, with three rows. The plan time row runs from terraform plan JSON to policy_gate.py to merge or block the PR. The admission time row runs from kubectl apply to ValidatingAdmissionPolicy to admit or reject the Pod. The call time row runs from agent picks a tool to the agent.authz decision to allow, deny or ask a human. Arrows point left to right from ingredient to product, and a caption reads one decision per boundary: CEL inside the API server, Rego either side." style="display: block;" width="600" height="400" loading="lazy"> <p>The same judgement happens in three places, and only the middle one is specific to Kubernetes.</p> <h2 id="heading-step-1-fetch-real-infrastructure-to-test-against">Step 1: Fetch Real Infrastructure to Test Against</h2> <p>I didn't want to invent a vulnerable Terraform file, because inventing one means inventing the bug, and then the policy only catches the bug I planted. So I went looking for code somebody else had written and published.</p> <p><a href="https://github.com/bridgecrewio/terragoat">TerraGoat</a> is a deliberately vulnerable Terraform repository published by Bridgecrew. Fetch its EC2 module at a pinned commit:</p> <pre><code class="language-bash">SHA=729f8da62c6a85ce4af5ad3d123de97776d954c4 curl -s "https://raw.githubusercontent.com/bridgecrewio/terragoat/$SHA/terraform/aws/ec2.tf" \ | sed -n '77,96p' </code></pre> <pre><code class="language-hcl">resource "aws_security_group" "web-node" { # security group is open to the world in SSH port name = "${local.resource_prefix.value}-sg" description = "${local.resource_prefix.value} Security Group" vpc_id = aws_vpc.web_vpc.id ingress { from_port = 80 to_port = 80 protocol = "tcp" cidr_blocks = [ "0.0.0.0/0"] } ingress { from_port = 22 to_port = 22 protocol = "tcp" cidr_blocks = [ "0.0.0.0/0"] } </code></pre> <p>The comment on line two is TerraGoat's own, and port 22 open to the world is the finding it points at.</p> <p>TerraGoat's module won't initialise on modern Terraform, because it still declares <code>type = "string"</code> in quotes, which Terraform 0.12 deprecated and 1.x rejects. So I lifted the resource into a minimal module of my own, replacing only the two references to TerraGoat's internal locals.</p> <p>Create <code>main.tf</code>:</p> <pre><code class="language-hcl">terraform { required_version = "&amp;gt;= 1.9" required_providers { aws = { source = "hashicorp/aws" version = "~&amp;gt; 6.0" } } } # Mock credentials. This configuration is only ever planned, never applied, # so the provider must not try to reach AWS. provider "aws" { region = "us-west-2" access_key = "mock" secret_key = "mock" skip_credentials_validation = true skip_metadata_api_check = true skip_requesting_account_id = true skip_region_validation = true } resource "aws_vpc" "web_vpc" { cidr_block = "10.0.0.0/16" } # Verbatim from bridgecrewio/terragoat, terraform/aws/ec2.tf, commit 729f8da. # Only the two references to TerraGoat's own locals are replaced with literals. resource "aws_security_group" "web-node" { name = "terragoat-sg" description = "terragoat Security Group" vpc_id = aws_vpc.web_vpc.id ingress { from_port = 80 to_port = 80 protocol = "tcp" cidr_blocks = [ "0.0.0.0/0"] } ingress { from_port = 22 to_port = 22 protocol = "tcp" cidr_blocks = [ "0.0.0.0/0"] } egress { from_port = 0 to_port = 0 protocol = "-1" cidr_blocks = [ "0.0.0.0/0"] } depends_on = [aws_vpc.web_vpc] tags = { git_commit = "d68d2897add9bc2203a5ed0632a5cdd8ff8cefb0" git_file = "terraform/aws/ec2.tf" git_last_modified_at = "2020-06-16 14:46:24" git_org = "bridgecrewio" git_repo = "terragoat" } } </code></pre> <p>Produce the plan JSON:</p> <pre><code class="language-bash">terraform init terraform plan -out=tfplan.binary terraform show -json tfplan.binary &amp;gt; plan.json </code></pre> <p>The mock credentials matter: <code>terraform plan</code> on a create-only configuration never calls AWS, so with <code>skip_credentials_validation</code> and its three siblings the provider won't try to authenticate, and nothing is ever applied.</p> <h2 id="heading-step-2-write-the-tests-before-the-policy">Step 2: Write the Tests Before the Policy</h2> <p>The rule I wanted was: <em>no security group may expose an administrative port to the public internet.</em></p> <p>That sounds like one line of code, and the tests are where I pin down why it's not. Create <code>policy/network_test.rego</code>:</p> <pre><code class="language-rego">package terraform.network_test import data.terraform.network plan(resources) := {"resource_changes": resources} security_group(ingress) := { "address": "aws_security_group.web", "type": "aws_security_group", "change": {"actions": ["create"], "after": {"ingress": [ingress]}}, } test_denies_ssh_open_to_the_world if { fixture := plan([security_group({ "from_port": 22, "to_port": 22, "protocol": "tcp", "cidr_blocks": ["0.0.0.0/0"], })]) count(network.deny) == 1 with input as fixture } # A from_port equality check would miss this. The range check does not. test_denies_wide_open_port_range if { fixture := plan([security_group({ "from_port": 0, "to_port": 65535, "protocol": "tcp", "cidr_blocks": ["0.0.0.0/0"], })]) count(network.deny) == 4 with input as fixture } test_denies_ipv6_route_to_the_world if { fixture := plan([security_group({ "from_port": 22, "to_port": 22, "protocol": "tcp", "ipv6_cidr_blocks": ["::/0"], })]) count(network.deny) == 1 with input as fixture } test_denies_standalone_ingress_rule if { fixture := plan([{ "address": "aws_vpc_security_group_ingress_rule.ssh", "type": "aws_vpc_security_group_ingress_rule", "change": {"actions": ["create"], "after": { "from_port": 22, "to_port": 22, "ip_protocol": "tcp", "cidr_ipv4": "0.0.0.0/0", "cidr_ipv6": null, }}, }]) count(network.deny) == 1 with input as fixture } test_denies_deprecated_standalone_rule if { fixture := plan([{ "address": "aws_security_group_rule.ssh", "type": "aws_security_group_rule", "change": {"actions": ["create"], "after": { "type": "ingress", "from_port": 22, "to_port": 22, "cidr_blocks": ["0.0.0.0/0"], }}, }]) count(network.deny) == 1 with input as fixture } test_allows_ssh_from_a_private_range if { fixture := plan([security_group({ "from_port": 22, "to_port": 22, "protocol": "tcp", "cidr_blocks": ["10.0.0.0/8"], })]) count(network.deny) == 0 with input as fixture } test_allows_https_from_the_world if { fixture := plan([security_group({ "from_port": 443, "to_port": 443, "protocol": "tcp", "cidr_blocks": ["0.0.0.0/0"], })]) count(network.deny) == 0 with input as fixture } # An all-protocols rule opens every port, whatever its port fields say. # The first version of this test asserted the opposite and hid the bug. test_denies_all_protocols_rule_open_to_the_world if { fixture := plan([security_group({ "from_port": 0, "to_port": 0, "protocol": "-1", "cidr_blocks": ["0.0.0.0/0"], })]) count(network.deny) == 4 with input as fixture } # Ports the policy cannot read are reported, never passed. test_reports_a_world_open_rule_with_unreadable_ports if { fixture := plan([security_group({ "from_port": 22, "to_port": null, "protocol": "tcp", "cidr_blocks": ["0.0.0.0/0"], })]) count(network.deny) == 1 with input as fixture } test_reports_string_ports if { fixture := plan([security_group({ "from_port": "22", "to_port": "22", "protocol": "tcp", "cidr_blocks": ["0.0.0.0/0"], })]) count(network.deny) == 1 with input as fixture } # Unreadable ports on a rule that is not open to the world stay quiet. test_ignores_unreadable_ports_on_a_private_range if { fixture := plan([security_group({ "from_port": 22, "to_port": null, "protocol": "tcp", "cidr_blocks": ["10.0.0.0/8"], })]) count(network.deny) == 0 with input as fixture } # `considered` drives the gate's pass-or-vacuous decision, so it needs # tests of its own even though it takes no part in the judgement. test_considers_every_ingress_bearing_type if { fixture := plan([ security_group({}), {"address": "aws_vpc_security_group_ingress_rule.a", "type": "aws_vpc_security_group_ingress_rule", "change": {"actions": ["create"], "after": {}}}, {"address": "aws_security_group_rule.b", "type": "aws_security_group_rule", "change": {"actions": ["create"], "after": {}}}, ]) count(network.considered) == 3 with input as fixture } test_does_not_consider_unrelated_types if { fixture := plan([{ "address": "aws_vpc.main", "type": "aws_vpc", "change": {"actions": ["create"], "after": {}}, }]) count(network.considered) == 0 with input as fixture } </code></pre> <p>Here's what that file encodes:</p> <ol> <li><p><code>with input as fixture</code> swaps in a fake plan for one expression, which is how a policy is tested without a cloud account.</p> </li> <li><p><code>test_denies_wide_open_port_range</code> expects <strong>four</strong> violations, one per administrative port, because an ingress rule describes a range. A <code>0-65535</code> rule opens SSH exactly as wide as an explicit port 22 rule while sailing past an equality check.</p> </li> <li><p>Three tests cover three <em>other</em> shapes Terraform uses for the same idea: the IPv6 field, the modern standalone <code>aws_vpc_security_group_ingress_rule</code>, and the deprecated <code>aws_security_group_rule</code>. I didn't write these first, and Step 9 explains where they came from.</p> </li> <li><p>The two <code>test_allows_</code> cases matter as much as the denials. A policy that rejects everything passes every deny test and is worthless.</p> </li> <li><p>The last three tests arrived after the policy was already "finished", and Step 9 explains where they came from. An all-protocols rule opens every port whatever its port fields say, and a rule whose ports the policy can't read has to be reported.</p> </li> </ol> <h2 id="heading-step-3-write-the-policy-until-the-tests-pass">Step 3: Write the Policy Until the Tests Pass</h2> <p>Terraform describes ingress in four shapes, so the policy normalises all four into one set and then judges that set once. Create <code>policy/network.rego</code>:</p> <pre><code class="language-rego"># METADATA # title: No admin port is reachable from the public internet # description: | # Terraform describes ingress in four different shapes. Each one is # normalised into a single `exposures` set first, so the judgement below # is written once and a new shape only costs one more helper rule. # # Two things here are deliberate rather than incidental. An all-protocols # rule covers every port whatever its port fields say, and a rule whose # ports this policy cannot read is reported rather than passed. package terraform.network admin_ports := {22, 3389, 3306, 5432} public_cidrs := {"0.0.0.0/0", "::/0"} # The AWS provider writes from_port 0 and to_port 0 for an all-protocols # rule, which opens every port, so the port fields cannot be read literally. all_protocols := {"-1", "all"} # Shape 1 and 2: inline ingress blocks, IPv4 and IPv6. exposures contains exposure if { some resource in input.resource_changes resource.type == "aws_security_group" some ingress in resource.change.after.ingress some field in ["cidr_blocks", "ipv6_cidr_blocks"] some cidr in object.get(ingress, field, []) exposure := { "address": resource.address, "protocol": object.get(ingress, "protocol", ""), "from_port": object.get(ingress, "from_port", null), "to_port": object.get(ingress, "to_port", null), "cidr": cidr, } } # Shape 3: the standalone rule the AWS provider has recommended since v5. exposures contains exposure if { some resource in input.resource_changes resource.type == "aws_vpc_security_group_ingress_rule" some field in ["cidr_ipv4", "cidr_ipv6"] cidr := object.get(resource.change.after, field, null) is_string(cidr) exposure := { "address": resource.address, "protocol": object.get(resource.change.after, "ip_protocol", ""), "from_port": object.get(resource.change.after, "from_port", null), "to_port": object.get(resource.change.after, "to_port", null), "cidr": cidr, } } # Shape 4: the deprecated standalone rule, still in most existing estates. exposures contains exposure if { some resource in input.resource_changes resource.type == "aws_security_group_rule" resource.change.after.type == "ingress" some cidr in object.get(resource.change.after, "cidr_blocks", []) exposure := { "address": resource.address, "protocol": object.get(resource.change.after, "protocol", ""), "from_port": object.get(resource.change.after, "from_port", null), "to_port": object.get(resource.change.after, "to_port", null), "cidr": cidr, } } # The ports a rule really covers. Undefined when the policy cannot tell. covered_ports(exposure) := [0, 65535] if { exposure.protocol in all_protocols } covered_ports(exposure) := [exposure.from_port, exposure.to_port] if { not exposure.protocol in all_protocols is_number(exposure.from_port) is_number(exposure.to_port) } deny contains msg if { some exposure in exposures exposure.cidr in public_cidrs # A rule covers a port if that port falls inside [from_port, to_port]. range := covered_ports(exposure) some port in admin_ports port &amp;gt;= range[0] port &amp;lt;= range[1] msg := sprintf( "%s: ingress rule exposes port %d to %s", [exposure.address, port, exposure.cidr], ) } # A rule open to the world whose ports this policy cannot read is reported. # Passing it would be the policy guessing in the permissive direction. deny contains msg if { some exposure in exposures exposure.cidr in public_cidrs not covered_ports(exposure) msg := sprintf( "%s: ingress rule to %s has ports this policy cannot evaluate (%v to %v)", [exposure.address, exposure.cidr, exposure.from_port, exposure.to_port], ) } # Addresses this policy knows how to inspect. The gate uses this to tell # "nothing violated" apart from "nothing examined". considered contains resource.address if { some resource in input.resource_changes resource.type in { "aws_security_group", "aws_vpc_security_group_ingress_rule", "aws_security_group_rule", } } </code></pre> <p>Reading that from the top:</p> <ol> <li><p>A rule body in Rego is a conjunction. Every line must hold, and <code>some ... in</code> lines iterate, so OPA explores every combination of resource, ingress rule, field, and port.</p> </li> <li><p>Three separate <code>exposures</code> rules define one set between them, which Rego calls an incremental definition. Adding a fifth shape later costs one more block and changes nothing below it.</p> </li> <li><p><code>object.get(ingress, field, [])</code> returns an empty list when a field is absent, so an IPv4-only rule doesn't error when the policy looks for <code>ipv6_cidr_blocks</code>.</p> </li> <li><p><code>covered_ports</code> is the safety valve here, because an all-protocols rule reports <code>0</code> to <code>0</code> in the plan while opening every port, so the port fields can't be read literally, and a rule whose ports are null or strings leaves the function undefined, which the second <code>deny</code> rule turns into a violation.</p> </li> <li><p><code>considered</code> isn't part of the judgement. It records which resources this policy can speak about at all, which Step 5 uses to avoid reporting a pass it hasn't earned.</p> </li> </ol> <p>Run it:</p> <pre><code class="language-bash">opa test policy opa check --strict policy opa fmt --diff policy </code></pre> <pre><code class="language-plaintext">PASS: 20/20 </code></pre> <p>Add <code>-v</code> to <code>opa test</code> for a line per test.</p> <p><code>opa check --strict</code> catches unsafe variables and shadowed imports, while <code>opa fmt --diff</code> prints nothing when the formatting is already canonical. OPA formats Rego with tabs. Both belong in CI, ahead of everything else.</p> <p>The second policy is ownership tagging, so create <code>policy/tags.rego</code>:</p> <pre><code class="language-rego"># METADATA # title: Every managed resource carries ownership tags # description: | # Terraform emits `tags: null` for a resource with no tags at all, so a # policy that reaches into `after.tags` skips exactly the resources with # the worst tagging. `tags_of` coerces that null to an empty object. package terraform.tags required_tags := {"owner", "cost-center", "data-classification"} # Resource types that genuinely cannot carry tags. untaggable := {"aws_iam_policy_attachment", "aws_route_table_association"} in_scope contains resource if { some resource in input.resource_changes some action in resource.change.actions action in {"create", "update"} not resource.type in untaggable } tags_of(resource) := tags if { tags := resource.change.after.tags is_object(tags) } else := {} deny contains msg if { some resource in in_scope some tag in required_tags value := object.get(tags_of(resource), tag, "") trim_space(value) == "" msg := sprintf("%s: missing required tag %q", [resource.address, tag]) } considered contains resource.address if { some resource in in_scope } </code></pre> <p>Here's why those two lines look the way they do:</p> <ol> <li><p><code>trim_space(value) == ""</code> does the check. It has to, because in Rego only <code>false</code> and undefined are falsy, so an empty string is truthy. A bare existence check happily accepts <code>owner = ""</code>, which is compliance theatre of exactly the kind a tagging policy exists to stop.</p> </li> <li><p><code>tags_of</code>, with its <code>else := {}</code> branch, is the fix for a bug I wrote and only found in Step 9.</p> </li> </ol> <h2 id="heading-step-4-point-it-at-the-real-plan">Step 4: Point It at the Real Plan</h2> <p>Evaluate one package against the TerraGoat plan:</p> <pre><code class="language-bash">opa eval --data policy --input plan.json --format pretty 'data.terraform.network.deny' </code></pre> <pre><code class="language-plaintext">[ "aws_security_group.web-node: ingress rule exposes port 22 to 0.0.0.0/0" ] </code></pre> <p>The policy reports one violation, and stays quiet about port 80. It's open to the world in the same resource, because a public web server is the point of a public web server. A check that flags both is a check people learn to ignore.</p> <p>Running one command per package doesn't scale, and Rego can aggregate across a namespace in a single query:</p> <pre><code class="language-bash">opa eval --data policy --input plan.json --format pretty \ 'union({v | v := data.terraform[_].deny})' </code></pre> <pre><code class="language-plaintext">[ "aws_security_group.web-node: ingress rule exposes port 22 to 0.0.0.0/0", "aws_security_group.web-node: missing required tag \"cost-center\"", "aws_security_group.web-node: missing required tag \"data-classification\"", "aws_security_group.web-node: missing required tag \"owner\"", "aws_vpc.web_vpc: missing required tag \"cost-center\"", "aws_vpc.web_vpc: missing required tag \"data-classification\"", "aws_vpc.web_vpc: missing required tag \"owner\"" ] </code></pre> <p><code>{v | v := data.terraform[_].deny}</code> is a comprehension that collects the <code>deny</code> set from every package under <code>data.terraform</code>, and <code>union</code> flattens them. Drop a new policy file into that namespace and it's picked up with no change to the command.</p> <h2 id="heading-step-5-turn-the-verdict-into-an-exit-code">Step 5: Turn the Verdict into an Exit Code</h2> <p>A CI gate communicates through its exit status, and conflating two kinds of failure into one code is how a broken pipeline passes for a month. This tool uses the following:</p> <table> <thead> <tr> <th>Code</th> <th>Verdict</th> <th>Meaning</th> </tr> </thead> <tbody><tr> <td>0</td> <td>pass</td> <td>policies ran, examined resources, found nothing</td> </tr> <tr> <td>1</td> <td>fail</td> <td>policies ran and found violations</td> </tr> <tr> <td>2</td> <td>vacuous</td> <td>policies ran but examined nothing, so the result means nothing</td> </tr> <tr> <td>2</td> <td>broken</td> <td>the tool or its input is unusable</td> </tr> </tbody></table> <p>The fourth row is the one most gates get wrong. If a plan contains no resource any policy knows about, reporting a pass claims an assurance the run can't give. It gets its own verdict name and it doesn't exit 0.</p> <p>Create <code>policy_gate.py</code>:</p> <pre><code class="language-python">"""Evaluate a Terraform plan against a directory of Rego policies.""" import argparse import json import pathlib import shutil import subprocess import sys PASS, FAIL, BROKEN = 0, 1, 2 DENY_QUERY = "union({v | v := data.terraform[_].deny})" CONSIDERED_QUERY = "union({v | v := data.terraform[_].considered})" def die(message: str) -&amp;gt; None: print(f"policy-gate: {message}", file=sys.stderr) sys.exit(BROKEN) def load_plan(path: pathlib.Path) -&amp;gt; dict: try: text = path.read_text() except OSError as exc: die(f"cannot read {path}: {exc.strerror}") try: return json.loads(text) except json.JSONDecodeError as exc: die(f"{path}:{exc.lineno}:{exc.colno}: invalid JSON: {exc.msg}") def query(opa: str, policy_dirs: list[pathlib.Path], plan: pathlib.Path, expr: str) -&amp;gt; list: command = [opa, "eval", "--input", str(plan), "--format", "raw"] for directory in policy_dirs: command += ["--data", str(directory)] command.append(expr) result = subprocess.run(command, capture_output=True, text=True) if result.returncode != 0: die(f"opa failed: {result.stderr.strip() or result.stdout.strip()}") values = json.loads(result.stdout) # A rule that yields anything but strings is a policy bug, and sorting a # mixed list would surface it as an unrelated TypeError. for value in values: if not isinstance(value, str): die(f"{expr} produced a {type(value).__name__}; " "deny and considered rules must yield strings") return values def main() -&amp;gt; int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--plan", required=True, type=pathlib.Path, help="JSON from `terraform show -json`") parser.add_argument("--policy", required=True, nargs="+", type=pathlib.Path, help="one or more directories of .rego files") parser.add_argument("--opa", default="opa", help="path to the opa binary") args = parser.parse_args() if shutil.which(args.opa) is None: die(f"{args.opa} is not on PATH") for directory in args.policy: if not directory.is_dir(): die(f"{directory} is not a directory") plan = load_plan(args.plan) if "resource_changes" not in plan: die(f"{args.plan} has no resource_changes key; is it a Terraform plan?") considered = query(args.opa, args.policy, args.plan, CONSIDERED_QUERY) if not considered: # Reporting a pass here would claim an assurance the run cannot give. print(f"VACUOUS: no policy examined any of the " f"{len(plan['resource_changes'])} planned resource(s)", file=sys.stderr) return BROKEN violations = sorted(query(args.opa, args.policy, args.plan, DENY_QUERY)) if violations: print(f"FAIL: {len(violations)} violation(s) " f"across {len(considered)} examined resource(s)", file=sys.stderr) for violation in violations: print(f" - {violation}", file=sys.stderr) return FAIL print(f"PASS: {len(considered)} resource(s) examined, no violations") return PASS if __name__ == "__main__": sys.exit(main()) </code></pre> <p>Here's what the script does:</p> <ol> <li><p><code>argparse</code> marks <code>--plan</code> and <code>--policy</code> as <code>required=True</code>, and <code>--policy</code> takes <code>nargs="+"</code>, so an empty policy list raises an error at parse time.</p> </li> <li><p><code>load_plan</code> reports the line and column of a JSON syntax error, because <code>json.JSONDecodeError</code> carries <code>lineno</code> and <code>colno</code> and a gate that says only "invalid JSON" wastes somebody's afternoon.</p> </li> <li><p>Every failure path routes through <code>die</code>, which always exits 2, because a missing <code>opa</code>, an unreadable file, and a plan with no <code>resource_changes</code> key are all failures of the tool itself.</p> </li> <li><p>The <code>considered</code> query runs <em>before</em> the <code>deny</code> query. If nothing was examined, the run ends at <code>VACUOUS</code> and never gets the chance to print a pass.</p> </li> <li><p>Violations are sorted, so the same plan produces byte-identical output on every run and a diff of two CI logs means something.</p> </li> </ol> <img src="https://cdn.hashnode.com/uploads/covers/5f3a74bfc4d5973f55c91c8c/8d6be1c9-d9ca-4379-8773-3e083cbe9475.png" alt="Terminal window titled policy-lab. Running opa test policy reports PASS colon 20 slash 20. Running python3 policy_gate.py with the TerraGoat plan prints FAIL colon 7 violations across 2 examined resources, listing one ingress rule exposing port 22 to 0.0.0.0/0 on aws_security_group.web-node and six missing required tags across aws_security_group.web-node and aws_vpc.web_vpc. echo dollar question mark returns 1." style="display: block;" width="600" height="400" loading="lazy"> <p>Seven violations across the two resources in this plan, and an exit code CI can act on.</p> <img src="https://cdn.hashnode.com/uploads/covers/5f3a74bfc4d5973f55c91c8c/d0d457fc-a7f4-4f7f-8a13-c0bc352f9ed6.png" alt="Terminal window titled policy-lab. Running python3 policy_gate.py against an empty plan prints VACUOUS colon no policy examined any of the 0 planned resources, and echo dollar question mark returns 2, not 0." style="display: block;" width="600" height="400" loading="lazy"> <p>The same tool on an empty plan, where a gate answering "pass" would be lying.</p> <p>The GitHub Actions workflow tests the policies before it uses them to judge anything:</p> <pre><code class="language-yaml">name: policy on: [pull_request] jobs: policy: runs-on: ubuntu-latest steps: - uses: actions/checkout@v5 - name: Install OPA run: | curl -L -o /usr/local/bin/opa \ https://openpolicyagent.org/downloads/v1.20.2/opa_linux_amd64_static chmod +x /usr/local/bin/opa # The policies are code. Lint and test them before trusting them. - name: Check policy syntax run: opa check --strict policy - name: Verify formatting run: opa fmt --fail --diff policy - name: Test policies run: opa test policy --verbose --coverage --format json &amp;gt; coverage.json # Only now does anything get judged. - name: Evaluate Terraform plan run: python3 policy_gate.py --plan plan.json --policy policy </code></pre> <p>Roll this out with the gate reporting only, for a fortnight, before you let it block. A policy that looks obviously correct will fail on something structural in your real estate, and you would rather find that out from a log line than from a blocked release.</p> <h2 id="heading-step-6-enforce-at-admission-time">Step 6: Enforce at Admission Time</h2> <p>The gate in Step 5 checks what you intended to deploy. It doesn't see a <code>kubectl apply</code> from somebody's laptop, a vendor's Helm chart, or an operator creating Pods on its own schedule. For those, you need admission control, and Kubernetes now has it built in.</p> <p><code>ValidatingAdmissionPolicy</code> has been generally available since <strong>v1.30</strong>, evaluating CEL inside the API server with no webhook to deploy or keep alive:</p> <pre><code class="language-yaml">apiVersion: admissionregistration.k8s.io/v1 kind: ValidatingAdmissionPolicy metadata: name: require-trusted-registry spec: failurePolicy: Fail matchConstraints: resourceRules: - apiGroups: [""] apiVersions: ["v1"] operations: ["CREATE", "UPDATE"] resources: ["pods"] variables: # A Pod has three container lists. A policy that reads only # spec.containers is bypassed by moving the image to an initContainer. - name: allImages expression: &amp;gt;- object.spec.containers.map(c, c.image) + object.spec.?initContainers.orValue([]).map(c, c.image) + object.spec.?ephemeralContainers.orValue([]).map(c, c.image) validations: - expression: &amp;gt;- variables.allImages.all(i, i.startsWith('registry.internal.example.com/')) messageExpression: &amp;gt;- 'images must come from registry.internal.example.com: ' + variables.allImages.filter(i, !i.startsWith('registry.internal.example.com/')).join(', ') reason: Forbidden </code></pre> <p>The policy does nothing until a binding activates it, which is what lets you pilot on one namespace:</p> <pre><code class="language-yaml">apiVersion: admissionregistration.k8s.io/v1 kind: ValidatingAdmissionPolicyBinding metadata: name: require-trusted-registry-binding spec: policyName: require-trusted-registry validationActions: ["Deny"] matchResources: namespaceSelector: matchLabels: policy.example.com/enforce: "true" </code></pre> <p>Set <code>validationActions: ["Warn", "Audit"]</code>, label one namespace, watch for a week, and then switch to <code>["Deny"]</code> and widen the selector.</p> <p>The <code>initContainers</code> handling matters, because it's the most common way an image-provenance policy gets bypassed, and the <code>?</code> optional-field syntax with <code>.orValue([])</code> is how you read a list that may be absent without the whole expression erroring.</p> <p>I couldn't apply these two manifests, because I had no cluster to hand. They're checked against the v1 reference schema, and no real API server has admitted them, so treat them as a starting point and roll them out in <code>Warn</code> mode (which you should be doing anyway).</p> <p>Two other engines are in wide production use, starting with <a href="https://kyverno.io/">Kyverno</a>. It <strong>graduated in the CNCF in March 2026</strong> with production use at Bloomberg, Coinbase, Deutsche Telekom, LinkedIn, and Spotify. Its policies are written in YAML, so a platform team needs no new language, and it handles generation, image-signature verification, and cleanup that built-in policies leave alone.</p> <p><a href="https://open-policy-agent.github.io/gatekeeper/">OPA Gatekeeper</a> is the right answer when you want one Rego codebase covering Kubernetes <em>and</em> Terraform <em>and</em> CI, which is the position this tutorial builds toward. Mutation is now built in too: <code>MutatingAdmissionPolicy</code> became stable in <strong>v1.36</strong>.</p> <h2 id="heading-step-7-let-a-model-write-the-policy">Step 7: Let a Model Write the Policy</h2> <p>Policies are tedious, and models are good at tedious. So the obvious move is to have the model write them.</p> <p>There's a catch that you can measure yourself in about a minute, and I do exactly that at the end of this step: a great deal of the Rego in public training data is <strong>Rego v0</strong>, the dialect that stopped parsing when OPA 1.0 shipped in January 2025. A model reaching for the most common pattern it has seen reaches for a dialect the current parser rejects.</p> <p>A 2025 preprint from a group at the University of Calabria, <a href="https://arxiv.org/abs/2507.10584"><em>ARPaCCino</em></a>, reports the same effect on a Terraform case study: asked for Rego with no tools, Qwen3-30B and GPT-4o each produced 0 of 5 syntactically correct policies. Adding retrieval over the OPA documentation changed nothing. Giving the model a loop that could run <code>opa check</code> and read the errors took those to 4 of 5 and 5 of 5.</p> <p>Those counts come from one small case study, so treat the direction as the durable part of the result.</p> <p>A feedback loop is what fixed it, and <strong>the loop costs nothing</strong>, because you already built it out of <code>opa check --strict</code>, <code>opa fmt</code>, and <code>opa test</code>.</p> <p>So build the loop with one inversion that makes it trustworthy. <strong>I write the tests, and the model writes the policy.</strong> Test fixtures are concrete and cheap to review, since you read a JSON blob and say "yes, that should be rejected" in three seconds. Rego with nested comprehensions takes real effort to read and is easy to misread. Put the human where review is cheap, and let the machine work where its output can be checked mechanically.</p> <p>Create <code>policy_forge.py</code>:</p> <pre><code class="language-python">"""Generate a Rego policy from a rule in English, and keep it only if the toolchain agrees.""" import argparse import pathlib import re import subprocess import sys import tempfile WRITTEN, REJECTED, BROKEN = 0, 1, 2 SYSTEM = """You write Open Policy Agent policies in Rego v1 (OPA 1.0+). Rules: - Use `if` on every rule body and `contains` for multi-value rules. - Do not emit `import rego.v1`; it is redundant on OPA 1.0+. - The input is the JSON from `terraform show -json`. - Return one ```rego block and nothing else.""" def extract_rego(reply: str) -&amp;gt; str: blocks = re.findall(r"```rego\n(.*?)```", reply, re.DOTALL) if not blocks: raise ValueError("model returned no rego block") if len(blocks) &amp;gt; 1: raise ValueError(f"model returned {len(blocks)} rego blocks; expected one") return blocks[0] def verify(opa: str, policy: str, tests: pathlib.Path) -&amp;gt; tuple[bool, str]: with tempfile.TemporaryDirectory() as tmp: bundle = pathlib.Path(tmp) # The tests keep their own name; the policy gets one that cannot # collide with it, whatever the caller named the test file. (bundle / "candidate_policy.rego").write_text(policy) (bundle / tests.name).write_text(tests.read_text()) for command in ([opa, "check", "--strict"], [opa, "test"]): result = subprocess.run(command + [str(bundle)], capture_output=True, text=True) if result.returncode != 0: return False, (result.stdout + result.stderr).strip() return True, "opa check and opa test both passed" def forge(rule: str, tests: pathlib.Path, ask, opa: str, attempts: int) -&amp;gt; str: transcript = [{ "role": "user", "content": ( f"Write a Rego policy for this rule:\n\n{rule}\n\n" f"It must satisfy these tests:\n\n```rego\n{tests.read_text()}```" ), }] for attempt in range(1, attempts + 1): reply = ask(transcript) policy = extract_rego(reply) ok, output = verify(opa, policy, tests) headline = next(iter(output.splitlines()), "no output from the toolchain") print(f"attempt {attempt}: {'PASS' if ok else 'FAIL'} - {headline}", file=sys.stderr) if ok: return policy transcript += [ {"role": "assistant", "content": reply}, {"role": "user", "content": f"The toolchain rejected that:\n\n{output}\n\nFix it."}, ] raise RuntimeError(f"no policy survived {attempts} attempts") def claude(model: str): import anthropic client = anthropic.Anthropic() def ask(transcript: list[dict]) -&amp;gt; str: response = client.messages.create( model=model, max_tokens=16000, system=[{ "type": "text", "text": SYSTEM, "cache_control": {"type": "ephemeral"}, }], thinking={"type": "adaptive"}, messages=transcript, ) return "".join(b.text for b in response.content if b.type == "text") return ask def main() -&amp;gt; int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--rule", required=True, help="the policy, in one English sentence") parser.add_argument("--tests", required=True, type=pathlib.Path, help="a _test.rego file you wrote by hand") parser.add_argument("--out", required=True, type=pathlib.Path, help="where to write the policy, only if it passes") parser.add_argument("--model", default="claude-opus-5") parser.add_argument("--attempts", type=int, default=4) parser.add_argument("--opa", default="opa") args = parser.parse_args() if args.attempts &amp;lt; 1: print("policy-forge: --attempts must be at least 1", file=sys.stderr) return BROKEN if not args.tests.is_file(): print(f"policy-forge: {args.tests} does not exist", file=sys.stderr) return BROKEN try: policy = forge(args.rule, args.tests, claude(args.model), args.opa, args.attempts) except RuntimeError as exc: print(f"policy-forge: {exc}; nothing written", file=sys.stderr) return REJECTED except ValueError as exc: print(f"policy-forge: {exc}", file=sys.stderr) return BROKEN args.out.write_text(policy) print(f"policy-forge: verified policy written to {args.out}", file=sys.stderr) return WRITTEN if __name__ == "__main__": sys.exit(main()) </code></pre> <pre><code class="language-bash">export ANTHROPIC_API_KEY=... python3 policy_forge.py \ --rule "No security group may expose an administrative port to the public internet." \ --tests policy/network_test.rego \ --out policy/generated.rego </code></pre> <p>Here's what the loop guarantees:</p> <ol> <li><p><strong>Verification runs as a subprocess:</strong> the model is never asked whether its policy is correct. <code>opa check</code> and <code>opa test</code> decide, and their exit codes are the only evidence the loop accepts.</p> </li> <li><p><strong>Failures go back as raw tool output, never summarised:</strong> compiler errors and test failures are the highest-signal feedback a model can receive, and paraphrasing throws away the part that helps.</p> </li> <li><p><strong>Nothing reaches disk until it passes:</strong> <code>forge</code> either returns a verified policy or raises, so there's no path where an unverified policy lands in the repository just because the retry budget ran out.</p> </li> </ol> <p>I drove the loop with a scripted model so the result is reproducible without an API key. The three replies were a realistic v0-syntax policy, a realistic-but-wrong v1 policy, and the policy from Step 3:</p> <pre><code class="language-plaintext">attempt 1: FAIL - 2 errors occurred during loading: attempt 2: FAIL - policy/network_test.rego:63: attempt 3: PASS - opa check and opa test both passed </code></pre> <p>Attempt 1 was Rego v0, the <code>deny[msg] { ... }</code> form, which stopped parsing when OPA 1.0 shipped in January 2025. And it's overwhelmingly what public training data contains. <code>opa check --strict</code> rejected it before it reached a test.</p> <p>Attempt 2 was valid Rego v1. It would have passed review from most engineers, and it still scored only <strong>4 of 8</strong> on the suite. This is because it compared <code>ingress.from_port</code> against <code>admin_ports</code> directly, ignored the range, and read only <code>cidr_blocks</code>. <code>opa check</code> had no complaint, because the code was perfectly well-formed.</p> <p><strong>A well-formed policy can still be the wrong policy, and the only thing in this loop that knows what you wanted is the test suite you wrote.</strong></p> <p>A repair loop isn't monotonic, because each attempt is a fresh generation conditioned on an error message, with nothing carrying forward what already worked, so attempt four can lose a property attempt three had. Nothing in this design detects that, because the only thing being checked is the test suite you wrote.</p> <p>Cap the retries, keep the suite growing, and treat every generated policy as a pull request that somebody approves before it merges.</p> <h2 id="heading-step-8-govern-the-agent-itself">Step 8: Govern the Agent Itself</h2> <p>An AI agent is also an actor, and it calls tools, so every tool call becomes an authorisation decision that something has to make.</p> <p>The industry converged on this quickly: Amazon Bedrock AgentCore Policy reached general availability in March 2026, evaluating agent tool calls at the gateway in <a href="https://www.cedarpolicy.com/">Cedar</a>. The common open-source pattern is an OPA sidecar in front of an MCP tool gateway.</p> <p>Research is pushing the same boundary harder: a 2026 preprint from the University of Washington group behind Defects4J, <a href="https://arxiv.org/abs/2603.20449"><em>Solver-Aided Verification of Policy Compliance in Tool-Augmented LLM Agents</em></a> (Winston, Winston, and Just), compiles natural-language policies into SMT constraints and blocks non-compliant calls with the Z3 solver.</p> <p>They share one claim: <strong>a policy in the system prompt isn't enforcement.</strong> Enforcement is an interceptor sitting in the call path that can return "no" and stop the call from happening.</p> <p>Create <code>agent/authz.rego</code>:</p> <pre><code class="language-rego"># METADATA # title: Agent tool-call authorisation # description: | # Evaluated once per tool call, before the tool runs. The decision has # three values rather than two, because an agent worth deploying will # sometimes need to do something that a human, not the policy, should # approve. package agent.authz tool_grants := { "support": {"search_orders", "read_customer", "issue_refund"}, "analytics": {"search_orders", "run_query"}, } write_tools := {"issue_refund", "run_query"} refund_ceiling_cents := 10000 # An unmapped role, an unknown tool or a malformed input all land here. default decision := {"effect": "deny", "reasons": ["no matching grant"]} decision := {"effect": effect_for(reasons), "reasons": reasons} if { count(granted) &amp;gt; 0 reasons := escalations } granted contains role if { some role in input.agent.roles input.tool in object.get(tool_grants, role, set()) } effect_for(reasons) := "allow" if count(reasons) == 0 effect_for(reasons) := "require_approval" if count(reasons) &amp;gt; 0 escalations contains reason if { input.tool in write_tools not input.session.human_in_loop reason := sprintf("%q writes state and the session is unattended", [input.tool]) } # A refund with no readable amount cannot be checked against the ceiling, # so it escalates. Silence here would clear the exact call an attacker # would craft. escalations contains reason if { input.tool == "issue_refund" not positive_amount reason := "refund amount is missing, unreadable, or not positive" } # A negative amount is a charge wearing a refund's name. positive_amount if { amount := object.get(input, ["arguments", "amount_cents"], null) is_number(amount) amount &amp;gt; 0 } escalations contains reason if { input.tool == "issue_refund" amount := object.get(input, ["arguments", "amount_cents"], null) is_number(amount) amount &amp;gt; refund_ceiling_cents reason := sprintf( "refund of %d cents exceeds the %d cent ceiling", [amount, refund_ceiling_cents], ) } # Keyword matching is a coarse guard, and it is here to show the shape of an # argument-level rule. Anything holding real data wants a SQL parser: this # catches `DROP TABLE` and misses a statement that spells it another way. destructive_sql := `(?i)\b(drop|truncate|delete|alter|grant|revoke)\b` escalations contains reason if { input.tool == "run_query" regex.match(destructive_sql, object.get(input, ["arguments", "statement"], "")) reason := "statement contains a destructive SQL keyword" } </code></pre> <p>The decision vocabulary is closed, and I'll state it plainly here:</p> <table> <thead> <tr> <th>Effect</th> <th>What the caller does</th> </tr> </thead> <tbody><tr> <td>allow</td> <td>run the tool</td> </tr> <tr> <td>require_approval</td> <td>pause, show the reasons to a human, run only on approval</td> </tr> <tr> <td>deny</td> <td>refuse, and don't offer an approval path</td> </tr> </tbody></table> <p>Here's why the policy is shaped that way:</p> <ol> <li><p><code>default decision</code> <strong>is deny:</strong> an unrecognised tool, a role you forgot to map, or a malformed input all end up there. A policy that defaults to allow fails open on exactly the inputs nobody anticipated, which is the set an attacker picks from.</p> </li> <li><p><strong>Three values:</strong> binary authorisation forces a choice between blocking useful work and permitting dangerous work, and the third value is what makes a high-autonomy agent tolerable.</p> </li> <li><p><strong>Reasons come back as a set:</strong> every applicable reason is collected. When somebody gets an approval prompt at three in the morning, "refund of 250000 cents exceeds the 10000 cent ceiling" tells them what to do. "Policy violation" does not.</p> </li> <li><p><strong>Arguments are inspected too:</strong> <code>issue_refund</code> is routine at £5 and serious at £2,500, so tool-name granularity is far too coarse for agents, given that the agent chooses the arguments.</p> </li> </ol> <p>Fifteen tests cover the decision table, including an agent with an empty role list, a refund with no amount at all, and a query that hides <code>DROP</code> behind a newline:</p> <pre><code class="language-bash">opa test agent -v </code></pre> <pre><code class="language-plaintext">PASS: 15/15 </code></pre> <p>Serve it and try a call:</p> <pre><code class="language-bash">opa run --server --addr localhost:8181 agent/ </code></pre> <pre><code class="language-bash">curl -s localhost:8181/v1/data/agent/authz/decision \ -d '{"input":{"agent":{"roles":["support"]},"tool":"issue_refund", "arguments":{"amount_cents":250000},"session":{"human_in_loop":true}}}' | jq .result </code></pre> <pre><code class="language-json">{ "effect": "require_approval", "reasons": [ "refund of 250000 cents exceeds the 10000 cent ceiling" ] } </code></pre> <p>The policy is inert until something refuses to proceed on its answer. That's the client:</p> <pre><code class="language-python">import json import urllib.request OPA_URL = "http://localhost:8181/v1/data/agent/authz/decision" class PolicyDenied(Exception): pass class ApprovalRequired(Exception): pass def authorize(agent, tool, arguments, session): payload = json.dumps({"input": { "agent": agent, "tool": tool, "arguments": arguments, "session": session, }}).encode() req = urllib.request.Request( OPA_URL, data=payload, headers={"Content-Type": "application/json"} ) with urllib.request.urlopen(req, timeout=2) as resp: body = json.load(resp) # OPA returns {} with a 200 when a query matches nothing. Fail closed. decision = body.get("result", {"effect": "deny", "reasons": ["policy unavailable"]}) if decision["effect"] == "deny": raise PolicyDenied("; ".join(decision["reasons"])) if decision["effect"] == "require_approval": raise ApprovalRequired("; ".join(decision["reasons"])) return decision </code></pre> <pre><code class="language-plaintext">search_orders -&amp;gt; ALLOWED issue_refund -&amp;gt; NEEDS APPROVAL (refund of 250000 cents exceeds the 10000 cent ceiling) delete_account -&amp;gt; DENIED (no matching grant) </code></pre> <p>Note <code>body.get("result", ...)</code>: OPA returns <code>{}</code> with a 200 status when a query matches nothing, so a bare <code>body["result"]</code> raises <code>KeyError</code>, and depending on how your agent framework handles exceptions that may fail <em>open</em>. Every layer defaults to deny, including the parsing.</p> <p>Call <code>authorize()</code> from your framework's tool-execution hook, before the tool function runs. It's about fifteen lines, and it turns a system prompt's polite suggestions into an actual boundary.</p> <h2 id="heading-step-9-what-i-got-wrong">Step 9: What I Got Wrong</h2> <p>Steps 2 and 3 show the finished policies. I reached for something simpler first (the version most tutorials stop at), and the gap between that and what you have just read is the most useful thing here.</p> <h3 id="heading-the-tagging-policy-skipped-the-worst-resources">The Tagging Policy Skipped the Worst Resources</h3> <p>My naïve <code>in_scope</code> rule ended with <code>resource.change.after.tags</code>, which reads as "only resources that have tags".</p> <p>What it actually does is worse than that, because Terraform emits <code>tags: null</code> for a resource with <strong>no tags at all</strong>, and an undefined lookup makes the rule body fail, so the resource drops out of scope entirely.</p> <p>The TerraGoat plan has two resources: the security group carries five <code>git_*</code> tags and no ownership tags, while the VPC carries nothing at all.</p> <pre><code class="language-bash">jq -r '.resource_changes[] | "\(.address): tags=\(.change.after.tags | type)"' plan.json </code></pre> <pre><code class="language-plaintext">aws_security_group.web-node: tags=object aws_vpc.web_vpc: tags=null </code></pre> <pre><code class="language-bash">opa eval --data naive --input plan.json --format pretty 'count(data.terraform.tags.deny)' opa eval --data policy --input plan.json --format pretty 'count(data.terraform.tags.deny)' </code></pre> <pre><code class="language-plaintext">3 6 </code></pre> <p>The three it missed were all on the completely untagged resource, so the policy flagged the resource with some tags and silently cleared the one with none.</p> <p><code>tags_of</code> with its <code>else := {}</code> branch is the fix, and it's three lines.</p> <h3 id="heading-the-network-policy-read-one-of-four-shapes">The Network Policy Read One of Four Shapes</h3> <p>My naïve network policy read <code>aws_security_group</code> and <code>cidr_blocks</code>, which is what every tutorial shows, but Terraform has four ways to express the same ingress rule.</p> <p>![Diagram titled "Four ways Terraform describes one ingress rule". Four boxes are shown. Top left, aws_security_group.ingress[].cidr_blocks, filled pale blue and labelled read. The other three are outlined in red with red hatching and labelled not read: aws_security_group.ingress[].ipv6_cidr_blocks, aws_vpc_security_group_ingress_rule.cidr_ipv4, and the deprecated aws_security_group_rule. A key states that solid blue fill means the policy looks here and red hatch means it does not.](<a href="https://cdn.hashnode.com/uploads/covers/5f3a74bfc4d5973f55c91c8c/ab61d36d-78bf-45f7-a073-e7013216e3e1.png">https://cdn.hashnode.com/uploads/covers/5f3a74bfc4d5973f55c91c8c/ab61d36d-78bf-45f7-a073-e7013216e3e1.png</a> align="center")</p> <p>One rule, four encodings, and the naïve policy read only the top-left one.</p> <p>I planned a second real configuration with three security groups that open SSH to the world using the shapes the policy didn't read. Both versions are in <code>code/</code>, so this reproduces:</p> <pre><code class="language-bash">opa eval --data naive --input plan-evasion.json \ --format pretty 'data.terraform.network.deny' </code></pre> <pre><code class="language-plaintext">[] </code></pre> <p>That is three publicly reachable SSH ports, zero violations, and a gate that would have printed <code>PASS</code> and exited 0.</p> <h3 id="heading-a-test-that-was-holding-a-hole-open">A Test That Was Holding a Hole Open</h3> <p>The port range had a second problem, and my own test suite was protecting it. I had written a case called <code>test_tolerates_null_ports</code>, asserting that an ingress rule with <code>protocol: "-1"</code> produced no violations, on the reasoning that a comparison against a null port shouldn't crash the policy.</p> <p>An all-protocols rule opens every port, and the AWS provider records it as <code>from_port: 0</code> and <code>to_port: 0</code>. The the range check read it literally as the single port zero, so the most permissive rule in AWS scored clean.</p> <p>The plan is in <code>code/plan-all-protocols.json</code>, and it's one resource:</p> <pre><code class="language-bash">jq -c '.resource_changes[] | select(.type=="aws_security_group") | .change.after.ingress[0] | {protocol,from_port,to_port,cidr_blocks}' \ plan-all-protocols.json opa eval --data naive --input plan-all-protocols.json \ --format pretty 'data.terraform.network.deny' </code></pre> <pre><code class="language-plaintext">{"protocol":"-1","from_port":0,"to_port":0,"cidr_blocks":["0.0.0.0/0"]} [] </code></pre> <p>Every protocol and every port, open to the whole internet, and a test I wrote on purpose certified it as fine. The fix is <code>covered_ports</code>, which maps an all-protocols rule onto the full range and goes undefined for ports it can't read, with a second <code>deny</code> rule that reports the undefined case. That test is gone and four took its place:</p> <pre><code class="language-plaintext">opa eval --data policy --input plan-all-protocols.json --format pretty 'data.terraform.network.deny' </code></pre> <pre><code class="language-plaintext">[ "aws_security_group.wide_open: ingress rule exposes port 22 to 0.0.0.0/0", "aws_security_group.wide_open: ingress rule exposes port 3306 to 0.0.0.0/0", "aws_security_group.wide_open: ingress rule exposes port 3389 to 0.0.0.0/0", "aws_security_group.wide_open: ingress rule exposes port 5432 to 0.0.0.0/0" ] </code></pre> <p>Step 7 argues that the test suite is the only artifact in the loop that knows what you wanted. This is the cost of that property: a test that's wrong is a specification that's wrong, and nothing downstream of it will argue.</p> <h3 id="heading-what-the-numbers-actually-were">What the Numbers Actually Were</h3> <img src="https://cdn.hashnode.com/uploads/covers/5f3a74bfc4d5973f55c91c8c/ce90751d-1eb9-45d1-8514-6481d9b3612d.png" alt="orizontal bar chart titled &quot;The naive policy missed 5 of the 9 violations present&quot;, showing violations found as a fraction of violations present on two real Terraform plans. For evasion plan open ports the naive policy scores 0.000 in red hatching and the hardened policy scores 1.000 in blue hatching. For terragoat plan missing tags the naive policy scores 0.500 in red and the hardened policy scores 1.000 in blue. For terragoat plan open ports both score 1.000, drawn in grey." style="display: block;" width="600" height="400" loading="lazy"> <p>Both versions score 1.000 on the plan I designed the policy against. The gap only appears on the plan I did not.</p> <p>Across both real plans, <strong>the naïve policies found 4 of the 9 violations present</strong>. They scored 1.000 on the TerraGoat security group, which is the case I had in mind while writing them, and 0.000 and 0.500 on the two cases I did not.</p> <h3 id="heading-the-part-that-genuinely-surprised-me">The Part That Genuinely Surprised Me</h3> <p>I assumed test coverage would have caught this, and it doesn't. I reconstructed the naïve tagging policy with the two tests I originally wrote for it:</p> <pre><code class="language-bash">opa test . --coverage --format json | jq '{overall: .coverage}' </code></pre> <pre><code class="language-plaintext">{ "overall": 100 } </code></pre> <p><strong>The naïve policy scored 100% coverage, passed 2 of 2 tests, and cleared a resource that carried no tags at all.</strong></p> <p>Coverage measures which lines of a policy your tests executed, and says nothing at all about which shapes of input you failed to imagine. For policy code, this is the entire failure mode. Coverage is worth reporting, and the evidence that a policy actually works comes from running it against infrastructure you didn't write.</p> <h2 id="heading-limits-of-the-check">Limits of the Check</h2> <p>Here's what this gate still can't do, because its false clearances matter more than its catches.</p> <h3 id="heading-1-unknown-values-are-invisible">1. Unknown Values Are Invisible</h3> <p>Terraform marks anything it can't resolve until apply time as unknown, which appears as <code>null</code> in the plan JSON alongside an <code>after_unknown</code> map. A policy reading <code>change.after.some_field</code> doesn't fire when that field is unknown.</p> <p>This bites hardest on cross-resource rules, where "every bucket has a public access block" is genuinely hard at plan time, because the block references a bucket ID that's usually <code>(known after apply)</code>.</p> <h3 id="heading-2-it-sees-the-plan-and-only-the-plan">2. It Sees the Plan, and Only the Plan</h3> <p>Anything applied outside the pipeline, changed in a console, or drifted since creation stays invisible to it. Plan-time checks and admission-time checks have <em>different</em> blind spots, and both leave work for a periodic scan of deployed state.</p> <h3 id="heading-3-coverage-of-controls-isnt-measurable-from-inside">3. Coverage of Controls Isn't Measurable from Inside</h3> <p>A hundred green checks say nothing about the rules nobody wrote. Keep the mapping from your control requirements to your policy files somewhere explicit and audit it on a schedule, because the gate can't tell you what it was never asked.</p> <h3 id="heading-4-the-cidr-list-is-an-exact-match">4. The CIDR List is an Exact Match</h3> <p><code>public_cidrs</code> holds <code>0.0.0.0/0</code> and <code>::/0</code> and nothing else, so a rule opening <code>0.0.0.0/1</code> reaches half the internet and passes. Widening it means deciding which prefix lengths count as public and carving out RFC 1918 space, and that decision belongs to your organisation.</p> <h3 id="heading-5-four-shapes-is-what-i-found">5. Four Shapes is What I Found</h3> <p>The <code>exposures</code> set covers the four encodings I went looking for, and AWS offers more. Security group references (<code>security_groups</code>), prefix lists, and <code>self</code> rules are all ways to reach a port that this policy doesn't model, and it never sees them. I would expect a fifth shape to turn up the first time this runs against a large estate.</p> <h3 id="heading-6-regulatory-dates-move">6. Regulatory Dates Move</h3> <p>If you're building toward the EU AI Act, the Digital Omnibus published in July 2026 pushed Annex III high-risk obligations from 2 August 2026 to <strong>2 December 2027</strong>, and Annex I obligations to 2 August 2028, while the Article 50 transparency duties kept their original 2 August 2026 date.</p> <p>Encode the controls, and look the dates up in the <a href="https://artificialintelligenceact.eu/implementation-timeline/">official timeline</a> every time you need one, including when a blog post from last quarter tells you otherwise (this one included).</p> <h2 id="heading-conclusion">Conclusion</h2> <p>Models produce infrastructure code that parses almost every time, is secure about 56% of the time, and arrives faster than anybody can read it. Manual review stopped being a real control somewhere in that gap. The rules were always meant to be executable, and the volume is what finally forced the issue.</p> <p>In this tutorial, you:</p> <ul> <li><p>Pulled a real vulnerable security group from TerraGoat at commit <code>729f8da</code> and planned it with Terraform 1.14.2.</p> </li> <li><p>Wrote 20 policy tests covering four Terraform encodings of one ingress rule, and got them to pass on OPA 1.20.2.</p> </li> <li><p>Caught 7 real violations in the TerraGoat plan, while correctly ignoring port 80.</p> </li> <li><p>Built a gate with a four-verdict contract that exits 2 when it examined nothing.</p> </li> <li><p>Watched the naïve version find 4 of 9 violations while reporting 100% test coverage, and hardened it to find 9 of 9.</p> </li> <li><p>Wired a generation loop where <code>opa check</code> and your own tests decide what reaches disk.</p> </li> <li><p>Authorised agent tool calls from the same engine, with 15 tests and a default of deny.</p> </li> </ul> <p>My policies handled the cases I wrote them for and missed two I hadn't imagined, and every signal available to me (tests passing, coverage at 100%, and a clean <code>opa check</code>) agreed they were fine. The one thing that disagreed was infrastructure somebody else had written. Point your policies at code you didn't write, early, and keep the failures.</p> <p>All the code, the policies, the plan JSON, and the scripts that build the figures are in the <code>code/</code> directory alongside this handbook. The figures regenerate with <code>python3 build/make_images.py</code> and <code>python3 build/make_terminals.py</code>. The terminal screenshots re-run their commands at build time, so they can't drift from the truth.</p>