如何使用 Policy as Code 和 OPA 治理 AI 生成的基础设施[完整手册]

现代模型几乎在 100% 的时间里生成语法正确的代码。Veracode 的 2026 年报告直言不讳：“语法问题实际上已经解决。”

这听起来像是一个里程碑，但正是它导致了你的问题。

该同一份报告对超过一百个模型进行了测试，发现平均安全通过率为 56%，“相比首份报告的 55% 几乎没有变化”，大约有 44% 的代码生成任务引入了风险漏洞。

功能正确性和安全性实际上是两个独立的问题，而只有其中一个问题接近被解决。

这一结果既不是异常值，也不是新发现。2022 年 IEEE 安全与隐私会议上，纽约大学坦顿分校的一个团队让 GitHub Copilot 经历了 89 个与安全相关的场景，生成了 1,689 个程序，并发现大约 40% 的程序存在漏洞，这些漏洞涉及 MITRE CWE Top 25 中的某项。该论文后来被选为 Communications of the ACM 研究亮点。

2024 年 11 月，乔治城大学安全与新兴技术中心评估了五个大语言模型，并报告称它们生成的代码片段中几乎有一半包含可能导致被利用的漏洞。四年、四个独立团队、四种方法论，每次得到的答案都相同。

在 2023 年 ACM CCS 上，斯坦福大学的 Neil Perry、Megha Srivastava、Deepak Kumar 和 Dan Boneh 将开发者纳入实验：47 名参与者，五项与安全相关的编程任务，三种编程语言，其中 33 人获得了 AI 助手，14 人未获得。使用助手的组别编写的代码安全性显著降低，并且他们更相信自己编写的代码是安全的。

这是一项小规模研究，它解释了为什么问题不会自行纠正：本来能够捕捉这一问题的机制（开发者会更仔细地查看让他们担心的代码）正是工具所关闭的机制。

这些研究都衡量的是应用程序代码。但基础设施代码则是更难的情况，因为一个有问题的安全组永远不会失败：它完全按照编写的方式运行，向任何请求的人提供流量，而唯一会提出异议的是阅读差异的人。

我无法通过阅读来审查如此庞大的代码量，其他人也做不到。我能做的就是将规则写成一种形式，让计算机在每次更改时进行检查，这就是 Policy as Code 的含义。

在本手册中，我将带你一步步构建这一检查。我们将把它指向一个真实存在漏洞的仓库，观察其明显版本如何清除其面前的九个违规项中的五个，然后进行修复。

到最后，你将了解如何：

编写一个针对 terraform show -json 输出的 JSON 的 Rego 策略。
以测试应用代码的方式测试策略，使用 fixtures 和覆盖率报告。
构建一个命令行门禁，其退出码契约能被 CI 管道信任。
使用 CEL 在 Kubernetes 准入时阻止不合规的工作负载。
让模型编写策略，并让 opa check 和你自己的测试决定是否保留它。
从同一个策略引擎授权 AI 代理的工具调用。

先决条件
通俗易懂的关键术语
步骤 1：获取用于测试的真实基础设施
步骤 2：在策略之前编写测试
步骤 3：编写策略直至测试通过
步骤 4：指向真实计划
步骤 5：将判决转换为退出码
步骤 6：在准入时强制执行
步骤 7：让模型编写策略
步骤 8：治理代理自身
步骤 9：我哪里错了
检查的局限性
结论

先决条件

您需要：

一个终端和一个可用的 python3（3.10 或更高版本）。
jq，用于在命令行读取 JSON。
大约 700 MB 的磁盘空间，因为 AWS Terraform 提供者很大。
一个 Anthropic API 密钥，但仅用于步骤 7。其他所有步骤均可离线运行。

mkdir policy-lab && 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 && 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 && sudo mv terraform /usr/local/bin/

在 Windows 上，使用 .venv\Scripts\activate 激活环境，并交换以下两个下载 URL：opa_windows_amd64.exe 和 terraform_1.14.2_windows_amd64.zip。

