Policy as Code (OPA/Conftest)

Declarative policy enforcement using Open Policy Agent (OPA) and Conftest for infrastructure, Kubernetes, and CI/CD validation.

Overview

Open Policy Agent (OPA) is a general-purpose policy engine that uses Rego, a declarative query language, to define policies as code. Conftest is a utility built on OPA specifically designed for testing configuration files. Together, they enable policy enforcement across Kubernetes admission control, Terraform plans, Docker images, CI/CD pipelines, and more.

Key Use Cases:

Kubernetes admission control (OPA Gatekeeper)
Terraform plan validation
Docker image security scanning
Configuration file validation in CI/CD
Cloud resource compliance checking
API authorisation and access control

flowchart TB
    subgraph Input
        A[Kubernetes YAML]
        B[Terraform Plan]
        C[Docker Config]
        D[JSON/YAML Config]
    end

    subgraph Policy Engine
        E[OPA/Conftest]
        F[Rego Policies]
        G[Policy Bundle]
    end

    subgraph Output
        H{Policy Check}
        I[✓ Allow]
        J[✗ Deny with Violation]
    end

    A --> E
    B --> E
    C --> E
    D --> E
    F --> E
    G --> E
    E --> H
    H -->|Pass| I
    H -->|Fail| J

Rego Language Basics

Core Concepts

Rego is a declarative language based on Datalog. Policies are expressed as rules that evaluate to true or false.

Key Principles:

Rules define conditions that must be met
Variables are immutable
No loops or recursion (use comprehensions, walk(), and iteration)
Data is queried, not mutated
Everything is either true, false, or undefined

Basic Syntax

# Package declaration (required)
package mypackage

# Import statements
import data.kubernetes

# OPA 1.0+ defaults to Rego v1: `if` and `contains` are keywords now,
# so the `import future.keywords.*` lines are no longer needed (they
# still parse as a no-op on older policies). On OPA < 1.0, either add
# `import rego.v1` or those `future.keywords` imports to use this syntax.

# Simple rule (boolean)
allow if {
    input.user == "admin"
}

# Rule with value assignment
default allow := false

allow := true if {
    input.user == "admin"
}

# Partial set rule with iteration (note `contains` + `if`)
deny contains msg if {
    some container in input.spec.containers
    not container.securityContext.runAsNonRoot
    msg := sprintf("Container %s must run as non-root", [container.name])
}

# Comprehensions
container_names := [name | some container in input.spec.containers; name := container.name]

# Helper functions
is_production if {
    input.metadata.labels.environment == "production"
}

# Sets
approved_images := {
    "nginx:1.21",
    "redis:6.2",
    "postgres:14"
}

Data Types

# Strings
message := "Policy violation"

# Numbers
max_replicas := 10

# Booleans
is_valid := true

# Arrays (ordered)
allowed_ports := [80, 443, 8080]

# Sets (unordered, unique)
blocked_registries := {"docker.io", "gcr.io"}

# Objects (key-value)
limits := {
    "cpu": "500m",
    "memory": "512Mi"
}

# Null
default value := null

Control Flow

# Multiple conditions (AND)
allow if {
    input.user == "admin"
    input.action == "write"
}

# Alternative rules (OR)
allow if input.user == "admin"
allow if input.user == "superuser"

# Negation
deny contains msg if {
    not input.securityContext.readOnlyRootFilesystem
    msg := "Root filesystem must be read-only"
}

# Conditionals with else
access_level := "full" if {
    input.user == "admin"
} else := "limited" if {
    input.user == "developer"
} else := "none"

Writing Policies

Deny Pattern (Most Common)

Used to reject non-compliant configurations. Returns violation messages.

package kubernetes.admission

# Deny containers without resource limits
deny contains msg if {
    input.request.kind.kind == "Pod"
    some container in input.request.object.spec.containers
    not container.resources.limits.cpu
    msg := sprintf("Container '%s' must specify CPU limits", [container.name])
}

deny contains msg if {
    input.request.kind.kind == "Pod"
    some container in input.request.object.spec.containers
    not container.resources.limits.memory
    msg := sprintf("Container '%s' must specify memory limits", [container.name])
}

# Deny privileged containers
deny contains msg if {
    input.request.kind.kind == "Pod"
    some container in input.request.object.spec.containers
    container.securityContext.privileged == true
    msg := sprintf("Privileged container '%s' is not allowed", [container.name])
}

# Deny unapproved registries
deny contains msg if {
    input.request.kind.kind == "Pod"
    some container in input.request.object.spec.containers
    image := container.image
    not startswith(image, "myregistry.io/")
    not startswith(image, "docker.io/library/")
    msg := sprintf("Container '%s' uses unapproved registry: %s", [container.name, image])
}

Allow Pattern

Used for positive authorisation (e.g., RBAC). Default is deny unless explicitly allowed.

package authz

import future.keywords.if

# Default deny
default allow := false

# Allow admins everything
allow if {
    input.user.role == "admin"
}

# Allow developers to read
allow if {
    input.user.role == "developer"
    input.action in ["read", "list"]
}

# Allow specific users for specific resources
allow if {
    input.user.name == "ci-bot"
    input.resource.type == "deployment"
    input.action == "update"
    input.resource.namespace == "staging"
}

Warn Pattern

Non-blocking violations that generate warnings.

package kubernetes.admission

warn contains msg if {
    input.request.kind.kind == "Pod"
    some container in input.request.object.spec.containers
    not container.livenessProbe
    msg := sprintf("Container '%s' should define a liveness probe", [container.name])
}

warn contains msg if {
    input.request.kind.kind == "Deployment"
    replicas := input.request.object.spec.replicas
    replicas < 2
    msg := "Deployment should have at least 2 replicas for high availability"
}

Kubernetes Policy Examples

Security Context Policies

package kubernetes.admission

import future.keywords.contains
import future.keywords.if
import future.keywords.in

# Deny pods not running as non-root
deny contains msg if {
    input.request.kind.kind == "Pod"
    some container in input.request.object.spec.containers
    not container.securityContext.runAsNonRoot
    msg := sprintf("Container '%s' must set runAsNonRoot: true", [container.name])
}

# Deny containers with privilege escalation
deny contains msg if {
    input.request.kind.kind == "Pod"
    some container in input.request.object.spec.containers
    container.securityContext.allowPrivilegeEscalation == true
    msg := sprintf("Container '%s' must set allowPrivilegeEscalation: false", [container.name])
}

# Deny containers without read-only root filesystem
deny contains msg if {
    input.request.kind.kind == "Pod"
    some container in input.request.object.spec.containers
    not container.securityContext.readOnlyRootFilesystem
    msg := sprintf("Container '%s' must set readOnlyRootFilesystem: true", [container.name])
}

# Deny dangerous capabilities
dangerous_capabilities := {"SYS_ADMIN", "NET_ADMIN", "SYS_MODULE", "SYS_RAWIO"}

deny contains msg if {
    input.request.kind.kind == "Pod"
    some container in input.request.object.spec.containers
    some capability in container.securityContext.capabilities.add
    capability in dangerous_capabilities
    msg := sprintf("Container '%s' cannot add capability: %s", [container.name, capability])
}

