CKS — Rego Policies for CKS

9 minute read

Rego Policies for CKS

CKS Exam Domain: Minimize Microservice Vulnerabilities (20%)

Rego is the policy language used by OPA and Gatekeeper. For the CKS exam, you don’t need to be a Rego expert — you need to read, understand, and write simple Rego rules that enforce common Kubernetes security policies.

Rego Language Fundamentals

What Is Rego?

Rego (pronounced “ray-go”) is a declarative query language designed for policy evaluation. It operates on structured data (JSON/YAML converted to JSON).

Key characteristics:

  • Declarative — you describe what violations look like, not step-by-step logic.
  • Set-based — rules evaluate to sets of results.
  • Unification — variables are bound by matching, not by assignment.

The Gatekeeper Input Object

When Gatekeeper receives an admission request, it provides the full context via input:

{
  "input": {
    "review": {
      "kind": {
        "kind": "Pod",
        "group": "",
        "version": "v1"
      },
      "object": {
        "metadata": {
          "name": "my-pod",
          "namespace": "default",
          "labels": {
            "app": "web"
          }
        },
        "spec": {
          "containers": [
            {
              "name": "nginx",
              "image": "nginx:latest",
              "securityContext": {
                "privileged": false,
                "runAsUser": 1000
              },
              "resources": {
                "limits": {
                  "cpu": "500m",
                  "memory": "128Mi"
                }
              }
            }
          ]
        }
      },
      "oldObject": null,
      "operation": "CREATE",
      "userInfo": {
        "username": "developer@example.com",
        "groups": ["system:authenticated"]
      }
    },
    "parameters": {
      "labels": ["app", "team"]
    }
  }
}

Key Paths

Path Description
input.review.object The incoming Kubernetes resource
input.review.object.metadata Name, namespace, labels, annotations
input.review.object.spec Resource spec (containers, volumes, etc.)
input.review.oldObject Previous version (for UPDATE operations)
input.review.operation CREATE, UPDATE, DELETE
input.review.userInfo Who submitted the request
input.parameters Values from the Constraint’s spec.parameters

The Violation Rule

In Gatekeeper, the Rego rule must be named violation. Every truthy evaluation of a violation rule produces a policy violation.

Basic Pattern

package <template-name>

violation[{"msg": msg}] {
  # condition 1
  # condition 2
  # ...all conditions must be true for violation to fire
  msg := "Human-readable violation message"
}
  • All statements in a rule body are AND-ed together.
  • Multiple violation rules with the same head are OR-ed (any one can fire).
  • The rule produces a violation when all conditions in the body are satisfied.

Multiple Rules = OR

# Violation if container is privileged
violation[{"msg": msg}] {
  container := input.review.object.spec.containers[_]
  container.securityContext.privileged == true
  msg := sprintf("Container '%v' is privileged", [container.name])
}

# OR violation if init container is privileged
violation[{"msg": msg}] {
  container := input.review.object.spec.initContainers[_]
  container.securityContext.privileged == true
  msg := sprintf("Init container '%v' is privileged", [container.name])
}

Common CKS Rego Patterns

Pattern 1: Require Labels

package k8srequiredlabels

violation[{"msg": msg}] {
  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])
}

Line-by-line:

  • provided — builds a set of all label keys on the object.
  • required — builds a set of required label keys from parameters.
  • missing — set difference: required labels not found on the object.
  • If any are missing, produce a violation.

Pattern 2: Block Privileged Containers

package k8sblockprivileged

violation[{"msg": msg}] {
  container := input.review.object.spec.containers[_]
  container.securityContext.privileged == true
  msg := sprintf("Privileged container not allowed: %v", [container.name])
}

violation[{"msg": msg}] {
  container := input.review.object.spec.initContainers[_]
  container.securityContext.privileged == true
  msg := sprintf("Privileged init container not allowed: %v", [container.name])
}

Key detail: containers[_] iterates over all containers. The rule fires for each container that matches.

Pattern 3: Enforce Allowed Registries (String Prefix Check)

package k8sallowedregistries

violation[{"msg": msg}] {
  container := input.review.object.spec.containers[_]
  not registry_allowed(container.image)
  msg := sprintf("Image '%v' is not from an allowed registry. Allowed: %v",
    [container.image, input.parameters.registries])
}

registry_allowed(image) {
  registry := input.parameters.registries[_]
  startswith(image, registry)
}

Key Rego concepts:

  • not — negation: the image is NOT from an allowed registry.
  • Helper function registry_allowed — returns true if any registry prefix matches.
  • startswith(image, registry) — built-in string function.

Pattern 4: Require Resource Limits

package k8srequirelimits

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

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

Two separate rules — one for CPU, one for memory. Both fire independently.

Pattern 5: Block hostPath Volumes

package k8sblockhostpath

violation[{"msg": msg}] {
  volume := input.review.object.spec.volumes[_]
  volume.hostPath
  msg := sprintf("hostPath volume '%v' is not allowed", [volume.name])
}

Security impact: hostPath volumes expose the host filesystem to the container. An attacker can read /etc/shadow, write to /var/run/docker.sock, or access any host file.