我在 macOS 上对以下内容进行了测试：OPA 1.20.2、Terraform 1.14.2 和 AWS provider 6.x。策略语法在 OPA 1.x 系列中保持稳定。

如果您使用的是 OPA 0.x，则此处的每条规则都需要在顶部添加 import rego.v1，我建议直接升级。违规计数仅通过计划 JSON 的形状随 AWS provider 版本变化，而自 provider 5 起该形状一直保持稳定。

关键术语（通俗解释）

策略即代码: 您的组织已经达成一致的规则，以程序形式编写，接受提出的更改并返回决策。
Rego: Open Policy Agent 评估的查询语言。它是声明式的：规则主体是必须全部满足的条件列表。
计划 JSON: 由 terraform show -json 生成的机器可读描述，说明 Terraform 即将执行的操作。这是策略读取的内容，因此您的 .tf 文件不会直接被策略读取。
准入控制: Kubernetes API 服务器内部的一点，对象在此可以在被存储之前被拒绝。
CEL: Common Expression Language（CEL），Kubernetes 在 API 服务器内部原生评估的小型表达式语言，无需部署 webhook。
误放行: 策略本应拒绝却放行的资源。几乎没有人会注意到这一点，因此除非你主动去查找，否则它不会被测量。

同样的判断在三个位置发生，且只有中间的那个与 Kubernetes 特定相关。

步骤 1：获取真实基础设施以进行测试

我不想编造一个易受攻击的 Terraform 文件，因为编造等同于制造漏洞，而策略只会捕捉我自己植入的漏洞。于是我去寻找他人编写并发布的代码。

TerraGoat 是 Bridgecrew 发布的一个故意易受攻击的 Terraform 仓库。获取其 EC2 模块的固定提交：

SHA=729f8da62c6a85ce4af5ad3d123de97776d954c4
curl -s "https://raw.githubusercontent.com/bridgecrewio/terragoat/$SHA/terraform/aws/ec2.tf" \
  | sed -n '77,96p'

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"]
  }

第二行的注释出自 TerraGoat，而面向全球开放的端口 22 正是它所指向的发现。

TerraGoat 的模块在现代 Terraform 上无法初始化，原因是它仍在引号中写死了 type = "string"——这正是 Terraform 0.12 已弃用、1.x 直接禁止的语法。于是我把该资源移到自己写的最小模块里，只改动了对 TerraGoat 内部局部变量的两处引用。

创建 main.tf：

terraform {
  required_version = ">= 1.9"
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 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"
  }
}

生成计划 JSON：

terraform init
terraform plan -out=tfplan.binary
terraform show -json tfplan.binary > plan.json

模拟凭据很重要：在仅创建的配置上运行 terraform plan 时永不会调用 AWS，因此在使用 skip_credentials_validation 及其三个兄弟属性时，提供程序不会尝试进行身份验证，且永不会应用任何内容。

步骤 2：在策略之前编写测试

我希望的规则是：没有安全组应将管理端口暴露给公共互联网。

这听起来就像一行代码，而测试正是我确定它不成立的原因。创建 policy/network_test.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
}

以下是该文件所包含的内容：

使用 with input as fixture 为一个表达式注入假计划，这就是在没有云账户的情况下测试策略的方式。
test_denies_wide_open_port_range 期望 four 次违规，每个管理端口各一次，这是因为入站规则描述了一个端口范围。0-65535 规则打开 SSH 的范围与显式端口 22 规则完全相同，因而能够绕过等式检查。
三个测试覆盖了 Terraform 用于同一概念的三种 other 形态：IPv6 字段、现代的独立 aws_vpc_security_group_ingress_rule 以及已废弃的 aws_security_group_rule。我最初并未编写这些测试，第 9 步解释了它们的来源。
两个 test_allows_ 测试用例与拒绝测试同样重要。一种拒绝一切的策略会通过所有拒绝测试，但毫无价值。
最后三个测试是在策略已经 "完成" 之后才加入的，第 9 步解释了它们的来源。允许所有协议的规则会打开所有端口，不管其端口字段如何设置；而策略无法读取端口的规则必须被报告。

