Enforce policies with OPA Gatekeeper and test with Conftest

Introduction

When a teammate ships a Kubernetes manifest pulling an image from docker.io/random-user/llm-server:latest with no resource limits and privileged: true, the API server admits it without complaint — schema validation passed, organizational standards did not. By the end of this lesson you'll be able to author an OPA ConstraintTemplate that Gatekeeper enforces at the admission webhook, point Conftest at the same Rego source so CI fails before merge, and roll the policy out via dryrun so existing workloads don't break the moment you ship.

Key Terminology

• OPA Gatekeeper — a Kubernetes admission controller that runs Rego policies on every create/update request; rejected resources never reach etcd, so the gate is enforced even when CI is bypassed.
• Conftest — a CLI that runs the same Rego policies against YAML files on disk during CI; catches violations in the pull request instead of at kubectl apply.
• Rego — the declarative policy language shared by both tools; a violation rule that produces a message is what blocks the resource.
• ConstraintTemplate — a Kubernetes CRD that registers reusable Rego logic with Gatekeeper; a Constraint instance parameterizes and activates it for specific resource kinds.
• Enforcement action — Gatekeeper's deny (block) vs dryrun (audit-only) mode, the rollout knob you turn after watching the audit log on a new policy.

Concepts

Two enforcement points, one policy language

Policy-as-code closes the gap between "valid YAML" and "manifest that meets our standards" at two checkpoints. Conftest runs in CI against rendered manifests — fast, fails the PR, never touches a cluster. Gatekeeper runs as an admission webhook — slower, blocks kubectl apply even when CI is bypassed. The same Rego source powers both, so a single policy authored once enforces twice (see Code Walkthrough).

ConstraintTemplate and Constraint separation