# Require dropping ALL capabilities
deny contains msg if {
    input.request.kind.kind == "Pod"
    some container in input.request.object.spec.containers
    not "ALL" in container.securityContext.capabilities.drop
    msg := sprintf("Container '%s' must drop ALL capabilities", [container.name])
}

Resource Limit Policies

package kubernetes.admission

import future.keywords.if

# Both CPU and memory limits required
# Note: `not cpu` AND `not memory` only fires when *both* are absent. To
# require each independently, split into two rules (as the Deny Pattern
# section does) so a container missing just one limit is still flagged.
deny contains msg if {
    input.request.kind.kind in ["Pod", "Deployment", "StatefulSet", "DaemonSet"]
    containers := object.get(input.request.object.spec, "containers", [])
    some container in containers
    not container.resources.limits.cpu
    not container.resources.limits.memory
    msg := sprintf("Container '%s' must specify both CPU and memory limits", [container.name])
}

# Maximum resource limits
max_cpu := "2000m"
max_memory := "4Gi"

deny contains msg if {
    input.request.kind.kind == "Pod"
    some container in input.request.object.spec.containers
    cpu_limit := container.resources.limits.cpu
    cpu_value := parse_cpu(cpu_limit)
    cpu_value > parse_cpu(max_cpu)
    msg := sprintf("Container '%s' CPU limit %s exceeds maximum %s", [container.name, cpu_limit, max_cpu])
}

# Helper to parse CPU values
parse_cpu(cpu) := result if {
    endswith(cpu, "m")
    result := to_number(trim_suffix(cpu, "m"))
}

parse_cpu(cpu) := result if {
    not endswith(cpu, "m")
    result := to_number(cpu) * 1000
}

# Ensure requests are set and less than limits
deny contains msg if {
    input.request.kind.kind == "Pod"
    some container in input.request.object.spec.containers
    container.resources.limits.memory
    not container.resources.requests.memory
    msg := sprintf("Container '%s' must specify memory requests when limits are set", [container.name])
}

Namespace and Label Policies

package kubernetes.admission

import future.keywords.if
import future.keywords.in

# Required labels
required_labels := {"app", "environment", "owner"}

deny contains msg if {
    input.request.kind.kind in ["Deployment", "StatefulSet", "DaemonSet"]
    labels := object.get(input.request.object.metadata, "labels", {})
    some required_label in required_labels
    not labels[required_label]
    msg := sprintf("Missing required label: %s", [required_label])
}

# Validate environment label values
valid_environments := {"dev", "staging", "production"}

deny contains msg if {
    input.request.kind.kind in ["Deployment", "StatefulSet", "DaemonSet"]
    environment := input.request.object.metadata.labels.environment
    not environment in valid_environments
    msg := sprintf("Invalid environment label '%s'. Must be one of: %v", [environment, valid_environments])
}

# Prevent deployment to default namespace
deny contains msg if {
    input.request.kind.kind in ["Pod", "Deployment", "StatefulSet", "DaemonSet", "Service"]
    input.request.namespace == "default"
    msg := "Resources cannot be created in the default namespace"
}

# Namespace naming convention
deny contains msg if {
    input.request.kind.kind == "Namespace"
    name := input.request.object.metadata.name
    not regex.match("^(dev|staging|prod)-[a-z0-9-]+$", name)
    msg := sprintf("Namespace '%s' does not follow naming convention: (dev|staging|prod)-<name>", [name])
}

Image Security Policies

package kubernetes.admission

import future.keywords.if
import future.keywords.in

# Approved registries
approved_registries := {
    "myregistry.io",
    "gcr.io/my-project",
    "docker.io/library"
}

deny contains msg if {
    input.request.kind.kind == "Pod"
    some container in input.request.object.spec.containers
    image := container.image
    not image_from_approved_registry(image)
    msg := sprintf("Container '%s' uses unapproved registry: %s", [container.name, image])
}

image_from_approved_registry(image) if {
    some registry in approved_registries
    startswith(image, registry)
}

# Deny latest tag
deny contains msg if {
    input.request.kind.kind == "Pod"
    some container in input.request.object.spec.containers
    image := container.image
    endswith(image, ":latest")
    msg := sprintf("Container '%s' cannot use ':latest' tag", [container.name])
}

deny contains msg if {
    input.request.kind.kind == "Pod"
    some container in input.request.object.spec.containers
    image := container.image
    not contains(image, ":")
    msg := sprintf("Container '%s' must specify image tag", [container.name])
}

# Require image pull policy
deny contains msg if {
    input.request.kind.kind == "Pod"
    some container in input.request.object.spec.containers
    not container.imagePullPolicy
    msg := sprintf("Container '%s' must specify imagePullPolicy", [container.name])
}

deny contains msg if {
    input.request.kind.kind == "Pod"
    some container in input.request.object.spec.containers
    container.imagePullPolicy != "Always"
    contains(container.image, ":latest")
    msg := sprintf("Container '%s' using ':latest' must use imagePullPolicy: Always", [container.name])
}

Terraform Policy Examples

AWS Resource Policies

package terraform.aws

import future.keywords.contains
import future.keywords.if
import future.keywords.in

# Deny public S3 buckets
deny contains msg if {
    resource := input.resource_changes[_]
    resource.type == "aws_s3_bucket"
    resource.change.after.acl == "public-read"
    msg := sprintf("S3 bucket '%s' cannot have public-read ACL", [resource.name])
}

deny contains msg if {
    resource := input.resource_changes[_]
    resource.type == "aws_s3_bucket_public_access_block"
    resource.change.after.block_public_acls == false
    msg := sprintf("S3 bucket '%s' must block public ACLs", [resource.name])
}

# Require encryption at rest
deny contains msg if {
    resource := input.resource_changes[_]
    resource.type == "aws_s3_bucket"
    not resource.change.after.server_side_encryption_configuration
    msg := sprintf("S3 bucket '%s' must enable server-side encryption", [resource.name])
}

deny contains msg if {
    resource := input.resource_changes[_]
    resource.type == "aws_ebs_volume"
    resource.change.after.encrypted != true
    msg := sprintf("EBS volume '%s' must be encrypted", [resource.name])
}

# Require versioning for production S3 buckets
deny contains msg if {
    resource := input.resource_changes[_]
    resource.type == "aws_s3_bucket"
    tags := object.get(resource.change.after, "tags", {})
    tags.environment == "production"
    not resource.change.after.versioning[0].enabled
    msg := sprintf("Production S3 bucket '%s' must enable versioning", [resource.name])
}

# Deny public EC2 instances
deny contains msg if {
    resource := input.resource_changes[_]
    resource.type == "aws_instance"
    resource.change.after.associate_public_ip_address == true
    tags := object.get(resource.change.after, "tags", {})
    tags.environment == "production"
    msg := sprintf("Production EC2 instance '%s' cannot have public IP", [resource.name])
}

# Require VPC flow logs
deny contains msg if {
    resource := input.resource_changes[_]
    resource.type == "aws_vpc"
    vpc_id := resource.change.after.id
    not has_flow_log(vpc_id)
    msg := sprintf("VPC '%s' must have flow logs enabled", [resource.name])
}

has_flow_log(vpc_id) if {
    some resource in input.resource_changes
    resource.type == "aws_flow_log"
    resource.change.after.vpc_id == vpc_id
}

# Enforce instance type restrictions
allowed_instance_types := {"t3.micro", "t3.small", "t3.medium"}