步骤 3：编写策略直至所有测试通过

Terraform 以四种形式描述入站规则，因此策略将这四种形式归一为一组，然后对该组进行一次判断。创建 policy/network.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 >= range[0]
	port <= 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",
	}
}

从上往下读：

Rego 中的规则主体是一个合取式。每一行都必须成立，且 some ... in 行会进行迭代，因此 OPA 会遍历资源、入站规则、字段和端口的所有组合。
三个独立的 exposures 规则共同定义一组，Rego 称之为增量定义。以后再添加第五种形状只会增加一个块，且不会对其下方的内容产生任何影响。
object.get(ingress, field, []) 在字段缺失时返回空列表，因此 IPv4 仅规则在策略查找 ipv6_cidr_blocks 时不会出错。
covered_ports 在这里充当安全阀，因为一个所有协议规则在计划中会报告 0 到 0，同时打开所有端口，因此端口字段不能被字面读取；若规则的端口为 null 或字符串，则该函数未定义，而第二个 deny 规则会将其视为违规。
considered 不属于判断的一部分。它记录该策略能够讨论的所有资源，而步骤 5 利用它来避免报告尚未获得的通过。

运行它：

opa test policy
opa check --strict policy
opa fmt --diff policy

PASS: 20/20

为 opa test 添加 -v 以每行显示一个测试。

opa check --strict 会捕获不安全的变量和被遮蔽的导入，而 opa fmt --diff 在格式已经是规范的情况下不输出任何内容。OPA 使用制表符格式化 Rego。这两者都应放在 CI 中，位于其他所有步骤之前。

第二个策略是所有权标记，因此创建 policy/tags.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
}

以下是这两行代码如此呈现的原因：

trim_space(value) == "" 进行此检查。之所以必须这样，是因为在 Rego 中只有 false 和 undefined 为假值，因此空字符串为真值。直接的存在性检查会愉快地接受 owner = ""，这正是标签策略旨在阻止的那种合规表演。
tags_of 带有其 else := {} 分支，是我编写且仅在第 9 步发现的 bug 的修复。

步骤 4：指向真实计划

针对 TerraGoat 计划评估一个软件包：

opa eval --data policy --input plan.json --format pretty 'data.terraform.network.deny'

[
  "aws_security_group.web-node: ingress rule exposes port 22 to 0.0.0.0/0"
]

该策略报告了一次违规，却对端口 80 保持沉默。在同一资源中，它对外开放——因为公共 Web 服务器的意义就在于它是公开的。能够同时标记这两项的检查，往往会被人们学会忽略。

为每个包运行单独的命令无法扩展，而 Rego 能够在单个查询中跨命名空间进行聚合：

opa eval --data policy --input plan.json --format pretty \
  'union({v | v := data.terraform[_].deny})'

[
  "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\""
]

{v | v := data.terraform[_].deny} 是一个推导式，用于从 data.terraform 下的每个包中收集 deny 集合，而 union 将它们展开。将新的策略文件放入该命名空间，它会被自动采纳，且无需更改命令。

步骤 5：将裁决转换为退出码

CI 网关通过其退出状态进行通信，而将两种故障合并为一个退出码正是导致损坏的管道能够持续一个月而未被发现的原因。此工具采用以下方式：

代码	裁决	含义
0	通过	策略已运行，检查了资源，未发现任何问题
1	失败	策略已运行并发现违规
2	空洞	策略已运行但未检查任何内容，因此结果无意义
2	故障	工具或其输入不可用

第四行是多数网关容易出错的地方。如果一个计划不包含任何策略已知的资源，则报告通过意味着在运行时无法提供的保证。它拥有自己的裁决名称，并且不会退出码为 0。

创建 policy_gate.py：

