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

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

2026年9月29日

/

#infrastructure

Kayode Adeniyi

现代模型生成语法正确的代码准确率接近100%。Veracode 2026年报告直言不讳："语法问题实际上已经解决。"

这看似是一个里程碑，但恰恰说明你遇到了问题。

同一份报告测试了超过100个模型，发现平均安全通过率仅为56%，"与首次报告的55%基本持平"，约44%的生成任务会引入潜在风险漏洞。

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

这个结果并非异常现象，也非新发现。2022年IEEE安全与隐私会议上，纽约大学坦登工程学院团队对GitHub Copilot进行了测试，将其置于89个安全相关场景中，生成了1689个程序，发现约40%的程序存在MITRE CWE Top 25漏洞。该论文后来被选为ACM通讯研究亮点。

2024年11月，乔治城大学安全与新兴技术中心评估了五个大语言模型，报告称他们生成的代码片段中近一半包含可能导致被利用的漏洞。四年间，四支独立团队采用四种不同方法进行研究，得出的结论却始终如一。

2023年ACM CCS会议上，斯坦福大学的Neil Perry、Megha Srivastava、Deepak Kumar和Dan Boneh将开发者纳入实验：47名参与者完成五个安全相关编程任务，使用三种编程语言，其中33人获得AI助手协助，14人没有。使用AI助手的组别编写的代码安全性显著降低，且更可能误认为自己编写的代码是安全的。

这是一项小型研究，但解释了为何问题不会自行纠正：通常能发现这类问题的机制（开发者对可疑代码进行更仔细检查）恰恰被工具关闭了。

这些研究都针对应用代码。但基础设施代码才是更棘手的情况，因为一个存在安全问题的代码组从不会失败：它会严格按照编写方式运行，为任何请求流量的用户服务，唯一会反对的只有阅读代码差异的人。

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

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

最终你将掌握：

• 针对terraform show -json生成的JSON编写Rego策略。
• 像测试应用代码一样使用固定用例和覆盖率报告测试策略。
• 构建一个CI流水线可信任的命令行门禁并定义退出码契约。
• 使用CEL在Kubernetes准入阶段阻止不合规的工作负载。
• 让模型编写策略，并通过OPA检查和你自己的测试决定是否保留。
• 从同一策略引擎授权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 激活环境，并将 opa_windows_amd64.exe 和 terraform_1.14.2_windows_amd64.zip 的下载 URL 互换。

我在 macOS 上使用 OPA 1.20.2、Terraform 1.14.2 和 AWS 提供商 6.x 运行了以下所有内容。OPA 1.x 的策略语法是稳定的。

如果你使用的是 OPA 0.x，此处每个规则都需要在顶部添加 import rego.v1，我建议你升级版本。违规计数仅通过 plan JSON 的结构依赖于 AWS 提供商版本，而该结构自提供商 5 版本以来一直保持稳定。

平实英语中的关键术语

• 代码化策略：组织已达成一致的规则，以程序形式编写，接收提议更改并返回决策。

• Rego：Open Policy Agent 评估的查询语言。它是声明式的：规则体是必须全部满足的条件列表。

• Plan JSON：Terraform 即将执行的操作的机器可读描述，由 terraform show -json 生成。这是策略读取的内容，因此你的 .tf 文件永远不会接触到它。

• 准入控制：Kubernetes API 服务器内部拒绝对象的点，在对象存储之前。

• CEL：Common Expression Language，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" {
  # 安全组在 SSH 端口对世界开放
  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"
    }
  }
}

# 模拟凭证。该配置仅用于规划阶段，永不实际应用，
# 因此提供程序不得尝试连接 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"
}

# 直接从 bridgecrewio/terragoat，terraform/aws/ec2.tf，提交记录 729f8da 复制。
# 仅将对 TerraGoat 本地变量的两个引用替换为字面量。
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"
  }
}

生成 plan.json 文件：

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

