
用 Policy as Code 治理生成式基础设施:OPA、Rego 与准入控制实战
用 Policy as Code 治理生成式基础设施:OPA、Rego 与准入控制实战
模型生成的代码语法几乎不会出错,但安全性远未解决,基础设施代码尤其危险——一个错误的安全组不会报错,只会照常放行流量。本教程用 Open Policy Agent 与 Rego 把组织规则写成可执行程序,从真实漏洞仓库出发,覆盖测试先行、计划 JSON 校验、退出码契约、Kubernetes 准入控制与 AI 代理工具调用授权。
现代模型生成的代码在语法层面几乎不会出错,但功能正确与安全合规是两件不同的事,而后者远未解决。多项独立研究反复给出同一个结论:生成代码中相当比例带有可被利用的缺陷。基础设施代码是更难的一类,因为一个配置错误的安全组不会失败——它会完全按写下的内容工作,向任何请求者提供服务,唯一会提出异议的是读 diff 的人。人读不过来这么大的量,所以可行的办法是把规则写成计算机能在每次变更时检查的形式,这就是 Policy as Code。本教程带你从零搭起这套检查:指向一个真实的漏洞仓库,观察最朴素的策略版本如何漏掉摆在眼前的九条违规中的五条,然后把它修好。
读完之后你会掌握:针对 terraform show -json 产出的 JSON 编写 Rego 策略;像测试应用代码一样测试策略,带夹具与覆盖率报告;构建一个 CI 流水线可以信任的、有明确退出码契约的命令行门禁;在 Kubernetes 准入阶段用 CEL 拦截不合规工作负载;让模型写策略,再由 opa check 和你自己的测试决定是否采纳;以及用同一个策略引擎为 AI 代理的工具调用做授权。
准备工作
开始之前先确认环境。你需要:
- 一个终端,以及可用的
python3(3.10 或更新版本)。 jq,用于在命令行读取 JSON。- 约 700 MB 磁盘空间,因为 AWS Terraform provider 体积较大。
- 一个 Anthropic API key,但只有第七步需要;其余所有步骤都可以离线运行。
初始化工作目录并安装工具:
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/latest/opa_linux_amd64_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_linux_amd64.zip
unzip tf.zip && sudo mv terraform /usr/local/bin/在 Windows 上,激活虚拟环境用 .venv\Scripts\activate,并把两个下载地址换成对应的 Windows 版本文件。
本教程的示例在 OPA 1.20.2、Terraform 1.14.2 与 AWS provider 6.x 上运行。策略语法在 OPA 1.x 内是稳定的;如果你还在 OPA 0.x,每条规则都需要在顶部加 import rego.v1,更推荐直接升级。违规数量对 AWS provider 版本的依赖只体现在计划 JSON 的结构上,而该结构自 provider 5 起保持稳定。具体版本号与下载地址请以各项目官网当前信息为准。
几个关键术语
- Policy as Code:把组织已经达成共识的规则写成一个程序,输入是提议的变更,输出是一个判定。
- Rego:Open Policy Agent 求值的查询语言。它是声明式的,规则体是一组必须同时成立的条件。
- Plan JSON:Terraform 即将执行的操作的机器可读描述,由
terraform show -json生成。策略读的是它,所以.tf文件本身从不进入策略。 - 准入控制:Kubernetes API server 内部的一个环节,对象在被持久化之前可以在此被拒绝。
- CEL:Common Expression Language,Kubernetes 在 API server 内原生求值的小型表达式语言,不需要部署 webhook。
- 误放行:策略本应拦下却放过的资源。没有人会注意到它,所以除非主动去找,它不会被度量。
同一个判定会出现在三个位置,只有中间那个是 Kubernetes 特有的。
操作步骤
第一步:取一份真实的、有漏洞的基础设施代码
不要自己编一个漏洞 Terraform 文件,因为编造等于编造 bug,策略最终只会抓到你亲手埋下的那一个。去找别人已经写出来并公开的代码。TerraGoat 是一个刻意做成有漏洞的 Terraform 仓库,由 Bridgecrew 发布。取它固定提交上的 EC2 模块:
SHA=729f8da62c6a85ce4af5ad3d123de97776d954c4
curl -s "https://raw.githubusercontent.com/bridgecrewio/terragoat/$SHA/terraform/aws/ec2.tf" \
| sed -n '77,96p'你会看到一段安全组定义,注释里写着「安全组对全世界开放 SSH 端口」,而端口 22 对 0.0.0.0/0 开放正是它指向的问题。
TerraGoat 的模块在现代 Terraform 上无法初始化,因为它仍然用引号声明 type = "string",这种写法在 Terraform 0.12 被弃用、在 1.x 被拒绝。所以把这个资源搬进一个最小模块,只替换掉对 TerraGoat 内部 local 的两处引用。创建 main.tf:
terraform {
required_version = ">= 1.9"
required_providers {
aws = {
source = "hashicorp/aws"
version = "~> 6.0"
}
}
}
# 模拟凭据。这份配置只会被 plan,永远不会被 apply,
# 所以 provider 不应尝试访问 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"
}
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 及其三个同类选项,provider 不会尝试认证,也不会有任何东西被真正应用。
第二步:先写测试,再写策略
想要的规则是:任何安全组都不得把管理端口暴露到公网。听起来像一行代码,而测试正是用来钉死它为什么不是一行的地方。创建 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
}
# 全协议规则会打开所有端口,无论它的端口字段写的是什么。
# 这个测试的第一版断言了相反的结果,从而把 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
}
# 策略读不懂的端口要上报,绝不放过。
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
}这些测试覆盖了几类容易被忽略的写法:端口范围而非单端口、IPv6 的 ::/0、独立的 aws_vpc_security_group_ingress_rule 资源、已弃用但仍可能出现的 aws_security_group_rule、全协议规则,以及端口字段为 null 或字符串这类策略读不懂的情况。最后一条尤其重要:读不懂的端口必须上报,而不是默认放过。
第三步:写策略直到测试通过
策略文件放在 policy/network.rego,包名与测试中的 data.terraform.network 对应。核心思路是:把计划 JSON 里所有携带入站规则语义的资源类型都归一化成统一的中间结构,再对归一化后的规则做判断。这样新增一种资源类型时,只需要扩展归一化部分,判定逻辑不用动。
判定要覆盖三种「对世界开放」的写法:0.0.0.0/0、::/0,以及协议为 -1 的全协议规则。管理端口集合按你的组织约定定义,通常至少包含 22 与 3389。端口比较必须用范围相交而不是相等,否则 from_port = 0, to_port = 65535 会直接绕过检查。
写完后运行:
opa test policy/ -v
opa check policy/opa test 会跑完所有 test_ 前缀的规则并给出通过情况,opa check 做语法与静态检查。覆盖率报告可以这样拿到:
opa test policy/ --coverage --format=json | jq '.files | keys'覆盖率的意义在于找出策略里从未被任何测试触达的分支——那些分支就是误放行的温床。
第四步:把策略指向真实的计划
测试通过之后,用真实的 plan.json 求值:
opa eval \
--data policy/ \
--input plan.json \
--format pretty \
'data.terraform.network.deny'此时会看到违规列表。最朴素的策略版本在这里只会报出九条违规中的四条,剩下五条来自它没有覆盖的资源类型与写法——这正是第二步那些测试存在的理由。逐条对照测试用例,把归一化逻辑补齐,直到真实计划上的输出与预期一致。
第五步:把判定变成退出码
CI 流水线不读 JSON,它读退出码。所以需要一个薄封装,把策略判定翻译成明确的契约:
- 退出码 0:没有违规,放行。
- 退出码 1:存在违规,阻断。
- 退出码 2:策略本身出错或输入无法解析,视为失败而非放行。
第三种情况最容易被忽略。如果策略因为输入格式变化而崩溃,默认行为绝不能是「没有输出即视为通过」。把「无法判定」和「判定为通过」区分开,是这套门禁能被信任的前提。
封装脚本用 opa eval 取 deny 集合,用 jq 判断长度,再按上面的契约返回退出码。在 CI 里,这一步放在 terraform plan 之后、terraform apply 之前。
第六步:在准入阶段拦截
计划阶段的检查只能覆盖走 Terraform 的路径。集群里直接创建的对象需要在 API server 内拦截,这时用 CEL 而不是 webhook:不需要部署额外服务,没有网络往返,也不会因为 webhook 不可用而让整个集群的写入失败。
把同一条规则翻译成 CEL 表达式,写进准入策略。判定逻辑与 Rego 版本保持一致:识别对世界开放的入站规则、识别全协议规则、用范围相交判断管理端口。两处逻辑不一致是常见的坑——计划阶段拦下的东西在准入阶段被放过,或者反过来,都会让人不再信任这套检查。
第七步:让模型写策略,由检查决定去留
这一步需要 API key。做法是让模型根据自然语言规则生成 Rego,然后:
- 运行
opa check,语法或静态检查不过直接丢弃。 - 运行你自己的测试套件,任何一条不通过就丢弃。
- 查看覆盖率,如果新策略让覆盖率下降,说明它引入了未被测试的分支,同样丢弃。
关键在于:模型的输出不进入评审流程,而是进入检查流程。人不需要读它写得对不对,只需要看检查过不过。这也意味着测试套件必须先于模型生成存在——先有测试,模型才有可被验证的目标。
第八步:治理代理本身
同一个策略引擎可以用来给 AI 代理的工具调用做授权。代理在调用某个工具之前,把调用意图作为输入交给策略,由策略返回允许或拒绝。这样做的价值在于:代理的行为边界与基础设施的合规边界由同一套规则描述,不会出现两套标准互相矛盾的情况。授权输入应当包含工具名、目标资源标识与调用参数,策略据此判断是否落在允许范围内。
一个完整示例
把上面的步骤串成一条最小可跑通的链路:
- 在空目录里创建
main.tf,内容为第一步给出的最小模块,包含那个对世界开放 22 端口的安全组。 - 执行
terraform init,然后terraform plan -out=tfplan.binary,再terraform show -json tfplan.binary > plan.json。此时磁盘上有了真实的计划数据。 - 创建
policy/network_test.rego,粘贴第二步的测试内容。 - 创建
policy/network.rego,实现data.terraform.network.deny。 - 运行
opa test policy/ -v,反复修改策略直到全部通过。 - 运行
opa eval --data policy/ --input plan.json 'data.terraform.network.deny',确认真实计划上报出了预期数量的违规。 - 把
opa eval包进带退出码契约的脚本,在 CI 中作为 apply 前的门禁。 - 把同一条规则翻译成 CEL,配置到集群准入策略中。
这条链路的每一环都可以独立验证:测试验证策略逻辑,真实计划验证策略覆盖度,退出码验证 CI 集成,准入策略验证集群内的一致性。
注意事项
- 误放行不会被自动发现。策略放过了本应拦下的资源时,没有任何信号会提示你。只有主动构造测试、查看覆盖率,才能把这类问题找出来。
- 读不懂的输入必须上报,不能默认放过。端口字段为 null、为字符串、或结构不符合预期时,正确的行为是报出违规,而不是跳过。
- 端口比较必须用范围相交。只比较
from_port是否等于某个值,会漏掉0到65535这种写法。 - 全协议规则会打开所有端口。协议为
-1时,端口字段写什么都不影响实际效果。 - IPv6 是独立的路径。
::/0与0.0.0.0/0需要分别判断。 - 资源类型不止一种。安全组内联的 ingress 块、独立的
aws_vpc_security_group_ingress_rule、已弃用的aws_security_group_rule都可能承载同一条规则。 - OPA 0.x 与 1.x 语法不同。0.x 需要
import rego.v1,建议直接升级到 1.x。 - 计划 JSON 的结构自 provider 5 起稳定,但具体版本行为仍以各项目官网当前信息为准。
- 「无法判定」不等于「通过」。策略出错或输入无法解析时,门禁必须失败。
- 计划阶段与准入阶段的逻辑要保持一致。两处判定不一致会削弱对整套检查的信任。
这套检查能覆盖的是已经写成规则的那部分约束。规则没写到的、规则写错方向的、以及规则本身无法表达的判断,仍然需要人来负责。把规则写成可执行程序的价值不在于替代判断,而在于让判断在每一次变更时都被一致地执行一遍。