"""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) -> None:
    print(f"policy-gate: {message}", file=sys.stderr)
    sys.exit(BROKEN)


def load_plan(path: pathlib.Path) -> 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) -> 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() -> 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())

脚本的作用如下：

argparse 将 --plan 和 --policy 标记为 required=True，且 --policy 接受 nargs="+"，因此空策略列表在解析时会报错。
load_plan 会报告 JSON 语法错误的行号和列号，因为 json.JSONDecodeError 携带 lineno 和 colno，而仅返回 “invalid JSON” 的检查只会浪费别人的下午。
所有失败路径都会经过 die，die 始终返回退出码 2，因为缺少 opa、不可读的文件以及缺少 resource_changes 键的 plan 都属于工具自身的失败。
considered 查询会在之前运行，即在 deny 查询之前。如果未检查到任何内容，运行将在 VACUOUS 状态结束，因而没有机会打印通过。
违规会被排序，因而相同的 plan 在每次运行时会产生字节完全相同的输出；两次 CI 日志的 diff 若有变化则表示实际存在差异。

该 plan 中的两个资源共计七条违规，CI 可依据退出码采取行动。

对空 plan 使用相同的工具，若门禁返回 “pass” 则是在说谎。

GitHub Actions 工作流在使用这些策略判断任何内容之前，会先对其进行测试：

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 > coverage.json

      # Only now does anything get judged.
      - name: Evaluate Terraform plan
        run: python3 policy_gate.py --plan plan.json --policy policy

先让该网关仅进行报告，运行两周，再让它开始阻止。一个看起来显然正确的策略在你的实际环境中可能会因某些结构性问题而失效，你宁愿从日志行中发现这一点，也不希望在发布被阻止时才察觉。

步骤 6：在准入时强制执行

步骤 5 中的网关检查你打算部署的内容。它不会看到来自某人笔记本的 kubectl apply、供应商的 Helm chart，或操作员按照自己的调度创建 Pod。对于这些情况，你需要准入控制，而 Kubernetes 现在已经内置了此功能。

ValidatingAdmissionPolicy 自 v1.30 起已成为通用可用功能，它在 API 服务器内部评估 CEL，无需部署或维护 webhook：

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: >-
        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: >-
        variables.allImages.all(i, i.startsWith('registry.internal.example.com/'))
      messageExpression: >-
        'images must come from registry.internal.example.com: ' +
        variables.allImages.filter(i,
          !i.startsWith('registry.internal.example.com/')).join(', ')
      reason: Forbidden

策略本身不会产生任何效果，需要通过绑定来激活——正因如此，你才能先在单个命名空间中试点：

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"

设置 validationActions: ["Warn", "Audit"]，为一个命名空间添加标签，观察一周，然后切换到 ["Deny"] 并扩大选择器。

initContainers 的处理很重要，因为这是镜像来源策略最常见的被绕过方式，而 ? 可选字段语法配合 .orValue([]) 是读取可能缺失的列表而不导致整个表达式出错的方法。

我无法应用这两个清单，因为我手头没有集群。它们已根据 v1 参考模式进行检查，且没有真实的 API 服务器接受它们，因此请将它们视为起点，并在 Warn 模式下推出（无论如何你都应该这样做）。

两个其他引擎在广泛的生产环境中使用，起始于 Kyverno。它 在 2026 年 3 月 CNCF 毕业，在 Bloomberg、Coinbase、德意志电信、领英和 Spotify 有生产使用。其策略使用 YAML 编写，因此平台团队无需学习新语言，并且它处理生成、镜像签名验证以及内置策略未覆盖的清理工作。

OPA Gatekeeper 是您想要一个覆盖 Kubernetes 和 Terraform 和 CI 的单一 Rego 代码库时的正确答案，这也是本教程旨在达到的目标。突变现已内置：MutatingAdmissionPolicy 在 v1.36 中变得稳定。

步骤 7：让模型编写策略

策略很繁琐，而模型擅长处理繁琐的工作。因此，明显的做法是让模型来编写它们。

有一个你可以在大约一分钟内自行测量的陷阱，我在这一步的末尾正好这样做：公开训练数据中大量的 Rego 是 Rego v0，这是 OPA 1.0 在 2025 年 1 月发布时停止解析的方言。模型在寻求它见过的最常见模式时，会转向当前解析器拒绝的方言。

卡拉布里亚大学一个小组的 2025 年预印本 ARPaCCino 在一个 Terraform 案例研究中报告了相同的效果：在没有工具的情况下要求生成 Rego，Qwen3-30B 和 GPT-4o 各自产生了 0/5 个语法正确的策略。在 OPA 文档上添加检索没有改变任何东西。给模型一个可以运行 opa check 并读取错误的循环，使其分别达到 4/5 和 5/5。

这些计数来自一个小规模的案例研究，因此请将方向视为结果的持久部分。

反馈循环是解决它的方法，且 该循环没有成本，因为您已经通过 opa check --strict、opa fmt 和 opa test 构建了它。

构建带有一次反转的循环，使其值得信赖。我编写测试，模型生成策略。测试夹具具体且审查成本低，因为你只需读取一个 JSON 对象，三秒内就能说“应该拒绝”。带有嵌套推导的 Rego 阅读起来费力，且易于误解。把人放在审查成本低的环节，让机器在其输出可机械验证的地方工作。

创建 policy_forge.py：

"""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) -> str:
    blocks = re.findall(r"```rego\n(.*?)```", reply, re.DOTALL)
    if not blocks:
        raise ValueError("model returned no rego block")
    if len(blocks) > 1:
        raise ValueError(f"model returned {len(blocks)} rego blocks; expected one")
    return blocks[0]


