如何利用“政策即代码”与开放隐私框架来管理由人工智能生成的基础设施【完整手册】

现代模型在近100%的情况下都能生成语法正确的代码。Veracode在2026年发布的报告明确指出：“语法问题已经基本得到解决。”

这听起来像是一个里程碑，但实际上正是这个原因导致了当前存在的问题。

同一份报告测试了上百种模型，发现它们的平均安全通过率仅为56%，“与第一份报告中的55%几乎没有变化”；大约44%的代码生成任务会引入安全隐患。

功能正确性与安全性其实是两个不同的问题，而目前只有其中一个问题接近得到解决。

这个结果既不是个特例，也不是什么新鲜事。2022年，在IEEE安全与隐私会议上，纽约大学坦登分校的一个团队让GitHub Copilot在89种与安全相关的场景中运行，生成了1,689份程序，结果发现其中大约40%的程序存在MITRE CWE Top 25列表中的安全漏洞。该研究论文后来还被选为《ACM通讯》的研究亮点。

2024年11月，乔治城大学的安全与新兴技术中心评估了五种大型语言模型，发现它们生成的代码中有一半以上包含可能导致被利用的漏洞。四年时间里，四个不同的团队使用四种不同的方法进行测试，但得出的结果都是一样的。

2023年在ACM CCS会议上，斯坦福大学的Neil Perry、Megha Srivastava、Deepak Kumar和Dan Boneh让开发者们参与了实验：共有47名参与者参与了五项与安全相关的编程任务，使用三种不同的编程语言进行测试。其中33名参与者使用了人工智能辅助工具，而14名没有使用。结果发现，使用辅助工具的团队生成的代码安全性明显更低，而且他们也更倾向于认为自己编写的代码是安全的。

这项研究规模较小，但它解释了为什么这个问题无法自行得到解决：那些通常能够检测出这些问题的机制，恰恰是被这些工具所屏蔽掉的。

所有这些研究都是针对应用程序代码进行的。但基础设施代码的情况更为复杂，因为那些存在安全漏洞的代码系统永远不会出错——它们会按照编写的方式正常运行，为任何请求者提供服务；唯一会对此提出异议的人，也只有那些仔细检查代码差异的人而已。

我无法通过阅读的方式来审查那么大量的代码，其他人也同样做不到。但我可以制定一些规则，让计算机在每次代码发生变化时都能自动进行检查，这就是“政策即代码”这一概念的含义。

在这本手册中，我会指导你如何构建这样的检查机制。我们会以一个存在安全漏洞的代码仓库为例，观察这个代码在经过修改后是如何消除其中存在的9个安全隐患中的5个的，然后一起修复这些漏洞。

学习完这本手册后，你将能够做到以下几件事：

• 根据`terraform show -json`命令输出的JSON格式，编写Rego政策规则。

• 像测试应用程序代码一样，使用测试用例和覆盖率报告来测试这些政策规则。

• 构建一个命令行工具，使其能够根据预设的退出码契约进行正确运行，从而让持续集成管道可以信赖这个工具。

• 在Kubernetes系统中，利用CEL机制在代码提交时阻止不符合安全要求的作业被执行。

• 让模型生成政策规则，然后通过`opa check`工具以及你自己的测试来决定这些规则是否应该被保留。

• 使用同一个政策引擎来授权人工智能代理执行相应的操作。

目录

• 先决条件

• 关键术语详解

• 步骤1：获取用于测试的真实基础设施

• 步骤2：在制定政策之前编写测试用例

• 步骤3：持续修改政策直到测试通过

• 步骤4：将测试结果应用于实际方案中

• 步骤5：将测试结果转化为退出代码

• 步骤6：在系统启动时执行相关规则

• 步骤7：让模型自动生成政策

• 步骤8：让代理系统自我管理

• 步骤9：我犯下的错误

• 检查的局限性

• 结论

先决条件

您需要准备以下工具：

• 一个终端以及可正常运行的 python3（版本3.10或更高）。

• jq工具，用于在命令行中解析JSON数据。

• 大约700MB的磁盘空间，因为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。

我在macOS系统上使用OPA 1.20.2、Terraform 1.14.2以及AWS provider 6.x进行了测试。OPA 1.x系列版本之间的政策语法格式是兼容的。