Gatekeeper splits policy into two CRDs: the ConstraintTemplate owns the Rego logic and declares an OpenAPI schema for parameters; the Constraint is an instance that fills those parameters in (allowed registries, required labels, resource ceilings). One Rego file then powers many environment-specific rules — staging accepts gcr.io/dev/*, production only gcr.io/prod/* — without forking the policy.

Rolling out with dryrun before deny

Switching a fresh constraint straight to enforcementAction: deny in a busy cluster can block legitimate workloads whose owners haven't seen the rule yet. The dryrun action evaluates every request and emits audit events without blocking — the safe rollout is dryrun first, watch the audit log for false positives, exempt namespaces that need it, then flip to deny.

Code Walkthrough

The snippet below ties the three Concepts together: it emits a ConstraintTemplate carrying Rego logic, a parameterized Constraint instance that starts in dryrun, and a Conftest runner that points at the same Rego file so CI and the admission webhook enforce identical rules.

Code snippetpython
1import json
2import subprocess
3from pathlib import Path
4from pydantic import BaseModel, Field
5import yaml
6
7ALLOWED_REGISTRY_REGO = """
8package k8sallowedregistries
9
10violation[{"msg": msg}] {
11    container := input.review.object.spec.containers[_]
12    allowed := input.parameters.allowedRegistries
13    not registry_allowed(container.image, allowed)
14    msg := sprintf(
15        "Image '%v' is not from an allowed registry",
16        [container.image],
17    )
18}
19
20registry_allowed(image, allowed) {
21    startswith(image, allowed[_])
22}
23"""
24
25class ConstraintTemplate(BaseModel):
26    name: str
27    rego_source: str
28
29def build_template(tpl: ConstraintTemplate) -> str:
30    doc = {
31        "apiVersion": "templates.gatekeeper.sh/v1beta1",
32        "kind": "ConstraintTemplate",
33        "metadata": {"name": tpl.name},
34        "spec": {
35            "crd": {
36                "spec": {
37                    "names": {"kind": "K8sAllowedRegistries"},
38                    "validation": {
39                        "openAPIV3Schema": {
40                            "properties": {
41                                "allowedRegistries": {
42                                    "type": "array",
43                                    "items": {"type": "string"},
44                                }
45                            }
46                        }
47                    },
48                }
49            },
50            "targets": [
51                {
52                    "target": "admission.k8s.gatekeeper.sh",
53                    "rego": tpl.rego_source,
54                }
55            ],
56        },
57    }
58    return yaml.dump(doc, default_flow_style=False)
59
60def build_constraint(allowed: list[str]) -> str:
61    doc = {
62        "apiVersion": "constraints.gatekeeper.sh/v1beta1",
63        "kind": "K8sAllowedRegistries",
64        "metadata": {"name": "require-approved-registry"},
65        "spec": {
66            "enforcementAction": "dryrun",
67            "match": {
68                "kinds": [{"apiGroups": [""], "kinds": ["Pod"]}],
69                "excludedNamespaces": ["kube-system", "gatekeeper-system"],
70            },
71            "parameters": {"allowedRegistries": allowed},
72        },
73    }
74    return yaml.dump(doc, default_flow_style=False)
75
76class ConftestResult(BaseModel):
77    filename: str
78    failures: list[str] = Field(default_factory=list)
79
80def run_conftest(manifest_dir: str, policy_dir: str) -> list[ConftestResult]:
81    results: list[ConftestResult] = []
82    for manifest in Path(manifest_dir).glob("*.yaml"):
83        proc = subprocess.run(
84            [
85                "conftest", "test", str(manifest),
86                "--policy", policy_dir,
87                "--namespace", "k8sallowedregistries",
88                "--output", "json",
89            ],
90            capture_output=True, text=True,
91        )
92        entries = json.loads(proc.stdout or "[]")
93        msgs = [
94            f["msg"]
95            for entry in entries
96            for f in entry.get("failures", [])
97        ]
98        results.append(ConftestResult(filename=manifest.name, failures=msgs))
99    return results
100
101if __name__ == "__main__":
102    tpl = ConstraintTemplate(
103        name="k8sallowedregistries",
104        rego_source=ALLOWED_REGISTRY_REGO,
105    )
106    Path("policies").mkdir(exist_ok=True)
107    Path("policies/registry.rego").write_text(ALLOWED_REGISTRY_REGO)
108    Path("template.yaml").write_text(build_template(tpl))
109    Path("constraint.yaml").write_text(
110        build_constraint([
111            "gcr.io/my-project/",
112            "us-docker.pkg.dev/my-project/",
113        ])
114    )
115    print(run_conftest("manifests/", "policies/"))

The same ALLOWED_REGISTRY_REGO string is embedded in the ConstraintTemplate YAML that Gatekeeper consumes AND written to policies/registry.rego for Conftest — single source of truth, two enforcement points. The Constraint ships with enforcementAction: dryrun and an excludedNamespaces list so the first deploy audits without blocking.

You'll know it works when kubectl apply -f bad-pod.yaml (pulling docker.io/library/nginx) is rejected by Gatekeeper with the Image '...' is not from an allowed registry message after you flip the constraint to deny, AND conftest test manifests/ --policy policies/ exits non-zero on the same manifest in CI before the apply ever runs.

Do's and Don'ts

Do's

1. ✓Do roll new constraints out in dryrun first — audit the violation log for a week before flipping to deny; you'll find policy bugs and exempt workloads you forgot existed.
2. ✓Do keep one Rego file as the source of truth — embed it in the ConstraintTemplate AND point Conftest at it; drift between CI and admission is how a manifest passes CI then fails at deploy.
3. ✓Do write a violation message that names the resource and the rule — Gatekeeper surfaces it verbatim to kubectl apply, so a clear message saves the next engineer ten minutes of guessing.

Don'ts

1. ✗Don't rely on Conftest alone — anyone bypassing CI (emergency hotfix, manual kubectl apply, controller-created pods) skips the gate; Gatekeeper is the backstop.
2. ✗Don't author one ConstraintTemplate per environment — parameterize via the Constraint CRD instead, or you'll triple-maintain the same Rego across staging, prod, and sandbox.
3. ✗Don't enable deny on a new policy in production without excludedNamespaces — at minimum exempt kube-system and your monitoring stack, otherwise a single bad rule can wedge cluster recovery.