def verify(opa: str, policy: str, tests: pathlib.Path) -> 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) -> 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]) -> 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() -> 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 < 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())

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

循环保证的内容如下：

验证以子进程方式运行：模型永远不会被询问其策略是否正确。opa check 和 opa test 进行判断，它们的退出码是循环接受的唯一证据。
故障以原始工具输出形式返回，永不被概括：编译器错误和测试失败是模型能够获得的最高信号反馈，而意译会丢掉有帮助的部分。
未通过前不会有任何内容写入磁盘：forge要么返回已验证的策略，要么抛出异常，因此不会出现仅因重试预算用尽而将未验证的策略写入仓库的路径。

我使用脚本化模型驱动循环，因此在不需要 API 密钥的情况下结果可重现。三个回复分别是一个符合 v0 语法的现实策略、一个看似正确但实际错误的 v1 策略，以及步骤 3 中的策略：

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

尝试 1 是 Rego v0，即 deny[msg] { ... } 形式，在 OPA 1.0 于 2025 年 1 月发布时停止解析。而且这几乎是公开训练数据所包含的全部。opa check --strict 在它到达测试之前就将其拒绝。

尝试 2 是有效的 Rego v1。大多数工程师审查后会通过，但它在套件上的得分仍仅为 4 of 8。这是因为它直接将 ingress.from_port 与 admin_ports 进行比较，忽略了范围，并且仅读取了 cidr_blocks。opa check 没有任何抱怨，因为代码结构完全正确。

结构正确的策略仍可能是错误的策略，而在这个循环中唯一知道你想要什么的只有你编写的测试套件。

修复循环不是单调的，因为每次尝试都是基于错误信息的全新生成，且不会继承之前有效的内容，因此第四次尝试可能会丢失第三次尝试所具备的属性。该设计中没有任何东西能够检测到这一点，因为唯一被检查的就是你编写的测试套件。

限制重试次数，保持套件持续增长，并将每个生成的策略视为在合并前需要有人批准的拉取请求。

步骤 8：治理代理自身

AI 代理也是一个行为体，它会调用工具，因此每次工具调用都成为必须由某方做出的授权决策。

行业很快就达成了共识：Amazon Bedrock AgentCore Policy 在 2026 年 3 月达到通用可用性，在网关处使用 Cedar 评估代理工具调用。常见的开源模式是在 MCP 工具网关前放置一个 OPA sidecar。