如果您使用的是OPA 0.x系列版本，那么每个规则都需要在文件开头添加import rego.v1这一行代码，因此我建议您升级到较新的版本。违规次数的计算仅与AWS提供商的版本有关，但这种关联性仅体现在计划JSON数据的结构上，而自AWS provider 5版本以来，这一结构一直保持稳定。

关键术语详解

• “政策即代码”：指你的组织已经达成一致的各项规则，这些规则被编写成程序形式，用于评估某项拟议的变更，并据此作出决策。

• Rego：Open Policy Agent所使用的查询语言。这种语言属于声明性语言：规则的具体内容由一系列必须同时满足的条件组成。

• Plan JSON：Terraform执行操作前会生成的一种机器可读的描述文件，该文件可通过`terraform show -json`命令获取。政策规则正是根据这份描述文件来做出决策的，因此你的`.tf`文件本身并不会被直接用于这些决策过程。

• 准入控制：Kubernetes API服务器中的某个环节，在对象被存储之前，该环节会判断是否允许其通过。

• 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" {
  # 这个安全组允许所有设备通过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"
    cidrBlocks = [
    "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"
    cidrBlocks = [
    "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 } # 仅通过检查“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 } # 如果规则中设置了“-1”作为协议类型，那么它会允许所有端口被访问。 # 第一个版本的这个测试得出了相反的结论，从而掩盖了这个错误。 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 }

以下是该文件所编码的内容：

1. with input as fixture这一代码会将某个表达式替换为虚假的配置，这样就可以在没有云账户的情况下测试相关策略了。

2. test_denies_wide_open_port_range这一测试预期会检测到四项违规行为，因为每个管理端口都会对应一项违规记录；而一条0-65535范围的规则实际上与明确的22号端口规则具有相同的效果，但这种写法会绕过匹配检查机制。

3. 还有三项测试分别针对Terraform使用的另外三种表达方式来验证这一逻辑：IPv6字段、现代独立的aws_vpc_security_group_ingress_rule规则，以及已被弃用的aws_security_group_rule规则。这些测试并不是我最初编写的，第9步会解释它们的来源。

4. 那两个test_allows_测试与那些拒绝某些行为的测试同样重要。如果一个策略会拒绝所有请求，那么它虽然能通过所有的拒绝测试，但实际上毫无意义。

5. 最后三项测试是在策略编写完成后才添加的，第9步也会解释它们的用途。一条允许所有协议的规则会打开所有端口，而那些该策略无法识别的端口必须被报告出来。

步骤3：编写策略直到测试通过为止

Terraform使用了四种不同的方式来描述入站规则，因此我们需要将这四种方式统一成一种格式，然后再进行一次验证。请创建policy/network.rego文件：

# METADATA
# 标题：禁止从公共互联网访问任何管理端口
# 描述：
#   Terraform使用了四种不同的方式来描述入站规则。首先，这些规则都会被统一成一种格式，
#   因此后面的验证代码只需要编写一次，对于每种新的规则格式，也只需要增加一条辅助规则即可。
#
# 这里有两点是故意设计的，并非偶然的结果：一条允许所有协议的规则会打开所有端口，
# 而那些该策略无法识别的端口必须被报告出来，而不会被视为合法规则。
package terraform.network

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

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

# AWS提供者规定，对于允许所有协议的规则，`from_port`和`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_cidrBlocks"]
	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：AWS提供者从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,
	}
}

# 方式4：大多数现有系统中仍在使用的废弃独立规则格式。
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,
	}
}

# 一条规则实际覆盖的端口范围。如果政策无法确定这个范围，那么这些值就会被设置为未定义。
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

	# 如果某个端口落在[from_port, to_port]这个范围内，那么这条规则就覆盖了这个端口。
	range := covered_ports(exposure)
	some port in admin_ports
	port >= range[0]
.port <= range[1]

(msg := sprintf(
		"%s: 入站规则允许端口 %d 被访问到地址 %s",
		[exposure.address, port, exposure.cidr],
	)
}