deny contains msg if {
    resource := input.resource_changes[_]
    resource.type == "aws_instance"
    instance_type := resource.change.after.instance_type
    not instance_type in allowed_instance_types
    msg := sprintf("Instance '%s' type '%s' not in allowed types: %v", [resource.name, instance_type, allowed_instance_types])
}

GCP Resource Policies

package terraform.gcp

import future.keywords.if
import future.keywords.in

# Require encryption for GCS buckets
deny contains msg if {
    resource := input.resource_changes[_]
    resource.type == "google_storage_bucket"
    not resource.change.after.encryption
    msg := sprintf("GCS bucket '%s' must enable encryption", [resource.name])
}

# Deny public GCS buckets
deny contains msg if {
    resource := input.resource_changes[_]
    resource.type == "google_storage_bucket_iam_member"
    member := resource.change.after.member
    member in ["allUsers", "allAuthenticatedUsers"]
    msg := sprintf("GCS bucket cannot grant access to %s", [member])
}

# Require private GKE clusters
deny contains msg if {
    resource := input.resource_changes[_]
    resource.type == "google_container_cluster"
    not resource.change.after.private_cluster_config
    msg := sprintf("GKE cluster '%s' must be private", [resource.name])
}

deny contains msg if {
    resource := input.resource_changes[_]
    resource.type == "google_container_cluster"
    resource.change.after.private_cluster_config[0].enable_private_nodes != true
    msg := sprintf("GKE cluster '%s' must enable private nodes", [resource.name])
}

# Require network policies for GKE
deny contains msg if {
    resource := input.resource_changes[_]
    resource.type == "google_container_cluster"
    not resource.change.after.network_policy[0].enabled
    msg := sprintf("GKE cluster '%s' must enable network policies", [resource.name])
}

Cost Control Policies

package terraform.cost

import future.keywords.if

# Maximum monthly cost estimate
max_monthly_cost := 10000

deny contains msg if {
    total_cost := sum([cost |
        some resource in input.resource_changes
        resource.change.actions[_] in ["create", "update"]
        cost := object.get(resource, "cost_estimate", 0)
    ])
    total_cost > max_monthly_cost
    msg := sprintf("Total estimated monthly cost £%v exceeds budget £%v", [total_cost, max_monthly_cost])
}

# Warn on expensive resources
warn contains msg if {
    resource := input.resource_changes[_]
    cost := object.get(resource, "cost_estimate", 0)
    cost > 1000
    msg := sprintf("Resource '%s' has high estimated monthly cost: £%v", [resource.name, cost])
}

Testing Policies

Unit Testing with OPA Test Framework

package kubernetes.admission_test

import data.kubernetes.admission
import future.keywords.if

# Test: Deny pod without resource limits
test_deny_pod_without_cpu_limit if {
    pod := {
        "request": {
            "kind": {"kind": "Pod"},
            "object": {
                "metadata": {"name": "test-pod"},
                "spec": {
                    "containers": [{
                        "name": "nginx",
                        "image": "nginx:1.21",
                        "resources": {
                            "limits": {"memory": "512Mi"}
                        }
                    }]
                }
            }
        }
    }

    violations := admission.deny with input as pod
    count(violations) > 0
    some violation in violations
    contains(violation, "CPU limits")
}

# Test: Allow pod with proper resource limits
test_allow_pod_with_resource_limits if {
    pod := {
        "request": {
            "kind": {"kind": "Pod"},
            "object": {
                "metadata": {"name": "test-pod"},
                "spec": {
                    "containers": [{
                        "name": "nginx",
                        "image": "nginx:1.21",
                        "resources": {
                            "limits": {
                                "cpu": "500m",
                                "memory": "512Mi"
                            },
                            "requests": {
                                "cpu": "250m",
                                "memory": "256Mi"
                            }
                        }
                    }]
                }
            }
        }
    }

    violations := admission.deny with input as pod
    count(violations) == 0
}

# Test: Deny privileged container
test_deny_privileged_container if {
    pod := {
        "request": {
            "kind": {"kind": "Pod"},
            "object": {
                "spec": {
                    "containers": [{
                        "name": "nginx",
                        "securityContext": {
                            "privileged": true
                        }
                    }]
                }
            }
        }
    }

    violations := admission.deny with input as pod
    count(violations) > 0
}

# Test with mock data
test_allow_admin_user if {
    result := admission.allow with input as {"user": {"role": "admin"}}
    result == true
}

test_deny_regular_user if {
    result := admission.allow with input as {"user": {"role": "user"}}
    result == false
}

Running Tests

# Run all tests in a directory
opa test policies/

# Run tests with verbose output
opa test -v policies/

# Run tests with coverage
opa test --coverage policies/

# Run specific test file
opa test policies/kubernetes_test.rego

# Run tests matching pattern
opa test --run test_deny policies/

# Generate coverage report
opa test --coverage --format=json policies/ > coverage.json

Test File Organisation

policies/
├── kubernetes/
│   ├── admission.rego          # Policy rules
│   ├── admission_test.rego     # Unit tests
│   └── helpers.rego            # Shared functions
├── terraform/
│   ├── aws.rego
│   ├── aws_test.rego
│   ├── gcp.rego
│   └── gcp_test.rego
└── conftest.toml               # Configuration

Conftest for Configuration Testing

Basic Usage

# Test Kubernetes manifests
conftest test deployment.yaml