Pattern 6: Block Latest Tag

package k8sblocklatest

violation[{"msg": msg}] {
  container := input.review.object.spec.containers[_]
  endswith(container.image, ":latest")
  msg := sprintf("Container '%v' uses :latest tag — use a specific version", [container.name])
}

violation[{"msg": msg}] {
  container := input.review.object.spec.containers[_]
  not contains(container.image, ":")
  msg := sprintf("Container '%v' has no tag — use a specific version", [container.name])
}

Images without a tag default to :latest, so we check for both cases.

Pattern 7: Require runAsNonRoot

package k8srunasnonroot

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

This checks both the container-level and Pod-level securityContext.

Rego Syntax Reference for CKS

Iteration

# Iterate over all containers
container := input.review.object.spec.containers[_]

# Iterate over all volumes
volume := input.review.object.spec.volumes[_]

# Iterate with index
container := input.review.object.spec.containers[i]

Comparison

# Equality
container.securityContext.privileged == true

# Inequality (negation of equality)
not container.securityContext.runAsNonRoot == true

# Numeric comparison
input.review.object.spec.replicas > 5

String Functions

# Prefix check
startswith(image, "gcr.io/")

# Suffix check
endswith(image, ":latest")

# Contains
contains(image, "malicious")

# Format string
msg := sprintf("Error: %v in %v", [detail, context])

Set Operations

# Create a set from an array
labels := {l | l := input.parameters.labels[_]}

# Set difference
missing := required - provided

# Count elements
count(missing) > 0

Negation

# "not" negates a condition or helper function
not container.resources.limits.cpu
not registry_allowed(container.image)

Helper Functions

# Define a helper (returns true if any condition matches)
registry_allowed(image) {
  registry := input.parameters.registries[_]
  startswith(image, registry)
}

# Use in a rule
violation[{"msg": msg}] {
  not registry_allowed(container.image)
  ...
}

Debugging Rego

Using opa eval Locally

Install OPA CLI:

# macOS
brew install opa

# Linux
curl -L -o opa https://openpolicyagent.org/downloads/latest/opa_linux_amd64
chmod +x opa && sudo mv opa /usr/local/bin/

Create a test input file:

// input.json
{
  "review": {
    "object": {
      "metadata": {
        "name": "test-pod",
        "labels": {}
      },
      "spec": {
        "containers": [
          {
            "name": "nginx",
            "image": "nginx:latest",
            "securityContext": {
              "privileged": true
            }
          }
        ]
      }
    }
  },
  "parameters": {
    "labels": ["app", "team"]
  }
}

Create the policy file:

# policy.rego
package k8srequiredlabels

violation[{"msg": msg}] {
  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])
}

Evaluate:

opa eval --input input.json --data policy.rego "data.k8srequiredlabels.violation"

Expected output:

{
  "result": [
    {
      "expressions": [
        {
          "value": [
            {"msg": "Missing required labels: {\"app\", \"team\"}"}
          ]
        }
      ]
    }
  ]
}

Using opa test

Write test cases:

# policy_test.rego
package k8srequiredlabels

test_missing_labels {
  results := violation with input as {
    "review": {"object": {"metadata": {"labels": {}}}},
    "parameters": {"labels": ["app"]}
  }
  count(results) > 0
}

test_labels_present {
  results := violation with input as {
    "review": {"object": {"metadata": {"labels": {"app": "web"}}}},
    "parameters": {"labels": ["app"]}
  }
  count(results) == 0
}

Run tests:

opa test policy.rego policy_test.rego -v

data.k8srequiredlabels.test_missing_labels: PASS (1.234ms)
data.k8srequiredlabels.test_labels_present: PASS (0.567ms)
--------------------------------------------------------------------------------
PASS: 2/2

Testing Rego Policies Before Deploying

Workflow

  1. Write the Rego policy locally.
  2. Test with opa eval and opa test.
  3. Embed in a ConstraintTemplate YAML.
  4. Deploy the ConstraintTemplate with kubectl apply.
  5. Create a Constraint in dryrun mode.
  6. Verify violations are reported correctly.
  7. Switch to deny enforcement.

Using Gatekeeper’s gator CLI

gator is Gatekeeper’s testing tool — it evaluates ConstraintTemplates and Constraints locally without a cluster:

# Install gator
go install github.com/open-policy-agent/gatekeeper/v3/cmd/gator@latest

# Test a template + constraint + test resource
gator test --filename template.yaml --filename constraint.yaml --filename test-pod.yaml

CKS Exam Rego Tips

  1. Don’t memorise complex Rego — understand the violation pattern and input.review.object path.

  2. Pattern-match from Kubernetes documentation — the exam provides access to Kubernetes docs. The Gatekeeper library examples are well-documented.

  3. Start with a working template and modify — it’s faster to adapt an existing ConstraintTemplate than to write one from scratch.

  4. Key paths to remember:

    input.review.object.metadata.name