研究正在更深入地推进这一边界：华盛顿大学（Defects4J 背后的团队）2026 年的预印本 Solver-Aided Verification of Policy Compliance in Tool-Augmented LLM Agents（Winston, Winston 和 Just）将自然语言策略编译为 SMT 约束，并使用 Z3 求解器阻止不合规的调用。

他们共同主张：系统提示中的策略并不是强制执行。强制执行是位于调用路径上的拦截器，它可以返回 “no” 并阻止调用发生。

创建 agent/authz.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) > 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) > 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 > 0
}

escalations contains reason if {
	input.tool == "issue_refund"
	amount := object.get(input, ["arguments", "amount_cents"], null)
	is_number(amount)
	amount > 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"
}

决策词汇已经封闭，我在这里直白地说明：

效果	调用方的行为
允许	运行工具
需要批准	暂停，向人展示原因，仅在获得批准后运行
拒绝	拒绝，且不提供批准路径

以下是政策如此设定的原因：

default decision 是拒绝： 未识别的工具、您忘记映射的角色或格式错误的输入都会落在这里。一种默认允许的政策会在没有人预料到的输入上打开漏洞，而这些输入正是攻击者会挑选的集合。
三个值： 二元授权迫使在阻止有用工作和允许危险工作之间做出选择，而第三个值正是使高自主性代理能够被容忍的关键。
原因以集合形式返回： 所有适用的原因都会被收集。当有人在凌晨三点收到批准提示时，“退款 250000 分超过了 10000 分上限”会告诉他们该怎么做。而“政策违规”则不会。
参数也会被检查： issue_refund 在 £5 时是常规操作，在 £2,500 时是严重操作，因此仅凭工具名的粒度对代理来说远远不够细致，因为代理会选择参数。

十五个测试覆盖了决策表，包括角色列表为空的代理、金额为零的退款，以及在换行符后隐藏 DROP 的查询：

opa test agent -v

PASS: 15/15

把服务跑起来，然后试着拨打一通电话：

opa run --server --addr localhost:8181 agent/

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

{
  "effect": "require_approval",
  "reasons": [
    "refund of 250000 cents exceeds the 10000 cent ceiling"
  ]
}

策略本身是惰性的——只有当某个环节依据它的判定拒绝继续执行时，它才会真正发挥作用。而这个环节，正是客户端：

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

search_orders    -> ALLOWED
issue_refund     -> NEEDS APPROVAL (refund of 250000 cents exceeds the 10000 cent ceiling)
delete_account   -> DENIED (no matching grant)

Note body.get("result", ...): 当查询没有匹配任何内容时，OPA 会返回状态码 200 的空对象 {}，因此直接使用 body["result"] 会引发 KeyError，并且根据您的代理框架如何处理异常，这可能导致开放失败。每一层默认拒绝，包括解析。

在工具函数运行之前，从框架的工具执行钩子中调用 authorize()。这大约有十五行代码，它将系统提示的礼貌建议转化为实际的边界。

步骤 9：我犯的错误

步骤 2 和 3 展示了完成的策略。我最初尝试了更简单的方法（大多数教程止步于此的版本），而那与您刚刚读到的内容之间的差距才是这里最有价值的部分。

标记策略跳过了最糟糕的资源

我天真的 in_scope 规则以 resource.change.after.tags 结束，其含义为“仅包含具有标签的资源”。

它实际上的作用更糟，因为 Terraform 会为完全没有标签的资源发出 tags: null，而未定义的查找会导致规则主体失败，因而该资源完全被排除在范围之外。

TerraGoat 计划包含两个资源：安全组携带五个 git_* 标签且没有所有权标签，而 VPC 则完全没有任何标签。

jq -r '.resource_changes[] | "\(.address): tags=\(.change.after.tags | type)"' plan.json

aws_security_group.web-node: tags=object
aws_vpc.web_vpc: tags=null

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

3
6