# Test multiple files
conftest test k8s/*.yaml

# Test with specific policy directory
conftest test -p policies/ deployment.yaml

# Test Terraform plan
terraform plan -out=plan.tfplan
terraform show -json plan.tfplan > plan.json
conftest test plan.json

# Test with all namespaces
conftest test --all-namespaces deployment.yaml

# Test and fail on warnings
conftest test --fail-on-warn deployment.yaml

# Test Docker files
conftest test Dockerfile

# Output formats
conftest test --output json deployment.yaml
conftest test --output table deployment.yaml
conftest test --output junit deployment.yaml

Conftest Configuration

# conftest.toml
policy = "policies"
namespace = "main"

# Fail on warnings
fail_on_warn = false

# Output format
output = "stdout"

# Trace output
trace = false

# Combine configs
combine = false

# Parser for specific file types
[[input]]
type = "yaml"

[[input]]
type = "json"

[[input]]
type = "dockerfile"

Writing Conftest Policies

# policies/deployment.rego
package main

import future.keywords.contains
import future.keywords.if
import future.keywords.in

# Deny rule for Conftest
deny contains msg if {
    input.kind == "Deployment"
    not input.spec.template.spec.securityContext.runAsNonRoot
    msg := "Deployment must run as non-root user"
}

deny contains msg if {
    input.kind == "Deployment"
    some container in input.spec.template.spec.containers
    not container.resources.limits
    msg := sprintf("Container '%s' must define resource limits", [container.name])
}

# Warning rule
warn contains msg if {
    input.kind == "Deployment"
    replicas := input.spec.replicas
    replicas < 2
    msg := "Deployment should have at least 2 replicas for HA"
}

# Violation rule (alternative to deny)
violation contains {"msg": msg, "details": details} if {
    input.kind == "Service"
    input.spec.type == "LoadBalancer"
    msg := "LoadBalancer services are not recommended"
    details := {"service": input.metadata.name}
}

Testing Docker Images

# policies/docker.rego
package main

import future.keywords.if

deny contains msg if {
    input[i].Cmd == "from"
    val := split(input[i].Value[0], ":")
    count(val) == 1
    msg := "Base image must specify a tag"
}

deny contains msg if {
    input[i].Cmd == "from"
    contains(input[i].Value[0], ":latest")
    msg := "Base image should not use ':latest' tag"
}

deny contains msg if {
    input[i].Cmd == "run"
    contains(input[i].Value[0], "sudo")
    msg := "Avoid using 'sudo' in Dockerfile"
}

warn contains msg if {
    not has_user_directive
    msg := "Dockerfile should specify USER directive"
}

has_user_directive if {
    some i
    input[i].Cmd == "user"
}

warn contains msg if {
    not has_healthcheck
    msg := "Dockerfile should include HEALTHCHECK"
}

has_healthcheck if {
    some i
    input[i].Cmd == "healthcheck"
}

OPA Gatekeeper for Kubernetes

Architecture

flowchart TB
    A[kubectl apply] --> B[Kubernetes API Server]
    B --> C{Admission Controller}
    C --> D[OPA Gatekeeper]
    D --> E[Constraint Templates]
    D --> F[Constraints]
    E --> G{Evaluate Policy}
    F --> G
    G -->|Allowed| H[Create Resource]
    G -->|Denied| I[Reject with Violation]

    J[Audit Controller] --> K[Scan Existing Resources]
    K --> G
    G --> L[Violation Reports]

Installation

# Install Gatekeeper (pin to a current release branch; 3.22 is latest as of 2026)
kubectl apply -f https://raw.githubusercontent.com/open-policy-agent/gatekeeper/release-3.22/deploy/gatekeeper.yaml

# Verify installation
kubectl get pods -n gatekeeper-system

# Check CRDs
kubectl get crd | grep gatekeeper

Constraint Templates

Constraint Templates define reusable policy schemas.

# constraint-template-require-labels.yaml
apiVersion: templates.gatekeeper.sh/v1
kind: ConstraintTemplate
metadata:
  name: k8srequiredlabels
spec:
  crd:
    spec:
      names:
        kind: K8sRequiredLabels
      validation:
        # Schema for constraint parameters
        openAPIV3Schema:
          type: object
          properties:
            labels:
              type: array
              items:
                type: string
  targets:
    # NOTE: the legacy `rego:` field is Rego v0 — partial rules use the
    # `violation[{...}] { }` form (no `if`/`contains`). For Rego v1 syntax
    # in Gatekeeper (3.19+), use the newer structure instead:
    #   code:
    #     - engine: Rego
    #       source:
    #         version: "v1"
    #         rego: |
    #           violation contains {"msg": msg} if { ... }
    - target: admission.k8s.gatekeeper.sh
      rego: |
        package k8srequiredlabels

        violation[{"msg": msg, "details": {"missing_labels": missing}}] {
          provided := {label | input.review.object.metadata.labels[label]}
          required := {label | label := input.parameters.labels[_]}
          missing := required - provided
          count(missing) > 0
          msg := sprintf("Missing required labels: %v", [missing])
        }

# constraint-template-resource-limits.yaml
apiVersion: templates.gatekeeper.sh/v1
kind: ConstraintTemplate
metadata:
  name: k8scontainerresources
spec:
  crd:
    spec:
      names:
        kind: K8sContainerResources
      validation:
        openAPIV3Schema:
          type: object
          properties:
            limits:
              type: object
              properties:
                cpu:
                  type: string
                memory:
                  type: string
  targets:
    - target: admission.k8s.gatekeeper.sh
      rego: |
        package k8scontainerresources

        violation[{"msg": msg}] {
          container := input.review.object.spec.containers[_]
          not container.resources.limits.cpu
          msg := sprintf("Container '%s' must specify CPU limits", [container.name])
        }

        violation[{"msg": msg}] {
          container := input.review.object.spec.containers[_]
          not container.resources.limits.memory
          msg := sprintf("Container '%s' must specify memory limits", [container.name])
        }

Constraints

Constraints are instances of Constraint Templates with specific parameters.

# constraint-production-labels.yaml
apiVersion: constraints.gatekeeper.sh/v1beta1
kind: K8sRequiredLabels
metadata:
  name: require-production-labels
spec:
  match:
    kinds:
      - apiGroups: ["apps"]
        kinds: ["Deployment", "StatefulSet"]
    namespaceSelector:
      matchLabels:
        environment: production
  parameters:
    labels:
      - app
      - environment
      - owner
      - version

# constraint-resource-limits.yaml
apiVersion: constraints.gatekeeper.sh/v1beta1
kind: K8sContainerResources
metadata:
  name: require-container-resources
spec:
  match:
    kinds:
      - apiGroups: [""]
        kinds: ["Pod"]
      - apiGroups: ["apps"]
        kinds: ["Deployment", "StatefulSet", "DaemonSet"]
  parameters:
    limits:
      cpu: "2000m"
      memory: "4Gi"

Security Constraint Templates

apiVersion: templates.gatekeeper.sh/v1
kind: ConstraintTemplate
metadata:
  name: k8spspsecuritycontext
spec:
  crd:
    spec:
      names:
        kind: K8sPSPSecurityContext
  targets:
    - target: admission.k8s.gatekeeper.sh
      rego: |
        package k8spspsecuritycontext

        violation[{"msg": msg}] {
          container := input.review.object.spec.containers[_]
          not container.securityContext.runAsNonRoot
          msg := sprintf("Container '%s' must set runAsNonRoot: true", [container.name])
        }

        violation[{"msg": msg}] {
          container := input.review.object.spec.containers[_]
          container.securityContext.privileged
          msg := sprintf("Container '%s' cannot be privileged", [container.name])
        }

        violation[{"msg": msg}] {
          container := input.review.object.spec.containers[_]
          container.securityContext.allowPrivilegeEscalation
          msg := sprintf("Container '%s' must set allowPrivilegeEscalation: false", [container.name])
        }

        violation[{"msg": msg}] {
          container := input.review.object.spec.containers[_]
          not container.securityContext.readOnlyRootFilesystem
          msg := sprintf("Container '%s' must set readOnlyRootFilesystem: true", [container.name])
        }

Audit and Compliance

# List all constraint templates
kubectl get constrainttemplates

# List all constraints
kubectl get constraints

# View violations for a constraint
kubectl get k8srequiredlabels require-production-labels -o yaml

# Check violation status
kubectl get constraint require-production-labels -o json | jq '.status.violations'

# Audit existing resources
kubectl get constraint -A -o json | jq '.items[] | select(.status.totalViolations > 0)'

# Get audit results
kubectl get constraint -o jsonpath='{range .items[*]}{.metadata.name}{"\t"}{.status.totalViolations}{"\n"}{end}'

Mutations (Alpha Feature)

apiVersion: mutations.gatekeeper.sh/v1alpha1
kind: Assign
metadata:
  name: add-security-context
spec:
  applyTo:
    - groups: [""]
      kinds: ["Pod"]
      versions: ["v1"]
  match:
    scope: Namespaced
  location: "spec.securityContext.runAsNonRoot"
  parameters:
    assign:
      value: true

CI/CD Integration

Policy Evaluation Workflow

flowchart TB
    A[Code Push] --> B[CI Pipeline Triggered]
    B --> C[Build Artifacts]
    C --> D{Artifact Type}

    D -->|Kubernetes| E[conftest test manifests/]
    D -->|Terraform| F[terraform plan + conftest]
    D -->|Docker| G[conftest test Dockerfile]

    E --> H{Policy Check}
    F --> H
    G --> H

    H -->|Pass| I[Continue Pipeline]
    H -->|Fail| J[Block Pipeline]

    I --> K[Deploy to Environment]
    J --> L[Notify Developer]
    L --> M[Fix Violations]
    M --> A

GitHub Actions Integration

# .github/workflows/policy-check.yml
name: Policy Validation

on:
  pull_request:
    paths:
      - 'k8s/**'
      - 'terraform/**'
      - 'policies/**'

jobs:
  conftest-kubernetes:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Install Conftest
        run: |
          wget https://github.com/open-policy-agent/conftest/releases/download/v0.68.2/conftest_0.68.2_Linux_x86_64.tar.gz
          tar xzf conftest_0.68.2_Linux_x86_64.tar.gz
          sudo mv conftest /usr/local/bin/

      - name: Test Kubernetes Manifests
        run: |
          conftest test k8s/*.yaml \
            -p policies/kubernetes/ \
            --fail-on-warn \
            --output github

      - name: Upload Results
        if: always()
        uses: actions/upload-artifact@v3
        with:
          name: policy-violations
          path: violations.json

  conftest-terraform:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Setup Terraform
        uses: hashicorp/setup-terraform@v2

      - name: Terraform Plan
        run: |
          terraform init
          terraform plan -out=plan.tfplan
          terraform show -json plan.tfplan > plan.json

      - name: Install Conftest
        run: |
          wget https://github.com/open-policy-agent/conftest/releases/download/v0.68.2/conftest_0.68.2_Linux_x86_64.tar.gz
          tar xzf conftest_0.68.2_Linux_x86_64.tar.gz
          sudo mv conftest /usr/local/bin/

      - name: Test Terraform Plan
        run: |
          conftest test plan.json \
            -p policies/terraform/ \
            --fail-on-warn \
            --output json > terraform-violations.json

      - name: Comment PR
        uses: actions/github-script@v6
        if: github.event_name == 'pull_request'
        with:
          script: |
            const fs = require('fs');
            const violations = JSON.parse(fs.readFileSync('terraform-violations.json'));

            let comment = '## Terraform Policy Violations\n\n';
            if (violations.length === 0) {
              comment += '✅ No policy violations found!';
            } else {
              violations.forEach(v => {
                comment += `❌ **${v.filename}**\n`;
                v.failures.forEach(f => comment += `  - ${f.msg}\n`);
              });
            }

            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: comment
            });

  opa-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Install OPA
        run: |
          curl -L -o opa https://openpolicyagent.org/downloads/latest/opa_linux_amd64
          chmod +x opa
          sudo mv opa /usr/local/bin/

      - name: Run OPA Tests
        run: |
          opa test policies/ -v --coverage --format=json > coverage.json

      - name: Check Coverage
        run: |
          coverage=$(jq '.coverage' coverage.json)
          echo "Policy coverage: $coverage%"
          if (( $(echo "$coverage < 80" | bc -l) )); then
            echo "Coverage below 80%"
            exit 1
          fi

GitLab CI Integration

# .gitlab-ci.yml
stages:
  - validate
  - test
  - deploy

variables:
  CONFTEST_VERSION: "0.68.2"
  OPA_VERSION: "1.17.1"

.install_conftest: &install_conftest
  - wget -q https://github.com/open-policy-agent/conftest/releases/download/v${CONFTEST_VERSION}/conftest_${CONFTEST_VERSION}_Linux_x86_64.tar.gz
  - tar xzf conftest_${CONFTEST_VERSION}_Linux_x86_64.tar.gz
  - mv conftest /usr/local/bin/

.install_opa: &install_opa
  - curl -L -o /usr/local/bin/opa https://openpolicyagent.org/downloads/v${OPA_VERSION}/opa_linux_amd64
  - chmod +x /usr/local/bin/opa

policy_test:
  stage: validate
  image: alpine:latest
  before_script:
    - apk add --no-cache wget curl jq
    - *install_opa
  script:
    - opa test policies/ -v --coverage
    - opa check policies/
  artifacts:
    reports:
      junit: opa-test-results.xml

kubernetes_policy:
  stage: validate
  image: alpine:latest
  before_script:
    - apk add --no-cache wget
    - *install_conftest
  script:
    - conftest test k8s/*.yaml -p policies/kubernetes/ --fail-on-warn
  only:
    changes:
      - k8s/**
      - policies/kubernetes/**

terraform_policy:
  stage: validate
  image: hashicorp/terraform:latest
  before_script:
    - apk add --no-cache wget
    - *install_conftest
  script:
    - terraform init
    - terraform plan -out=plan.tfplan
    - terraform show -json plan.tfplan > plan.json
    - conftest test plan.json -p policies/terraform/ --output json | tee violations.json
  artifacts:
    paths:
      - violations.json
    reports:
      codequality: violations.json
  only:
    changes:
      - terraform/**
      - policies/terraform/**

docker_policy:
  stage: validate
  image: alpine:latest
  before_script:
    - apk add --no-cache wget
    - *install_conftest
  script:
    - conftest test Dockerfile -p policies/docker/
  only:
    changes:
      - Dockerfile
      - policies/docker/**

Jenkins Integration

// Jenkinsfile
pipeline {
    agent any

    environment {
        CONFTEST_VERSION = '0.68.2'
        OPA_VERSION = '1.17.1'
    }

    stages {
        stage('Setup Tools') {
            steps {
                sh '''
                    # Install Conftest
                    wget https://github.com/open-policy-agent/conftest/releases/download/v${CONFTEST_VERSION}/conftest_${CONFTEST_VERSION}_Linux_x86_64.tar.gz
                    tar xzf conftest_${CONFTEST_VERSION}_Linux_x86_64.tar.gz
                    chmod +x conftest

                    # Install OPA
                    curl -L -o opa https://openpolicyagent.org/downloads/v${OPA_VERSION}/opa_linux_amd64
                    chmod +x opa
                '''
            }
        }

        stage('Test Policies') {
            steps {
                sh '''
                    ./opa test policies/ -v --coverage --format=json > coverage.json
                    coverage=$(jq -r '.coverage' coverage.json)
                    echo "Policy Test Coverage: $coverage%"
                '''
            }
        }

        stage('Validate Kubernetes') {
            when {
                changeset "k8s/**"
            }
            steps {
                sh './conftest test k8s/*.yaml -p policies/kubernetes/ --fail-on-warn --output json > k8s-violations.json'
            }
            post {
                always {
                    archiveArtifacts artifacts: 'k8s-violations.json', allowEmptyArchive: true
                }
            }
        }

        stage('Validate Terraform') {
            when {
                changeset "terraform/**"
            }
            steps {
                sh '''
                    terraform init
                    terraform plan -out=plan.tfplan
                    terraform show -json plan.tfplan > plan.json
                    ./conftest test plan.json -p policies/terraform/ --output json > tf-violations.json
                '''
            }
            post {
                always {
                    archiveArtifacts artifacts: 'tf-violations.json', allowEmptyArchive: true
                }
            }
        }
    }

    post {
        failure {
            emailext (
                subject: "Policy Violations in ${env.JOB_NAME} - Build ${env.BUILD_NUMBER}",
                body: "Check console output at ${env.BUILD_URL}",
                to: "${env.CHANGE_AUTHOR_EMAIL}"
            )
        }
    }
}

Policy Bundles and Distribution

Bundle Structure

policy-bundle/
├── .manifest
├── kubernetes/
│   ├── admission.rego
│   ├── admission_test.rego
│   └── data.json
├── terraform/
│   ├── aws.rego
│   ├── gcp.rego
│   └── terraform_test.rego
└── data/
    ├── approved_images.json
    └── configurations.yaml

Creating Bundles

# Build a bundle
opa build -b policies/ -o bundle.tar.gz

# Build with optimisation (requires at least one --entrypoint/-e)
opa build -b -O 3 -e kubernetes/admission policies/ -o bundle.tar.gz

# Include data files
opa build -b policies/ data/ -o bundle.tar.gz

# Sign a bundle
opa build -b --signing-key private_key.pem policies/ -o bundle.tar.gz

# Verify bundle signature
opa build --verification-key public_key.pem --bundle bundle.tar.gz

Bundle Distribution

flowchart LR
    A[Policy Repository] --> B[Build Bundle]
    B --> C[Sign Bundle]
    C --> D{Distribution Method}

    D -->|HTTP| E[Web Server]
    D -->|S3| F[AWS S3]
    D -->|GCS| G[Google Cloud Storage]
    D -->|OCI| H[Container Registry]

    E --> I[OPA Instances]
    F --> I
    G --> I
    H --> I

    I --> J[Load & Activate]
    J --> K[Policy Enforcement]

OPA Configuration for Bundles

# opa-config.yaml
services:
  policy-service:
    url: https://policy-bundles.example.com
    credentials:
      bearer:
        token: ${BUNDLE_TOKEN}

bundles:
  main:
    service: policy-service
    resource: bundles/production/bundle.tar.gz
    polling:
      min_delay_seconds: 60
      max_delay_seconds: 120
    persist: true
    signing:
      keyid: my-key
      scope: read

decision_logs:
  service: policy-service
  reporting:
    min_delay_seconds: 30
    max_delay_seconds: 60

status:
  service: policy-service

Running OPA with Bundles

# Run OPA server with bundle configuration
opa run --server --config-file opa-config.yaml

# Run with bundle from HTTP
opa run --server --set bundles.main.resource=https://example.com/bundle.tar.gz

# Run with bundle from S3
opa run --server \
  --set services.s3.url=https://s3.amazonaws.com \
  --set bundles.main.service=s3 \
  --set bundles.main.resource=my-bucket/bundle.tar.gz

# Run with local bundle
opa run --server --bundle bundle.tar.gz

Bundle Server (Simple HTTP)

# bundle_server.py
from flask import Flask, send_file
from datetime import datetime
import os

app = Flask(__name__)
BUNDLE_DIR = '/opt/opa/bundles'

@app.route('/bundles/<environment>/bundle.tar.gz')
def get_bundle(environment):
    bundle_path = os.path.join(BUNDLE_DIR, environment, 'bundle.tar.gz')
    if not os.path.exists(bundle_path):
        return {'error': 'Bundle not found'}, 404

    return send_file(
        bundle_path,
        mimetype='application/gzip',
        as_attachment=True,
        download_name='bundle.tar.gz'
    )

@app.route('/health')
def health():
    return {'status': 'healthy', 'timestamp': datetime.utcnow().isoformat()}

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

Automated Bundle Publishing

# .github/workflows/publish-bundle.yml
name: Publish Policy Bundle

on:
  push:
    branches:
      - main
    paths:
      - 'policies/**'

jobs:
  build-and-publish:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Install OPA
        run: |
          curl -L -o opa https://openpolicyagent.org/downloads/latest/opa_linux_amd64
          chmod +x opa
          sudo mv opa /usr/local/bin/

      - name: Run Tests
        run: opa test policies/ -v

      - name: Build Bundle
        run: |
          opa build -b policies/ -o bundle.tar.gz
          echo "BUNDLE_SHA=$(sha256sum bundle.tar.gz | cut -d' ' -f1)" >> $GITHUB_ENV

      - name: Sign Bundle
        run: |
          echo "${{ secrets.SIGNING_KEY }}" > private_key.pem
          opa build -b --signing-key private_key.pem policies/ -o bundle.tar.gz
          rm private_key.pem

      - name: Publish to S3
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
        run: |
          aws s3 cp bundle.tar.gz s3://policy-bundles/production/bundle.tar.gz

          # Create manifest
          cat > .manifest << EOF
          {
            "revision": "${{ github.sha }}",
            "sha256": "${{ env.BUNDLE_SHA }}",
            "published_at": "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
          }
          EOF

          aws s3 cp .manifest s3://policy-bundles/production/.manifest

      - name: Create Release
        uses: actions/create-release@v1
        env:
          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
        with:
          tag_name: v${{ github.run_number }}
          release_name: Policy Bundle v${{ github.run_number }}
          body: |
            Policy bundle built from commit ${{ github.sha }}
            SHA256: ${{ env.BUNDLE_SHA }}

Auditing and Remediation

Audit Workflow

flowchart TB
    A[Scheduled Audit] --> B[Scan Resources]
    B --> C[Apply Policies]
    C --> D{Violations Found?}

    D -->|No| E[Generate Report]
    D -->|Yes| F[Categorise Violations]

    F --> G{Severity}
    G -->|Critical| H[Alert Security Team]
    G -->|High| I[Create Tickets]
    G -->|Medium/Low| J[Log for Review]

    H --> K[Automatic Remediation?]
    I --> K
    J --> E

    K -->|Yes| L[Apply Fixes]
    K -->|No| M[Manual Review Required]

    L --> N[Verify Fix]
    M --> N
    N --> E
    E --> O[Store Results]

Gatekeeper Audit

# Enable audit (runs every 60 seconds by default)
kubectl get constrainttemplates

# View audit results
kubectl get constraints -o json | \
  jq '.items[] | select(.status.totalViolations > 0) | {
    name: .metadata.name,
    violations: .status.totalViolations,
    details: .status.violations
  }'

# Export violations to JSON
kubectl get constraints -o json > audit-results.json

# Check specific constraint
kubectl describe constraint require-resource-limits

OPA Decision Logging

# opa-config.yaml with decision logging
decision_logs:
  service: logging-service
  reporting:
    min_delay_seconds: 10
    max_delay_seconds: 30
    upload_size_limit_bytes: 524288

  # Log only decisions
  plugin: decision_logs_plugin

# Send to external service
services:
  logging-service:
    url: https://logging.example.com
    credentials:
      bearer:
        token: ${LOGGING_TOKEN}

Remediation Scripts

#!/bin/bash
# remediate-missing-labels.sh

# Get pods missing required labels
kubectl get pods --all-namespaces -o json | \
  jq -r '.items[] |
    select(.metadata.labels.app == null or
           .metadata.labels.environment == null) |
    {namespace: .metadata.namespace, name: .metadata.name}' | \
while read -r pod; do
  namespace=$(echo "$pod" | jq -r '.namespace')
  name=$(echo "$pod" | jq -r '.name')

  echo "Adding labels to pod: $namespace/$name"
  kubectl label pod "$name" -n "$namespace" \
    app=unknown \
    environment=unknown \
    --overwrite
done

#!/usr/bin/env python3
# remediate_violations.py

import json
import subprocess
import sys
from typing import Dict, List

def get_violations() -> List[Dict]:
    """Fetch current policy violations from Gatekeeper."""
    result = subprocess.run(
        ['kubectl', 'get', 'constraints', '-o', 'json'],
        capture_output=True,
        text=True
    )

    if result.returncode != 0:
        print(f"Error fetching constraints: {result.stderr}")
        sys.exit(1)

    constraints = json.loads(result.stdout)
    violations = []

    for item in constraints.get('items', []):
        if item.get('status', {}).get('totalViolations', 0) > 0:
            violations.extend(item['status']['violations'])

    return violations

def remediate_resource_limits(violation: Dict):
    """Add default resource limits to pods."""
    namespace = violation['namespace']
    name = violation['name']
    kind = violation['kind']

    patch = {
        'spec': {
            'template': {
                'spec': {
                    'containers': [{
                        'name': 'CONTAINER_NAME',  # Need to fetch actual name
                        'resources': {
                            'limits': {
                                'cpu': '500m',
                                'memory': '512Mi'
                            },
                            'requests': {
                                'cpu': '250m',
                                'memory': '256Mi'
                            }
                        }
                    }]
                }
            }
        }
    }

    print(f"Remediating {kind} {namespace}/{name}")
    # Apply patch (in production, be more careful!)
    # kubectl.patch(kind, name, namespace, patch)

def remediate_security_context(violation: Dict):
    """Add security context to pods."""
    namespace = violation['namespace']
    name = violation['name']

    patch = {
        'spec': {
            'template': {
                'spec': {
                    'securityContext': {
                        'runAsNonRoot': True,
                        'runAsUser': 1000,
                        'fsGroup': 1000
                    },
                    'containers': [{
                        'name': 'CONTAINER_NAME',
                        'securityContext': {
                            'allowPrivilegeEscalation': False,
                            'readOnlyRootFilesystem': True,
                            'capabilities': {
                                'drop': ['ALL']
                            }
                        }
                    }]
                }
            }
        }
    }

    print(f"Adding security context to {namespace}/{name}")
    # Apply patch

def main():
    violations = get_violations()

    print(f"Found {len(violations)} violations")

    for violation in violations:
        enforcement_action = violation.get('enforcementAction', 'deny')
        message = violation.get('message', '')

        print(f"\nViolation: {message}")
        print(f"Resource: {violation.get('kind')} {violation.get('namespace')}/{violation.get('name')}")

        # Route to appropriate remediation
        if 'resource limits' in message.lower():
            remediate_resource_limits(violation)
        elif 'security context' in message.lower():
            remediate_security_context(violation)
        else:
            print(f"No automatic remediation available")

if __name__ == '__main__':
    main()

Continuous Compliance Monitoring

# cronjob-policy-audit.yaml
apiVersion: batch/v1
kind: CronJob
metadata:
  name: policy-audit
  namespace: security
spec:
  schedule: "0 */6 * * *"  # Every 6 hours
  jobTemplate:
    spec:
      template:
        spec:
          serviceAccountName: policy-auditor
          containers:
          - name: auditor
            image: openpolicyagent/opa:latest
            command:
            - /bin/sh
            - -c
            - |
              # Fetch all resources
              kubectl get pods --all-namespaces -o json > pods.json
              kubectl get deployments --all-namespaces -o json > deployments.json

              # Run policies
              opa eval \
                --data policies/ \
                --input pods.json \
                --format pretty \
                'data.kubernetes.admission.deny' > violations.txt

              # Send to monitoring system
              if [ -s violations.txt ]; then
                curl -X POST https://monitoring.example.com/api/violations \
                  -H "Content-Type: application/json" \
                  -d @violations.txt
              fi
            volumeMounts:
            - name: policies
              mountPath: /policies
          volumes:
          - name: policies
            configMap:
              name: opa-policies
          restartPolicy: OnFailure