input.review.object.metadata.namespace
input.review.object.metadata.labels
input.review.object.spec.containers[_]
input.review.object.spec.initContainers[_]
input.review.object.spec.volumes[_]
input.review.object.spec.containers[_].securityContext
input.review.object.spec.containers[_].image
input.review.object.spec.containers[_].resources.limits
input.parameters.<param-name>

  5. Keep rules simple — one violation per check. Don’t combine multiple checks in one rule.

  6. Always check both containers and initContainers — common exam trap.

  7. Use sprintf for clear messages — it helps with debugging and the exam grader can see your intent.

Complete ConstraintTemplate with Rego — End-to-End

Here is a complete, exam-ready example that blocks containers with dangerous capabilities:

apiVersion: templates.gatekeeper.sh/v1
kind: ConstraintTemplate
metadata:
  name: k8sblockdangerouscaps
spec:
  crd:
    spec:
      names:
        kind: K8sBlockDangerousCaps
      validation:
        openAPIV3Schema:
          type: object
          properties:
            blockedCapabilities:
              type: array
              items:
                type: string
  targets:
  - target: admission.k8s.gatekeeper.sh
    rego: |
      package k8sblockdangerouscaps

      violation[{"msg": msg}] {
        container := input.review.object.spec.containers[_]
        cap := container.securityContext.capabilities.add[_]
        blocked := input.parameters.blockedCapabilities[_]
        upper(cap) == upper(blocked)
        msg := sprintf("Container '%v' adds blocked capability: %v", [container.name, cap])
      }

      violation[{"msg": msg}] {
        container := input.review.object.spec.initContainers[_]
        cap := container.securityContext.capabilities.add[_]
        blocked := input.parameters.blockedCapabilities[_]
        upper(cap) == upper(blocked)
        msg := sprintf("Init container '%v' adds blocked capability: %v", [container.name, cap])
      }

Constraint:

apiVersion: constraints.gatekeeper.sh/v1beta1
kind: K8sBlockDangerousCaps
metadata:
  name: block-dangerous-caps
spec:
  enforcementAction: deny
  match:
    kinds:
    - apiGroups: [""]
      kinds: ["Pod"]
    excludedNamespaces: ["kube-system"]
  parameters:
    blockedCapabilities:
    - "SYS_ADMIN"
    - "NET_RAW"
    - "SYS_PTRACE"

Practice Questions

Question 1

Write a Rego violation rule that blocks Pods with hostNetwork: true.

Show Answer 
package k8sblockhostnetwork

violation[{"msg": msg}] {
  input.review.object.spec.hostNetwork == true
  msg := sprintf("Pod '%v' uses hostNetwork which is not allowed",
    [input.review.object.metadata.name])
}

hostNetwork: true gives the Pod access to the node’s network namespace, including all listening services and network interfaces. This is a significant security risk.

Question 2

What does containers[_] mean in Rego?

Show Answer

containers[_] is Rego’s iteration syntax. The _ is a wildcard variable that iterates over all elements of the containers array.

container := input.review.object.spec.containers[_]

This binds container to each element of the containers array in turn. If the Pod has 3 containers, the rule evaluates 3 times — once for each container.

It’s equivalent to a for each loop in imperative languages.

Question 3

Fix this Rego policy — it’s supposed to require memory limits but doesn’t work:

package k8srequirelimits

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

The issue is that when memory is not set, the field doesn’t exist — it’s not an empty string. Comparing to "" won’t catch missing fields.

Fixed version:

package k8srequirelimits

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

Use not container.resources.limits.memory to check for the absence of the field. This also handles cases where resources or limits are entirely missing.

Question 4

Write a complete ConstraintTemplate that blocks Pods mounting /var/run/docker.sock as a hostPath volume.

Show Answer 
apiVersion: templates.gatekeeper.sh/v1
kind: ConstraintTemplate
metadata:
  name: k8sblockdockersock
spec:
  crd:
    spec:
      names:
        kind: K8sBlockDockerSock
  targets:
  - target: admission.k8s.gatekeeper.sh
    rego: |
      package k8sblockdockersock

      violation[{"msg": msg}] {
        volume := input.review.object.spec.volumes[_]
        volume.hostPath.path == "/var/run/docker.sock"
        msg := sprintf("Volume '%v' mounts Docker socket — this is not allowed", [volume.name])
      }

Mounting the Docker socket gives the container full control over the Docker daemon, enabling container escape and host compromise.

Question 5

How do you pass parameters from a Constraint to the Rego policy, and how are they accessed?

Show Answer

In the ConstraintTemplate — define the parameter schema:

spec:
  crd:
    spec:
      validation:
        openAPIV3Schema:
          type: object
          properties:
            maxReplicas:
              type: integer

In the Constraint — provide parameter values:

spec:
  parameters:
    maxReplicas: 3

In the Rego policy — access via input.parameters:

violation[{"msg": msg}] {
  input.review.object.spec.replicas > input.parameters.maxReplicas
  msg := sprintf("Replicas %v exceeds max allowed %v",
    [input.review.object.spec.replicas, input.parameters.maxReplicas])
}

The flow: Constraint spec.parameters → Gatekeeper → Rego input.parameters.

Previous Next
← Constraints Kyverno Alternative →

Updated:

You may also enjoy