# 如果有一条规则允许所有端口被访问，但该策略无法识别其中的一些端口，那么这条规则就需要被报告出来。
# 通过这种测试，就可以判断出政策是否在过于宽松的方向上进行了设置。
deny contains msg if {
	some exposure in exposures
/exposure.cidr in public_cidrs
—not covered_ports(exposure)

(msg := sprintf(
		"%s: 入站规则允许访问的端口范围是 %s，但该策略无法识别其中的一些端口 (%v to %v)",
		[exposure.address, exposure.cidr, exposure.from_port, exposure.to_port],
	)
}

# 这些地址是该策略能够识别的。通过检查这些地址，就可以判断出“是否违反了任何规则”。
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",
	}
}

从上到下阅读这些内容：

1. 在Rego中，规则主体实际上是一个逻辑与操作。每一行规则都必须成立，而some ... in这种写法会遍历多行规则，因此OPA会尝试所有资源、入口规则、字段以及端口的组合。

2. 三条独立的exposures规则共同定义了一组规则集，Rego将这种写法称为“增量式定义”。如果后来再添加第五条规则，只需要多消耗一个代码块，而不会影响下面的规则内容。

3. 当某个字段不存在时，object.get(ingress, field, [])会返回一个空列表；因此，对于仅针对IPv4协议的规则来说，当策略中查找ipv6_cidr_blocks时也不会出现错误。

4. covered_ports起到了“安全阀”的作用——因为一条涵盖所有协议的规则在生成计划时会将结果设置为0，但同时会打开所有的端口；因此不能直接将端口号视为字符串来处理。如果某条规则的端口字段值为空或字符串形式，那么该规则的功能就会变得不明确，而第二条deny规则正好可以用来检测这种情况。

5. 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代码，这两种工具都属于持续集成流程中必不可少的部分。

第二条规则是用于管理资源的所有权标签设置，因此需要创建文件policy-tags.rego：

# METADATA
# 标题：所有被管理的资源都会携带所有权标签
# 说明：
#   Terraform会为没有标签的资源生成`tags: null`，因此一条规则如果访问`after_tags`字段，就会跳过那些标签信息最差的资源。tags_of函数会将这些空的标签对象转换为空字典。
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
}

下面解释一下为什么这两行代码会写成这样的形式：

1. trim_space(value) == ""这一检查是必要的。因为在Rego中，只有false和未定义的值才被视为错误值，而空字符串则被视为有效值。因此，简单的存在性检查会接受owner = ""这样的配置，而这恰恰就是标签策略所要防止的情况。

2. tags_of以及其中的else := {}分支，其实是用来修复我之前编写的一个漏洞的，而这个漏洞直到第9步测试时才被发现。

步骤4：将检查结果与实际计划进行对比

需要根据TerraGoat的计划来评估各个包是否符合要求：

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

[
  "aws_security_group.web-node: 入站规则将端口22开放给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: 入站规则将端口22开放给0.0.0.0/0",
  "aws.security_group.web-node: 缺少必要的标签\"cost-center\"",
  "aws_security_group.web-node: 缺少必要的标签\"data-classification\"",
  "aws_security_group.web-node: 缺少必要的标签\"owner\"",
  "aws_vpc.web_vpc: 缺少必要的标签\"cost-center\"",
  "aws_vpc.web_vpc: 缺少必要的标签\"data-classification\"",
  "aws_vpc.web_vpc: 缺少必要的标签\"owner\""
]

{v | v := data.terraform[_].deny}这一表达式的作用是从data.terraform下的所有包中收集所有的违规信息，然后使用union操作将这些信息合并起来。如果在这个命名空间中添加新的政策文件，系统也会自动识别这些新文件，而无需修改原有的命令。

步骤5：将检查结果转换为退出代码

持续集成工具是通过退出状态来传递反馈信息的；如果将两种不同类型的错误都用同一个代码来表示，那么出现问题的流程就可能被忽略长达一个月之久。这个工具使用以下编码规则：

代码	含义
0	通过	所有策略都已执行，检查了相关资源，未发现任何问题
1	失败	策略已执行，但发现了违规情况
2	无效结果	策略虽然执行了，但没有检查任何内容，因此检查结果没有实际意义
2	系统故障	工具本身或其输入数据无法正常使用