Quick Reference

Common Rego Patterns

# Check if field exists
has_field(obj, field) if {
    obj[field]
}

# Get field with default
get_default(obj, field, default_value) := value if {
    value := obj[field]
} else := default_value

# String matching
starts_with_approved_prefix(str) if {
    startswith(str, "approved-")
}

# Array contains
array_contains(arr, item) if {
    arr[_] == item
}

# Set membership
in_allowed_set(value, allowed) if {
    value in allowed
}

# Iterate over array
deny contains msg if {
    some item in input.items
    not item.valid
    msg := sprintf("Invalid item: %v", [item])
}

# Iterate with index
deny contains msg if {
    some i, container in input.spec.containers
    not container.valid
    msg := sprintf("Container %d invalid", [i])
}

# Check all items satisfy condition
all_items_valid if {
    every item in input.items {
        item.status == "valid"
    }
}

# Check any item satisfies condition
any_item_invalid if {
    some item in input.items
    item.status == "invalid"
}

# Comprehension
valid_names := [name |
    some item in input.items
    item.valid
    name := item.name
]

# Object comprehension
name_map := {name: item |
    some item in input.items
    name := item.name
}

# Tree traversal (Rego forbids recursive rules/functions — use the
# built-in walk() to descend nested structures instead)
# Counts every value node in the document, at any depth.
node_count(obj) := count([value | walk(obj, [_, value])])