模拟凭证至关重要：对仅创建资源的配置执行 terraform plan 时永远不会调用 AWS，因此通过设置 skip_credentials_validation 及其三个相关参数，提供程序不会尝试进行身份验证，且永远不会实际应用任何配置。

第二步：在策略之前编写测试

我想要的规则是：任何安全组都不允许将管理端口暴露给公共互联网。

这听起来像是只需一行代码，但测试正是用来明确为何需要这样实现的。创建 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
}

# 仅检查 from_port 等值会遗漏这种情况。范围检查可以覆盖。
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 }

一个支持所有协议的规则会打开所有端口，无论其端口字段如何设置。

该测试的早期版本持相反观点，从而隐藏了错误。

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 }

策略无法读取的端口会被报告，但不会被通过。

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 }

对于非面向世界开放的规则，无法读取的端口不会触发告警。

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` 决定了网关的通过或无效判断，因此即使不参与判断也需要单独测试。

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 期望检测到四个违规项，每个管理端口对应一个违规项，因为入站规则描述的是一个端口范围。0-65535 的规则与显式指定端口 22 的规则在开放 SSH 端口时具有相同效果，但会绕过等值检查。

- 三个测试用例覆盖了 Terraform 用于相同概念的三种其他结构：IPv6 字段、现代独立资源 aws_vpc_security_group_ingress_rule 以及已弃用的 aws_security_group_rule。这些测试用例并非最先编写，步骤9 会解释它们的来源。

- 两个 test_allows_ 测试用例与拒绝规则同样重要。一个拒绝所有流量的策略虽然能通过所有拒绝测试，但毫无实际价值。

- 最后三个测试用例是在策略"完成"后才添加的，步骤9 会解释它们的来源。一个支持所有协议的规则会忽略端口字段的设置而开放所有端口，而策略无法读取端口字段的规则必须被报告。

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

Terraform 用四种结构描述入站规则，因此策略会先将这四种结构统一为一个集合，然后对该集合进行一次判断。创建 policy/network.rego：

METADATA

标题：禁止从公共互联网访问管理端口

描述：|

Terraform 用四种不同结构描述入站规则。每种结构都会先被统一为一个 `exposures` 集合，因此下面的判断只需编写一次，新增结构只需增加一个辅助规则。

#

这里的两个设计是刻意为之而非偶然。所有协议规则会覆盖所有端口，无论端口字段如何设置，而策略无法读取端口字段的规则必须被报告而非通过。

package terraform.network

admin_ports := {22, 3389, 3306, 5432}

public_cidrs := {"0.0.0.0/0", "::/0"}

AWS 提供商用 from_port 0 和 to_port 0 表示所有协议规则，这会开放所有端口，因此端口字段不能被字面解析。

all_protocols := {"-1", "all"}

结构1和2：内联入站块，IPv4和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, } }

结构3：自v5版本以来AWS推荐的独立规则。

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, } }

# 形状 4：仍存在于大多数现有环境中的已弃用独立规则
exposures 包含 exposure 如果 {
	某些资源在 input.resource_changes 中
	resource.type == "aws_security_group_rule"
	resource.change.after.type == "ingress"
	某些 cidr 在 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,
	}
}

# 规则实际覆盖的端口范围。当策略无法判断时为未定义
covered_ports(exposure) := [0, 65535] 如果 {
	exposure.protocol 在 all_protocols 中
}

covered_ports(exposure) := [exposure.from_port, exposure.to_port] 如果 {
	不 exposure.protocol 在 all_protocols 中
	is_number(exposure.from_port)
	is_number(exposure.to_port)
}

deny 包含 msg 如果 {
	某些 exposure 在 exposures 中
	exposure.cidr 在 public_cidrs 中

	# 如果端口位于 [from_port, to_port] 范围内，则规则覆盖该端口
	range := covered_ports(exposure)
	某些 port 在 admin_ports 中
	port >= range[0]
	port <= range[1]

	msg := sprintf(
		"%s: 入站规则将端口 %d 暴露给 %s",
		[exposure.address, port, exposure.cidr],
	)
}

# 对于策略无法读取端口信息的全球开放规则会进行报告
# 通过此规则意味着策略在宽松方向上进行了猜测
deny 包含 msg 如果 {
	某些 exposure 在 exposures 中
	exposure.cidr 在 public_cidrs 中
	不 covered_ports(exposure)

	msg := sprintf(
		"%s: 到 %s 的入站规则具有策略无法评估的端口 (%v 到 %v)",
		[exposure.address, exposure.cidr, exposure.from_port, exposure.to_port],
	)
}

# 策略能够检查的地址列表。网关使用此信息区分 "未违规" 和 "未检查"
considered 包含 resource.address 如果 {
	某些资源在 input.resource_changes 中
	resource.type 在 {
		"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：

/think

# METADATA
# 标题：每个托管资源都带有所有权标签
# 描述： |
#   当资源没有任何标签时，Terraform 会输出 `tags: null`，因此针对 `after.tags` 的策略会跳过那些标签最差的资源。`tags_of` 会将该 null 强制转换为空对象。
package terraform.tags

required_tags := {"owner", "cost-center", "data-classification"}

# 真正无法携带标签的资源类型。
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 步才发现的错误。

第 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 保持沉默。因为同一资源中该端口对世界开放，而公共网络服务器的职责就是提供公共访问。一个同时标记这两个端口的检查会导致人们逐渐忽略该检查。

为每个包运行单独命令无法扩展，而 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 下每个包中收集拒绝集，而 union 则将其扁平化。将新的策略文件放入该命名空间时，无需更改命令即可自动识别。

第 5 步：将判断结果转换为退出码

CI 管道通过其退出状态进行通信，而将两种类型的失败合并为一个代码会导致管道错误运行一个月。该工具使用以下规则：

代码

判断结果

含义

0

通过

策略已运行，检查了资源，未发现任何问题

1

失败

策略已运行并发现了违规项

2

无效

策略已运行但未检查任何内容，因此结果无意义

broken

工具或其输入不可用

第四行是大多数门最容易出错的地方。如果一个计划中不包含任何策略所知的资源，报告通过会声称运行可以提供的保证。它会获得自己的判决名称，并且不会返回0。

创建 policy_gate.py：

"""将 Terraform 计划与 Rego 策略目录进行对比评估。"""

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"无法读取 {path}: {exc.strerror}") try: return json.loads(text) except json.JSONDecodeError as exc: die(f"{path}:{exc.lineno}:{exc.colno}: 无效 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 失败: {result.stderr.strip() or result.stdout.strip()}") values = json.loads(result.stdout)

一个产生非字符串结果的规则是策略错误，对混合列表进行排序会以无关的 TypeError 暴露该问题。

for value in values: if not isinstance(value, str): die(f"{expr} 产生了 {type(value).__name__}; " "deny 和 considered 规则必须产生字符串") return values

def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--plan", required=True, type=pathlib.Path, help="来自 terraform show -json 的 JSON") parser.add_argument("--policy", required=True, nargs="+", type=pathlib.Path, help="一个或多个包含 .rego 文件的目录") parser.add_argument("--opa", default="opa", help="opa 可执行文件的路径") args = parser.parse_args()

if shutil.which(args.opa) is None: die(f"{args.opa} 不在 PATH 中") for directory in args.policy: if not directory.is_dir(): die(f"{directory} 不是目录")

plan = load_plan(args.plan) if "resource_changes" not in plan: die(f"{args.plan} 没有 resource_changes 键; 它是 Terraform 计划吗？")

considered = query(args.opa, args.policy, args.plan, CONSIDERED_QUERY) if not considered:

在此处报告通过会声称运行无法提供的保证。

print(f"VACUOUS: 没有策略检查 {len(plan['resource_changes'])} 个计划资源中的任何一个", file=sys.stderr) return BROKEN

violations = sorted(query(args.opa, args.policy, args.plan, DENY_QUERY)) if violations: print(f"FAIL: {len(violations)} 个违规项 " f"在 {len(considered)} 个检查资源中", file=sys.stderr) for violation in violations: print(f" - {violation}", file=sys.stderr) return FAIL

print(f"PASS: 检查了 {len(considered)} 个资源，没有违规") 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 函数处理，该函数始终以状态码 2 退出，因为缺少 opa、不可读文件或计划中没有 resource_changes 键都是工具本身的故障。

- 审核查询会在拒绝查询之前执行。如果未检查任何内容，流程会在 VACUOUS 阶段结束，不会有机会输出通过结果。

- 违规项会按顺序排序，因此相同计划在每次运行时都会生成字节完全一致的输出，两个 CI 日志的差异具有实际意义。

该计划涉及两个资源的七项违规，且包含 CI 可操作的退出码。

在空计划上使用相同工具时，一个回答 "pass" 的门控机制实际上是在撒谎。

GitHub Actions 工作流在使用策略进行判断前会先测试这些策略：

name: policy

on: [pull_request]

jobs: policy: runs-on: ubuntu-latest steps:

• uses: actions/checkout@v5

• name: 安装 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

策略是代码。在信任它们之前进行语法检查和测试。

• name: 检查策略语法

run: opa check --strict policy

• name: 验证格式

run: opa fmt --fail --diff policy

• name: 测试策略

run: opa test policy --verbose --coverage --format json > coverage.json

只有在此之后才会进行判断。

• name: 评估 Terraform 计划

run: python3 policy_gate.py --plan plan.json --policy policy

先让门控机制仅报告结果，持续运行两周后再启用阻断功能。一个看起来显然正确的策略可能会在您实际环境中因结构性问题而失败，您更希望从日志中发现这个问题，而不是被阻断的发布流程所困扰。

## 第6步：在准入阶段强制执行

第5步中的门控机制检查的是您打算部署的内容。它无法看到某人笔记本电脑上的 kubectl apply、供应商的 Helm chart 或按计划自行创建 Pod 的操作员。对于这些情况，您需要准入控制，而 Kubernetes 现在已内置了该功能。

自 v1.30 版本起，ValidatingAdmissionPolicy 已正式发布，它在 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:

Pod 包含三个容器列表。仅读取 spec.containers 的策略可能被绕过，因为可以将镜像移动到 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 服务器接受过它们，因此请将它们视为起点，并以警告模式（这本来就是您应该采取的措施）逐步部署。

目前还有两个其他引擎被广泛用于生产环境，首先是 Kyverno。它于 2026 年 3 月在 CNCF 毕业，已被 Bloomberg、Coinbase、Deutsche Telekom、LinkedIn 和 Spotify 等公司用于生产环境。其策略使用 YAML 编写，因此平台团队无需学习新语言，它还处理内置策略未涉及的生成、镜像签名验证和清理操作。

当您需要一个 Rego 代码库同时覆盖 Kubernetes、Terraform 和 CI 时，OPA Gatekeeper 是正确选择，这也是本教程最终要达到的目标。现在突变操作也已内置：MutatingAdmissionPolicy 在 v1.36 中已稳定。

## 第 7 步：让模型编写策略

策略编写过程繁琐，而模型擅长处理繁琐任务。因此，显而易见的方案是让模型来编写策略。

不过有一个需要注意的陷阱，您可以在大约一分钟内自行验证，而我将在本步骤结束时进行测试：大量公开训练数据中的 Rego 实际上是 Rego v0，这是 OPA 1.0 于 2025 年 1 月发布时停止解析的旧版本。当模型试图使用它见过的最常见模式时，实际上会使用当前解析器拒绝的方言。

2025年来自意大利卡拉布里亚大学ARPaCCino团队的一篇预印本报告在Terraform案例研究中发现了相同现象：当要求生成Rego策略且不提供任何工具时，Qwen3-30B和GPT-4o在5个语法正确的策略中均未能生成任何有效结果。即使增加对OPA文档的检索功能也未带来任何改善。当为模型添加一个能够运行opa check并读取错误信息的反馈循环后，两者的成功率分别提升至5/5和4/5。

这些数据来自一个小型案例研究，因此应将这种趋势视为结果中更可靠的部分。

反馈循环才是解决问题的关键，而构建这个循环本身成本为零，因为其本质上就是由opa check --strict、opa fmt和opa test这些已有工具构成的。

因此通过一个简单的逆向设计构建这个循环，使其具有可信度。我来编写测试用例，让模型生成策略。测试用例具有具体明确的特性，且审查成本低廉——只需读取JSON数据块并在三秒内判断"是的，这个应该被拒绝"。而包含嵌套综合的Rego策略则需要大量精力去阅读，且容易误读。应将人类审查工作集中在低成本环节，让机器在输出可被机械验证的领域发挥作用。

创建policy_forge.py：

"""根据英文规则生成Rego策略，并仅在工具链验证通过时保留。"""

import argparse import pathlib import re import subprocess import sys import tempfile

WRITTEN, REJECTED, BROKEN = 0, 1, 2

SYSTEM = """你使用Rego v1（OPA 1.0+）编写Open Policy Agent策略。

规则：

• 每个规则体都要使用if，多值规则使用contains。
• 不要输出import rego.v1；在OPA 1.0+中这是冗余的。
• 输入是来自terraform show -json的JSON数据。
• 仅返回一个```rego代码块，不要添加其他内容。"""

def extract_rego(reply: str) -> str: blocks = re.findall(r"``rego\n(.*?)``", reply, re.DOTALL) if not blocks: raise ValueError("模型未返回任何rego代码块") if len(blocks) > 1: raise ValueError(f"模型返回了{len(blocks)}个rego代码块；应只返回一个") return blocks[0]

def verify(opa: str, policy: str, tests: pathlib.Path) -> tuple[bool, str]: with tempfile.TemporaryDirectory() as tmp: bundle = pathlib.Path(tmp)

测试用例保留原有文件名；策略文件使用不会与测试用例冲突的名称

(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和opa test均通过验证"

def forge(rule: str, tests: pathlib.Path, ask, opa: str, attempts: int) -> str: transcript = [{ "role": "user", "content": ( f"为这个规则编写Rego策略：\n\n{rule}\n\n" f"必须满足以下测试：\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 步中的策略：

尝试 1：FAIL - 加载时发生 2 个错误： 尝试 2：FAIL - policy/network_test.rego:63: 尝试 3：PASS - opa check 和 opa test 均通过

尝试 1 使用的是 Rego v0 的 deny[msg] { ... } 形式，这种语法在 OPA 1.0 于 2025 年 1 月发布时就停止使用了。而且公共训练数据中绝大多数内容都是这种形式。opa check --strict 在测试前就已拒绝了这种语法。

尝试 2 是有效的 Rego v1 语法。这种策略即使通过大多数工程师的审查，也只能在测试套件中获得 8 项中的 4 分。这是因为策略直接比较了 ingress.from_port 和 admin_ports，忽略了范围匹配，且只读取了 cidr_blocks。opa check 没有报错，因为代码结构完全正确。

结构良好的策略仍可能是错误的策略，而这个循环中唯一能理解你真实意图的只有你编写的测试套件。

修复循环不是单调递增的过程，因为每次尝试都是基于错误信息生成的新策略，没有任何机制能保留之前成功的部分，因此第 4 次尝试可能丢失第 3 次尝试拥有的特性。这种设计无法检测到这种问题，因为唯一被验证的是你编写的测试套件。

限制重试次数，持续扩展测试套件，并将每个生成的策略视为需要有人批准才能合并的 Pull Request。

## 第 8 步：治理代理本身

AI 代理也是一种参与者，它会调用工具，因此每次工具调用都必须经过某种授权决策。

行业很快达成共识：Amazon Bedrock AgentCore 策略在 2026 年 3 月达到通用可用性，通过 Cedar 在网关评估代理工具调用。常见的开源模式是在 MCP 工具网关前使用 OPA 侧车。

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

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

创建 agent/authz.rego：

METADATA

标题：代理工具调用授权

描述： |

在工具运行前每次调用都会评估一次。决策有三个值而不是两个，因为

一个值得部署的代理有时需要执行某些操作，这些操作应由人类而非策略批准。

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

未映射的角色、未知工具或格式错误的输入都会进入此分支。

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

# 无法读取金额的退款无法与上限进行核对，
# 因此会触发升级。此处保持静默将使攻击者
# 能够精准构造调用。
escalations contains reason if {
	input.tool == "issue_refund"
	not positive_amount
	reason := "退款金额缺失、不可读或不为正数"
}

# 负数金额是披着退款外衣的收费。
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(
		"%d 分的退款超过 %d 分的上限",
		[amount, refund_ceiling_cents],
	)
}

# 关键字匹配是一种粗粒度的防护机制，此处用于展示参数级规则的形态。任何包含真实数据的内容都需要 SQL 解析器：这能拦截 `DROP TABLE`，但会遗漏用其他方式拼写的相同语句。
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 := "语句包含破坏性 SQL 关键字"
}

决策词汇是封闭的，我将在此明确说明：

效果

调用者执行的操作

allow

执行工具

require_approval

暂停，向人类展示原因，仅在获得批准后执行

deny

拒绝，且不提供批准路径

之所以设计成这样，原因如下：

• 默认决策为 deny：未识别的工具、忘记映射的角色或格式错误的输入都会落入此情况。默认为 allow 的策略会在完全未预料到的输入上失效，而这些输入正是攻击者选择的目标。

• 三个取值：二元授权机制迫使在阻止有用工作与允许危险工作之间做出选择，而第三个取值使高自主性代理变得可接受。

• 原因以集合形式返回：所有适用的原因都会被收集。当某人在凌晨三点收到批准提示时，"250000 分的退款超过 10000 分的上限"会明确告诉他们该怎么做，而"策略违规"则无法提供有效信息。

• 参数也会被检查：在 £5 的退款是常规操作，而 £2500 的退款需要特别关注的情况下，仅根据工具名称进行粒度控制对代理来说过于粗糙，因为代理会自行选择参数。

十五个测试用例覆盖了决策表，包括具有空角色列表的代理、完全没有金额的退款以及将 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": [
    "250000 分的退款超过 10000 分的上限"
  ]
}

该策略在没有任何东西因响应而拒绝继续执行之前是惰性的。这就是客户端的实现方式：

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 会返回空对象和 200 状态码。拒绝关闭。
    decision = body.get("result", {"effect": "deny", "reasons": ["策略不可用"]})

    if decision["effect"] == "deny":
        raise PolicyDenied("; ".join(decision["reasons"]))
    if decision["effect"] == "require_approval":
        raise ApprovalRequired("; ".join(decision["reasons"]))
    return decision

search_orders    -> 允许
issue_refund     -> 需要审批（250000 分的退款超过 10000 分的上限）
delete_account   -> 拒绝（无匹配授权）

注意 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=对象
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 的测试用例，断言协议为 "-1" 的入站规则不会产生违规，理由是与空端口的比较不应导致策略崩溃。

所有协议规则会打开每个端口，AWS 提供商将其记录为 from_port: 0 和 to_port: 0。范围检查将其字面解释为单个端口 0，因此 AWS 中最宽松的规则得分干净。

该计划在 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，它将所有协议规则映射到完整范围，并对无法读取的端口返回未定义，同时添加第二个拒绝规则报告未定义的情况。该测试已被移除，取而代之的是四个测试：

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

[
  "aws_security_group.wide_open: 入站规则将端口 22 暴露给 0.0.0.0/0",
  "aws_security_group.wide_open: 入站规则将端口 3306 暴露给 0.0.0.0/0",
  "aws_security_group.wide_open: 入站规则将端口 3389 暴露给 0.0.0.0/0",
  "aws_security_group.wide_open: 入站规则将端口 5432 暴露给 0.0.0.0/0"
]

第 7 步认为测试套件是循环中唯一了解你意图的工件。这种属性的代价是：错误的测试即错误的规范，其下游内容不会提出异议。

实际的数字

两个版本在针对我设计策略的计划中得分均为 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 将任何无法解析到 apply 阶段的内容标记为 unknown，在 plan JSON 中显示为 null，同时附带一个 after_unknown 映射。当字段为 unknown 时，策略读取 change.after.some_field 不会触发。

这在跨资源规则中影响最大，因为 "每个存储桶都必须有公共访问限制" 在 plan 阶段确实很难实现，因为该限制引用了一个通常在 apply 后才知道的存储桶 ID。

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

任何在流水线之外应用的更改、控制台中修改的内容或自创建以来发生漂移的资源，对它来说都是不可见的。计划阶段检查和准入阶段检查都有不同的盲区，两者都需要定期扫描已部署状态来完成工作。

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

一百个绿色勾选并不能说明那些从未写过的规则。将你的控制要求映射到策略文件的对应关系保存在某个显式位置，并定期进行审计，因为门禁无法告诉你那些从未被询问过的问题。

4. CIDR 列表是精确匹配

public_cidrs 仅包含 0.0.0.0/0 和 ::/0，没有任何其他内容，因此一个打开 0.0.0.0/1 的规则会覆盖互联网的一半并通过。扩大范围意味着决定哪些前缀长度算作公共网络，并划出 RFC 1918 空间，这个决策属于你的组织。

5. 我发现的四种形态

暴露的集合覆盖了我寻找的四种编码方式，而 AWS 还提供了更多。安全组引用（security_groups）、前缀列表和自规则都是到达端口的方式，而这个策略并未建模这些方式，也从未看到它们。我预计在首次对大型环境运行时，会发现第五种形态。

6. 监管日期在变动

如果你正在为欧盟人工智能法案做准备，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 的判决结果。

• 观察到朴素版本发现了 9 个违规项中的 4 个，同时报告了 100% 的测试覆盖率，并通过加固使其发现了全部 9 个违规项。

• 构建了一个生成循环，由 opa check 和你的测试共同决定哪些内容写入磁盘。

• 通过同一个引擎授权代理工具调用，包含 15 个测试用例，默认策略为拒绝。

我的策略处理了我编写测试时覆盖的案例，但遗漏了两个未预料到的情况。所有可用信号（测试通过、100% 覆盖率、干净的 opa check）都表明它们是正确的。唯一持不同意见的是其他人编写的基础设施代码。请尽早将策略指向你未编写的代码，并持续保留失败案例。

所有代码、策略、计划 JSON 和生成图表的脚本都位于本书籍旁边的 code/ 目录中。图表可通过 python3 build/make_images.py 和 python3 build/make_terminals.py 重新生成。终端截图在构建时会重新运行命令，因此不会偏离事实。

合作伙伴推荐课程

(在新标签页中打开)

ADVERTISEMENT

• Coursera 上的《Implementing Open Policy Agent for Policy-Driven Control》Mark Peters 和 Starweaver 3周课程，每周1小时 中级 证书 查看课程 (在新标签页中打开)

• Udemy 上的《AWS Infrastructure as Code With Terraform [2024 Updates]》Edward Viaene 4.5（满分5星）（16,885人评分） 12小时40分钟 初级 证书 查看课程 (在新标签页中打开)

• Coursera 上的《IAC from Zero》Noah Gift 5周课程，每周1小时 初级 证书 查看课程 (在新标签页中打开)

• Udemy 上的《[NEW] HCP Terraform: A Crash Course for Beginners》Bryan Krausen 全球250,000+学员 4.7（满分5星）（18人评分） 3小时 初级 证书 查看课程 (在新标签页中打开)

更多 Terraform 课程请访问 Class Central

→

阅读更多文章

如果你读到这里，请感谢作者以表达你的支持。说声谢谢

免费学习编程。freeCodeCamp 的开源课程已帮助超过 40,000 人成为开发者。立即开始学习