第四行是那些最容易被错误处理的行。如果某个计划中包含了任何策略所不了解的资源，那么报告“通过”就意味着该计划实际上无法成功执行；这类情况会被赋予专门的判断名称，并且系统不会将结果标记为“0”。

创建文件 `policy_gate.py`：

"""根据一系列 Rego 政策文件来评估 Terraform 计划。"""
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 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

以下是该脚本的功能说明：

1. argparse将--plan和--policy标记为“必需参数”，且允许接受多个参数，因此如果策略列表为空，在解析时就会引发错误。

2. load_plan会报告JSON语法错误发生的行号和列号，因为json.JSONDecodeError会包含lineno和colno这两个信息；而那些仅显示“无效JSON”信息的提示其实毫无意义，只会浪费人们的宝贵时间。

3. 所有失败路径都会经过die函数处理，这个函数总会导致程序退出。例如，缺少opa文件、文件无法读取，或者计划中没有resource_changes键，这些情况都属于工具本身的故障。

4. considered查询会在deny查询之前执行。如果没有任何错误被检测到，程序的执行结果将会是“VACUOUS”，也就不会输出任何通过的结果。

5. 违规项会被进行排序，这样同一个计划在不同次运行时产生的输出结果才会完全一致；而如果两个CI日志之间存在差异，那么这种差异肯定意味着某些地方存在问题。

这个计划中的2个资源共存在7处违规项，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图表，更不会识别操作员按照自己的计划创建的Pod。对于这些情况，你需要使用准入控制机制，而Kubernetes现在已经内置了这种功能。

ValidatingAdmissionPolicy这一功能从v1.30版本开始就已经正式提供了。它会在API服务器内部对相关规则进行验证，而不需要任何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: >-
        '所有图像必须来自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、Deutsche Telekom、LinkedIn以及Spotify等机构都在使用它进行生产环境部署。它的配置规则是用YAML格式编写的，因此开发团队无需学习新的编程语言；同时，该引擎能够完成生成配置文件、验证图像签名以及清理残留数据等功能——这些都是内置规则所无法处理的。

OPA Gatekeeper正是当你需要一个能够同时覆盖Kubernetes、Terraform以及持续集成流程的配置管理工具时，理想的选择。目前该工具也已经具备了代码生成功能：MutatingAdmissionPolicy在v1.36版本中就已经实现了这一功能。

步骤7：让模型来编写配置规则

配置规则的编写工作往往非常繁琐，而模型恰恰擅长处理这类重复性任务。因此，让模型来生成配置规则显然是一个明智的选择。

不过这里有一个需要注意的问题——你可以在大约一分钟内自己验证这一点。在这一步的最后，我也会进行这样的测试：在公共训练数据中，有很大一部分Rego配置规则实际上是Rego v0版本的语言规范，而这种语言规范在2025年1月OPA 1.0版本发布后就已经不再被解析器支持了。因此，如果让模型去生成最常见的配置模式，它很可能会产生当前解析器无法识别的代码。

卡利亚布里亚大学的一个研究团队在2025年发表的一篇预印本论文ARPaCCino中，通过对Terraform配置案例的研究得出了同样的结论。在没有任何辅助工具的情况下，Qwen3-30B和GPT-4o这两个模型分别生成了0份语法正确的配置规则；而在引入OPA文档信息后，这一数字也没有发生变化。然而，当给模型添加能够执行opa check命令并读取错误信息的功能后，正确生成的配置规则数量分别增加到了4份和5份。

不过这些数据仅仅来自一个小型案例研究，因此应该将这个结果视为一个初步的方向指示。

最终解决这个问题的是一个反馈循环机制，而且构建这样的循环并不需要额外的成本，因为你本来就可以使用opa check --strict、opa fmt和opa test这些工具来构建它。

因此，你应该通过加入一个能够确保结果准确性的验证环节来构建这个循环。具体来说，就是我来编写测试用例，而模型则负责生成配置规则。这样的测试方案非常直观，也便于审核——因为你只需要阅读一段JSON格式的代码，三秒钟内就能判断它是否正确。相比之下，使用嵌套列表结构来表示配置规则不仅阅读起来很麻烦，而且也很容易出错。因此，应该把那些需要人工进行仔细审核的工作交给人类来完成，而把那些可以通过机器进行自动验证的任务交给计算机来处理。