Conftest Commands

# Basic testing
conftest test <file>
conftest test <directory>
conftest test -p <policy-dir> <file>

# Output formats
conftest test --output json <file>
conftest test --output table <file>
conftest test --output junit <file>
conftest test --output github <file>

# Control behaviour
conftest test --fail-on-warn <file>
conftest test --all-namespaces <file>
conftest test --combine <file1> <file2>
conftest test --trace <file>

# File types
conftest test --parser json <file>
conftest test --parser yaml <file>
conftest test --parser toml <file>
conftest test Dockerfile  # Auto-detected

# Update policies
conftest pull <url>
conftest pull --update <url>

# Verify
conftest verify --policy <dir>

OPA Commands

# Evaluate policies
opa eval -d policy.rego -i input.json 'data.package.rule'
opa eval --bundle bundle.tar.gz -i input.json 'data.main.allow'

# Run tests
opa test policies/
opa test -v policies/
opa test --coverage policies/
opa test --run <test-pattern> policies/

# Format and check
opa fmt policies/
opa fmt -w policies/  # Write changes
opa check policies/

# Build bundles
opa build -b policies/ -o bundle.tar.gz
opa build -b -O 3 -e kubernetes/admission policies/  # Optimisation needs an entrypoint

# Run server
opa run --server
opa run --server --addr :8181
opa run --server --bundle bundle.tar.gz
opa run --server --config-file config.yaml