它漏掉的三个都在完全未标记的资源上，因此策略为该资源添加了一些标签，并静默清除了没有标记的那个。

tags_of 及其 else := {} 分支是修复方案，仅三行代码。

网络策略读取四种形状之一

我的天真网络策略读取了 aws_security_group 和 cidr_blocks，这正是每个教程所展示的，但 Terraform 有四种方式来表达相同的入站规则。

![“Terraform 描述一条入站规则的四种方式”的图示。图中显示四个框。左上角为 aws_security_group.ingress[].cidr_blocks，填充淡蓝色并标注为“已读取”。其余三个框用红色轮廓和红色斜线标出，标注为“未读取”：分别为 aws_security_group.ingress[].ipv6_cidr_blocks、aws_vpc_security_group_ingress_rule.cidr_ipv4 以及已废弃的 aws_security_group_rule。图例说明：实心蓝色填充表示策略会检查该位置，红色斜线表示策略不会检查该位置。](https://cdn.hashnode.com/uploads/covers/5f3a74bfc4d5973f55c91c8c/ab61d36d-78bf-45f7-a073-e7013216e3e1.png align="center")

我另外设计了一个真实配置，包含三个安全组，利用策略未读取的形状向全球开放 SSH。两个版本都位于 code/ 目录下，因而可以这样重现：

opa eval --data naive --input plan-evasion.json \
  --format pretty 'data.terraform.network.deny'

[]

那就是三个公开可达的 SSH 端口，零违规，以及一个本来会打印 PASS 并以退出码 0 退出的门禁。

一个一直让漏洞保持打开的测试

端口范围还存在第二个问题，而我自己的测试套件一直在保护它。我编写了一个名为 test_tolerates_null_ports 的测试用例，断言带有 protocol: "-1" 的入站规则不会产生任何违规，理由是对空端口的比较不应导致策略崩溃。

一个允许所有协议的规则会打开所有端口，AWS 提供方将其记录为 from_port: 0 和 to_port: 0。范围检查将其按字面理解为仅端口 0，因此 AWS 中最宽松的规则得分为通过（即 PASS）。

计划位于 code/plan-all-protocols.json 中，且它仅包含一个资源：

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'

{"protocol":"-1","from_port":0,"to_port":0,"cidr_blocks":["0.0.0.0/0"]}
[]

所有协议和所有端口面向整个互联网开放，我特意编写的测试认为一切正常。修复方案是 covered_ports，它把一个适用于所有协议的规则映射到完整端口范围，对无法读取的端口保持未定义，并通过第二个 deny 规则报告这种未定义情况。原来的测试已被删除，取而代之的是四个测试：

opa eval --data policy --input plan-all-protocols.json --format pretty 'data.terraform.network.deny'

[
  "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"
]

步骤 7 认为，测试套件是循环中唯一知道你想要什么的制品。这正是该属性的代价：一个错误的测试就是一个错误的规范，而它下游的任何东西都不会提出异议。

数字实际上是什么

这张图显示了在两个真实Terraform plan JSON文件中发现的违规行为的数量

两个版本在我为政策设计的计划上均得分 1.000。只有在我未设计的计划上才会出现差距。

在两个真实计划中，天真的策略发现了 9 个违规中的 4 个。它们在 TerraGoat 安全组上得分 1.000 —— 这就是我在编写它们时所想到的情况；而在我未考虑的两种情况下，它们的得分分别为 0.000 和 0.500。

真正让我惊讶的部分

我原本以为测试覆盖率会发现这个问题，但事实并非如此。我使用最初为它编写的两个测试重建了天真的标记策略：

opa test . --coverage --format json | jq '{overall: .coverage}'

{
  "overall": 100
}

天真的政策达到了 100% 的覆盖率，通过了 2/2 项测试，并清除了一个完全没有标签的资源。

覆盖率衡量您的测试执行了策略的哪些行，但完全未说明您未能想象到的输入形状。对策略代码而言，这就是全部的失效模式。覆盖率值得报告，而策略实际上是否有效的证据来自于在您未编写的基础设施上运行它。

检查的局限

这个门禁仍然无法做到的事情在于，其误放行的后果比它所捕获的问题更为严重。

1. 未知值是不可见的

Terraform 将在应用之前无法解析的任何内容标记为未知，这在计划 JSON 中表现为 null，并伴随一个 after_unknown 映射。当该字段为未知时，策略读取 change.after.some_field 不会触发。

这在跨资源规则上影响最大，因为在计划阶段，‘每个存储桶都有一个公共访问块’实际上很难实现，原因是该块引用的存储桶 ID 通常是 (known after apply)。

2. 它只能看到计划，且仅能看到计划

在流水线之外应用的更改、在控制台中修改的或自创建以来漂移的内容对它来说都是不可见的。计划时检查和准入时检查具有不同的盲点，两者都需要定期扫描已部署的状态来补充工作。

3. 控制的覆盖率无法从内部衡量

一百个通过的检查无法说明没有人编写的规则。请将您的控制需求与策略文件的映射放在显式位置，并按计划进行审计，因为门禁无法告知您它从未被要求检查的内容。

4. CIDR 列表是精确匹配

public_cidrs 包含 0.0.0.0/0 和 ::/0，除此之外什么也没有，因此打开 0.0.0.0/1 的规则将覆盖互联网的一半并通过。扩展它意味着决定哪些前缀长度算作公共地址，并划出 RFC 1918 地址空间，这一决定应由您的组织来做。

5. 我发现的四种形状

exposures 集合涵盖了我所寻找的四种编码，而 AWS 还提供更多。安全组引用（security_groups）、前缀列表以及 self 规则都是到达端口的方式，而此策略未建模这些方式，并且它永远不会看到它们。我预计在首次针对大规模环境运行时会出现第五种形状。

6. 监管日期会变动

如果你正在为欧盟AI法案做准备，2026年7月发布的《数字综合法案》将附件III的高风险义务从2026年8月2日推迟至2027年12月2日，附件I的义务推迟至2028年8月2日，而第50条透明度义务保持其原始的2026年8月2日日期。

编码这些控制措施，并在需要时查阅官方时间线上的日期，即使上一季度的博客文章告诉你相反的情况（本文也包括在内）。

结论

模型生成的基础设施代码几乎总是可以解析，大约有56%的时间是安全的，并且其生成速度快于任何人能够阅读的速度。手动审查在那个差距某处不再是真正的控制手段。这些规则本来就应该是可执行的，而规模正是最终推动问题的因素。

在本教程中，你将：

从TerraGoat的提交729f8da中拉取了一个真实的易受攻击的安全组，并使用Terraform 1.14.2进行了规划。
编写了20条策略测试，覆盖了一个入站规则的四种Terraform编码，并在OPA 1.20.2上使它们全部通过。
在TerraGoat计划中捕获了7个真实违规，同时正确地忽略了端口80。
构建了一个具有四种裁决的合约的门禁，当其未检查到任何内容时退出码为2。
观察到天真的版本在报告100%测试覆盖率的情况下仅发现了9个违规中的4个，随后对其进行加固以发现全部9个违规。
构建了一个生成循环，其中opa check和你自己的测试共同决定什么内容会写入磁盘。
授权来自同一引擎的代理工具调用，包含15项测试且默认拒绝。

我的策略正确处理了我为其编写的用例，但遗漏了我未曾想象的两种情况。我可获得的所有信号（测试通过、覆盖率100%、以及一次干净的opa check)都表明它们是没问题的。唯一不一致的是他人编写的基础设施。请将你的策略指向你未编写的代码，尽早进行，并保留失败案例。

所有代码、策略、计划JSON以及生成图形的脚本都位于与本手册并存的code/目录中。图形可通过python3 build/make_images.py和python3 build/make_terminals.py重新生成。终端截图在构建时会重新运行其命令，因此不会偏离真实情况。

目录