创建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格式来编写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("输入数据中包含了多个```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 forgerule: 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()), "工具链没有返回任何结果") print(f"第{attempt}次尝试：{'PASS' if ok else 'FAIL'} - {headline}", file=sys.stderr) if ok: return policy transcript += [ {"role": "assistant", "content": reply}, {"role": "user", "content": f"工具链拒绝了该策略，原因如下:\n\n{output}\n\n请修改后重新尝试。」}, ] raise RuntimeError(f"经过{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="策略内容，需用一句英文句子来表述") parser.add_argument("--tests", required=True, type=pathlib.Path, help="您手动编写的_test.rego文件") parser.add_argument("--out", required=True, type

export ANTHROPIC_API_KEY=...
python3 policy_forge.py \
  --rule "任何安全组都不得将管理端口暴露给公共互联网。" \
  --tests policy/network_test.rego \
  --out policy/generated.rego

这个循环能够确保以下几点：

1. 验证过程以子进程的形式执行：模型从未被要求确认自己的策略是否正确。最终由opa check和opa test来做出判断，而它们的退出代码才是该循环所接受的唯一依据。

2. 失败情况会以原始工具输出的形式被记录下来，而不会被进行任何总结：编译错误和测试失败是模型能够接收到的最重要反馈信息，如果对这些信息进行重新表述，就会失去其中有用的部分。

3. 只有当策略通过验证后才会被保存到磁盘上：forge要么返回经过验证的策略，要么抛出错误，因此不存在因为重试次数用完而导致未经过验证的策略被保存到仓库中的情况。

我使用了一个脚本化的模型来运行这个循环，因此即使没有API密钥，也能得到可重复的结果。实验中使用了三种策略：一种是符合v0语法的真实策略，另一种是语法正确但内容错误的v1策略，还有第三种就是步骤3中生成的策略。

尝试1：失败——在加载过程中出现了2个错误；
尝试2：失败——policy/network_test.rego文件的第63行存在问题；
尝试3：成功——opa check和opa test都通过了验证。

尝试1使用的是Reno v0语法，即deny[msg] { ... }这种写法，在2025年1月OPA 1.0版本发布后就已经不再被支持了。而且，大多数公共训练数据中使用的也都是这种语法。opa check --strict在测试开始之前就拒绝了这种格式的策略。

尝试2使用的是合法的Reno v1语法。这种策略应该能够通过大多数工程师的审核，但在整个测试套件中只获得了4分（满分8分）。这是因为该策略直接将ingress.from_port与admin_ports进行了比较，忽略了范围设置，并且只考虑了cidr_blocks这一因素。opa check并没有发现任何问题，因为代码的形式是完全正确的。

即使代码格式正确，策略也可能是错误的。在这个循环中，唯一能够判断策略是否正确的就是你编写的测试套件。

修复流程并不是单调的，因为每次尝试都是根据上一次出现的错误信息来重新生成的，之前的成功经验并不会被保留下来，因此第四次尝试可能会失去第三次尝试中获得的结果。但这种设计本身并没有检测机制来发现这一点，因为唯一被检查的对象就是你编写的测试套件。

应该为重试次数设定上限，让测试套件不断完善，并且将每次生成的策略都视作一个需要他人审核的“拉取请求”，只有获得批准后才能被合并到最终版本中。

步骤8：让代理自我管理

人工智能代理也是一种执行者，它会调用各种工具，因此每一次工具调用实际上都代表着一个授权决策，需要由该代理来做出。

业界很快就达成了这种共识：Amazon Bedrock AgentCore Policy在2026年3月正式投入使用，在Cedar平台上，该系统会在代理调用工具时对其进行验证。目前常见的开源实现方式是在MCP工具网关之前添加一个OPA侧车组件。

相关研究正在不断突破这些界限：华盛顿大学团队在2026年发布了一篇预印本论文，题为《利用Z3求解器验证工具增强型大语言模型的策略合规性》“利用求解器辅助验证工具增强型大语言模型的策略合规性”（作者为Winston、Winston和Just），该论文将自然语言制定的策略转化为SMT约束条件，然后通过Z3求解器来阻止那些不符合规定的操作。

他们得出了一个共同结论：系统提示中给出的策略本身并不能起到强制执行的作用。强制执行机制实际上是一种“拦截器”，它位于调用路径中，能够直接返回“拒绝”信号，从而阻止相关操作的执行。

下面是`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 countreasons > 0

escalations contains reason if {
 输入的工具属于write_tools集合
	且当前会话不是由人类操作的
	reason := sprintf("%q会修改系统状态，而当前会话无人值守", [输入的工具])
}

# 如果退款金额无法被正确识别，或者超过了规定的上限，就会触发升级处理。
escalations contains reason if {
	input.tool == "issue_refund"
	且退款金额既不是正数，也无法被识别
	reason := "退款金额缺失、无法识别或为负数"
}

# 如果输入的工具是“issue_refund”，但退款金额为负数，那就相当于在收取费用。
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命令"
}

这些决策规则是固定的，我现在就明确说明如下：

处理结果	调用者应执行的操作
允许	运行该工具
需要审批	暂停执行，将原因展示给人类用户，只有获得批准后才能继续运行
拒绝	直接拒绝请求，不提供任何审批途径

制定这样的规则是有原因的：

1. 默认处理结果是“拒绝”：对于那些未被识别的工具、你忘记配置的角色，或者输入格式不正确的情况，系统都会默认选择拒绝。如果默认设置为允许，那么在那些攻击者会选择的特殊情况下，系统就会错误地允许操作执行，从而带来安全风险。

2. 三种处理方式：二进制授权机制迫使人们在阻止有用操作与允许危险操作之间做出选择；而第三种处理方式则使得高自主性的智能体能够被合理使用。

3. 原因会以集合的形式返回：系统会收集所有适用的原因。例如，当有人在凌晨三点收到审批提示时，“250000美分的退款金额超过了10000美分的上限”这个信息就能告诉他们应该怎么做；而“违反政策”这样的表述则无法起到同样的指导作用。

4. 也会检查提供的参数：对于issue_refund这种操作，5英镑以下的请求属于常规操作，而2500英镑以上的请求则属于严重操作；因此，如果使用工具名称作为分类依据，对于智能体来说就太粗糙了，因为最终是智能体自己来选择需要使用的参数。

共有15项测试覆盖了这个决策规则表，其中包括那些角色列表为空的智能体、请求退款金额为零的情况，以及那些在请求中隐藏了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会返回一个包含“effect”: “deny”和“reasons”: [“policy unavailable”]的字典。此时应拒绝请求。
    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 amount of 250,000 cents exceeds the limit of 10,000 cents)
delete_account => DENIED (no matching grant available)

注意：body.get("result", ...)：当查询没有找到任何匹配结果时，OPA会返回一个状态码为200的空对象{}，因此直接使用body["result"]会导致KeyError。此外，你的代理框架在处理异常时也可能会遇到问题。默认情况下，所有环节都会拒绝请求，包括解析阶段。

请在工具函数执行之前，通过框架的工具执行钩子调用authorize()函数。这个函数的代码长度大约为15行，它的作用是将系统提示中提供的建议转化为实际的规则限制。

步骤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'

测试结果如下：没有违规行为发生，所有端口都正常开放，测试程序会输出“PASS”并结束执行。

一个导致安全漏洞存在的测试案例

不过这个配置方案还存在另一个问题——我的测试套件本身也在保护这些端口。我编写了一个名为`test_tolerates_null_ports`的测试用例，用来验证当使用`protocol: "-1"`这样的规则时是否会出现违规行为。我认为，与一个空值进行比较不应该导致策略执行失败。

实际上，任何类型的协议都会被这种规则允许通过；AWS系统会将这些规则的`from_port`和`to_port`字段都设置为0。而范围检查机制会将其理解为仅允许端口0通过，因此这种最宽松的配置方案在测试中通过了验证。

相关的配置文件保存在`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'

测试结果显示：所有协议和所有端口都被允许开放，我的测试也验证了这一配置是正确的。为了解决这个问题，我们添加了`covered_ports`这个规则：它将“允许所有协议通过”的设置应用到所有端口上；对于那些无法识别的端口，它会将其设置为默认值“未定义”，并再添加一条`deny`规则来处理这些情况。原来的那个测试用例已经被移除，取而代之的是四个新的测试用例：

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个测试案例，并且正确地识别出了那些没有任何标签的资源。

“覆盖率”实际上只是反映了你的测试代码执行了哪些部分，但并不能说明你是否考虑到了所有可能的情况。对于安全规则来说，这意味着你的规则是否能够有效地处理所有可能的故障情况。虽然报告覆盖率是有意义的，但只有在实际运行这些规则来检测你并未参与的基础设施时，才能真正证明这些规则是有效的。

这项检查的局限性

以下是一些这项检查仍然无法解决的问题，因为那些错误地认为规则是有效的情况，其影响往往比那些真正被检测到的违规行为更为严重。

1. 未知值是无法被检测到的

Terraform会将那些在应用之前无法确定的值标记为“未知”，在配置方案的JSON格式中，这些值会以null的形式出现，并且还会伴随着一个after_unknown映射。因此，如果规则中包含了像change.after.some_field这样的表达式，当该字段的值未知时，这个规则就不会被触发。

这一问题在处理跨资源的规则时表现得最为明显。例如，在配置方案生成阶段，我们很难确定“每个存储桶都应该设置公共访问限制”这一要求是否成立，因为其中涉及的存储桶ID通常是在应用规则之后才能确定的。

2. 它只能看到配置方案本身

任何在配置流程之外进行的修改、通过控制台进行的更改，或者自配置方案生成以来发生的变动，都会被这项检查忽略。配置方案生成阶段的检测和资源应用后的检测存在不同的盲点，因此还需要定期对已部署的资源状态进行再次检查。

3. 控制的覆盖范围无法从内部进行测量

即便有上百个绿色的勾选标记，也无法说明那些根本没有人制定过的规则究竟是否得到了遵守。请将您的控制要求与政策文件之间的对应关系明确记录下来，并定期对其进行审核，因为系统本身无从知晓哪些内容从未被要求过进行检查。

4. CIDR列表能够实现精确匹配

public_cidrs中仅包含0.0.0.0/0和::/0这两种地址格式，因此任何允许访问0.0.0.0/1的规则实际上都会覆盖一半的互联网流量，从而通过审核。但要想进一步扩展这种规则的适用范围，就必须决定哪些前缀长度属于公共网络范围，以及如何划分RFC 1918规定的地址空间——而这些决策权属于您的组织。

5. 我发现的情况其实是四种不同的形式

exposures这个集合涵盖了我所寻找的四种编码方式，而AWS还提供了其他更多的选项。安全组引用、前缀列表以及self规则等都是可以用来访问某些端口的手段，但这些在当前的策略模型中并未被考虑在内，因此系统也无法识别这些途径。我认为，在对规模较大的网络环境进行检测时，很可能会出现第五种不同的情况。

6. 监管规定的日期会发生变化

如果您正在按照欧盟的AI法案来构建自己的系统，那么2026年7月发布的《数字综合法规》将附件III中规定的高风险义务的实施期限从2026年8月2日推迟到了2027年12月2日，而附件I中的义务实施期限则仍为2026年8月2日；至于第50条规定的透明度相关义务，其实施日期依然保持不变。

请将所有的控制措施都编码出来，并且每次需要确认日期时，都去官方时间表中核对一下，即使上季度的某篇博客文章给出了不同的信息，也一定要以官方时间表为准。

结论

模型生成的基础设施代码几乎每次都能被正确解析，但在大约56%的情况下这些代码并不能确保系统的安全性；而且生成这些代码的速度远远快于人们阅读它们的速度。正是在这种速度与安全性的矛盾中，人工审核逐渐不再是一种有效的控制手段了。规则本来就应该是可以被直接执行的，而大量的规则使得这个问题变得愈发突出。

通过本教程，您已经完成了以下任务：

• 从TerraGoat中提取了一个存在安全漏洞的安全组，并使用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命令来重新生成这些图表。而用于生成终端截图的脚本会在构建过程中再次执行相应的命令，因此这些截图所显示的结果肯定是与实际情况相符的。