# Benchmarking
opa bench -d policy.rego 'data.package.rule'

Gatekeeper kubectl Commands

# Constraint Templates
kubectl get constrainttemplates
kubectl describe constrainttemplate <name>
kubectl apply -f constraint-template.yaml

# Constraints
kubectl get constraints
kubectl get <constraint-kind> <name> -o yaml
kubectl describe <constraint-kind> <name>

# Check violations
kubectl get constraints -o json | jq '.items[] | select(.status.totalViolations > 0)'

# Audit
kubectl get <constraint-kind> -o jsonpath='{.status.violations}'

# Config
kubectl get config -n gatekeeper-system
kubectl edit config -n gatekeeper-system

# Mutations
kubectl get assign,assignmetadata -A

Common Issues and Solutions

Policy Not Applied

Problem: Conftest doesn't detect violations that should exist.

Solutions:

# Check namespace
conftest test --namespace main file.yaml  # Default namespace
conftest test --all-namespaces file.yaml  # Check all

# Verify policy syntax
opa check policies/

# Debug evaluation
conftest test --trace file.yaml

# Check which policies are loaded
conftest test -p policies/ --show-builtin-errors file.yaml

Gatekeeper Not Blocking Resources

Problem: Resources are created despite constraint violations.

Solutions:

# Check constraint is enforced (not dryrun)
kubectl get constraint <name> -o yaml
# Look for enforcementAction: deny (not dryrun)

# Verify constraint matches the resource
kubectl get constraint <name> -o jsonpath='{.spec.match}'

# Check Gatekeeper is running
kubectl get pods -n gatekeeper-system

# View webhook configuration
kubectl get validatingwebhookconfigurations | grep gatekeeper

# Check logs
kubectl logs -n gatekeeper-system -l control-plane=controller-manager

Rego Syntax Errors

Problem: Policies fail with cryptic errors.

Solutions:

# OPA 1.0+ runs Rego v1 by default: `if` and `contains` are built-in
# keywords, so no `import future.keywords.*` is needed. On OPA < 1.0,
# add `import rego.v1` (or the individual future imports) to opt in.

# Common mistakes:

# ❌ Wrong: missing if/contains keyword (the #1 v1 gotcha)
deny[msg] {                  # rego_parse_error under v1
    input.user == "admin"
}

# ✓ Correct: partial set rules need `contains` + `if`
deny contains msg if {
    input.user == "admin"
    msg := "admin denied"
}

# ❌ Wrong: using = for comparison
deny contains msg if {
    input.user = "admin"  # Assignment/unification, not comparison!
    msg := "x"
}

# ✓ Correct: use ==
deny contains msg if {
    input.user == "admin"
    msg := "x"
}

# ❌ Wrong: undefined variable
deny contains msg if {
    container.name == "test"  # container not defined
    msg := "x"
}

# ✓ Correct: iterate first
deny contains msg if {
    some container in input.spec.containers
    container.name == "test"
    msg := "x"
}

# ❌ Wrong: missing if keyword on a complete rule
allow {
    input.user == "admin"  # Old (pre-v1) syntax — now a parse error
}

# ✓ Correct: use if
allow if {
    input.user == "admin"
}

Performance Issues

Problem: Policy evaluation is slow.

Solutions:

# Use indexing for large datasets
# ❌ Slow: linear search
deny contains msg if {
    some user in data.users
    user.id == input.user_id
    msg := "denied"
}

# ✓ Fast: indexed lookup
deny contains msg if {
    user := data.users[input.user_id]
    # ... conditions
    msg := "denied"
}

# Avoid unnecessary iterations
# ❌ Slow: nested loops
deny contains msg if {
    some container in input.spec.containers
    some image in data.approved_images
    container.image != image
    msg := "denied"
}

# ✓ Fast: use sets
approved_images := {"nginx:1.21", "redis:6.2"}
deny contains msg if {
    some container in input.spec.containers
    not container.image in approved_images
    msg := "denied"
}

# Pre-compute expensive operations
# Extract expensive computations to separate rules

Bundle Loading Failures

Problem: OPA server fails to load policy bundles.

Solutions:

# Check bundle format
opa build -b policies/ -o bundle.tar.gz
tar -tzf bundle.tar.gz  # Verify contents

# Verify bundle signature
opa build --verification-key public_key.pem --bundle bundle.tar.gz

# Check OPA logs
docker logs <opa-container>

# Test bundle locally
opa run --server --bundle bundle.tar.gz
curl http://localhost:8181/v1/policies  # Should list policies

# Verify service configuration
opa run --server --config-file config.yaml --log-level debug

Input Data Format Mismatches

Problem: Policies don't match expected input structure.

Solutions:

# Debug with opa eval
opa eval -d policy.rego -i input.json --format pretty 'input'

# Check actual input structure
opa eval -i input.json 'input' | jq

# Use object.get for optional fields
# ❌ Fails if field missing
deny contains msg if {
    input.metadata.labels.app == "test"
    msg := "x"
}

# ✓ Safe with default
deny contains msg if {
    labels := object.get(input.metadata, "labels", {})
    labels.app == "test"
    msg := "x"
}

# Handle both Pod and Deployment
deny contains msg if {
    input.kind in ["Pod", "Deployment"]
    containers := object.get(input.spec, "containers", [])
    containers == []
    msg := "x"
}

Testing Challenges

Problem: Writing comprehensive tests is difficult.

Solutions:

# Use table-driven tests.
# Note: any rule named `test_*` is auto-run by `opa test` with no
# arguments, so the per-case helper must NOT start with `test_` —
# call it `check_case` (or similar). Also avoid naming the fixture
# `input`; v1 rejects rules/locals that shadow the `input` document.
test_deny_privileged_containers if {
    cases := [
        {"privileged": true, "should_deny": true},
        {"privileged": false, "should_deny": false},
        {"privileged": null, "should_deny": false},
    ]

    every case in cases {
        check_case(case)
    }
}

check_case(case) if {
    fixture := {"securityContext": {"privileged": case.privileged}}
    violations := deny with input as fixture

    case.should_deny
    count(violations) > 0
}

check_case(case) if {
    fixture := {"securityContext": {"privileged": case.privileged}}
    violations := deny with input as fixture

    not case.should_deny
    count(violations) == 0
}

# Test with realistic fixtures. Rego has no file-reading built-in, so
# load fixtures as data at eval time instead, e.g.:
#   opa test -d policy.rego -d test.rego -d fixtures/
# and reference them via `data.fixtures.production_pod`.
test_production_pod if {
    violations := deny with input as data.fixtures.production_pod
    count(violations) == 0
}

CI/CD Integration Issues

Problem: Policy checks fail intermittently in CI.

Solutions:

# Pin tool versions
- name: Install Conftest
  run: |
    VERSION="0.68.2"
    wget https://github.com/open-policy-agent/conftest/releases/download/v${VERSION}/conftest_${VERSION}_Linux_x86_64.tar.gz

# Cache installations
- uses: actions/cache@v3
  with:
    path: ~/.local/bin
    key: ${{ runner.os }}-conftest-0.68.2

# Set explicit timeouts
- name: Run Policy Tests
  timeout-minutes: 5
  run: conftest test manifests/

# Handle missing files gracefully
- name: Test Kubernetes Manifests
  run: |
    if ls k8s/*.yaml 1> /dev/null 2>&1; then
      conftest test k8s/*.yaml
    else
      echo "No manifests to test"
    fi

Related Topics: Container Security, Kubernetes Advanced, Terraform, Vault, SOPS, Security Patterns