Security Policy as Code

The 3-Week Bottleneck: When Security Reviews Killed Velocity

"Sorry, security review is backed up. Your deployment will have to wait 2-3 weeks."

That email destroyed our sprint. We'd just finished a critical feature for a major customer. Everything worked. Tests passed. Code reviewed. Ready to deploy. But we needed security approval first.

The security team had become a bottleneck. Not because they were slow or incompetent—they were drowning. Every deployment required manual review:

• Is the container image from an approved registry?

• Does it run as non-root?

• Are resource limits set?

• Is the database encrypted?

• Are secrets properly managed?

• Does it comply with PCI-DSS requirements?

Manual checklists for 50+ deployments per day. The security team was overwhelmed, and engineering velocity had ground to a halt.

Then I discovered Policy as Code. Instead of humans checking policies manually, we encoded every security requirement as executable code. Automated policy gates that could evaluate thousands of deployments per day, instantly.

The result? Security approvals went from 3 weeks to 3 minutes. The security team shifted from gatekeepers to policy developers. Engineering velocity 10x'd. And our security posture actually improved because policies were enforced consistently, automatically, on every single deployment.

This article covers how I built a comprehensive Policy as Code framework using Open Policy Agent that transformed our security program.

What You'll Learn

• Policy as Code fundamentals and benefits

• Open Policy Agent (OPA) deep dive

• Rego policy language essentials

• Kubernetes admission control with OPA/Gatekeeper

• CI/CD policy enforcement

• Terraform policy validation

• Custom policy development

• Policy testing and versioning

The Policy as Code Revolution

Traditional security: humans enforcing rules manually. Policy as Code: rules enforced automatically by machines.

Benefits:

• Speed: Instant policy evaluation vs. days/weeks

• Consistency: Same rules applied every time

• Scale: Handle unlimited deployments

• Auditability: Every decision logged

• Version control: Policies in Git

• Testing: Automated policy testing

Open Policy Agent (OPA)

OPA is the industry standard for Policy as Code. It uses Rego, a declarative policy language.

OPA Installation

# macOS
brew install opa

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

# Docker
docker run -p 8181:8181 openpolicyagent/opa:latest run --server

Rego Basics

# policies/example.rego
package example

# Simple boolean policy
allow {
  input.user == "admin"
}

# Policy with conditions
allow {
  input.user == "developer"
  input.action == "read"
}

# Deny policy
deny[msg] {
  not input.authenticated
  msg := "User must be authenticated"
}

# Complex logic
max_deployment_replicas := 10

deny[msg] {
  input.kind == "Deployment"
  replicas := input.spec.replicas
  replicas > max_deployment_replicas
  msg := sprintf("Too many replicas: %d (max: %d)", [replicas, max_deployment_replicas])
}

Testing Policies

# Create test data
cat > input.json <<EOF
{
  "user": "admin",
  "action": "write"
}
EOF

# Evaluate policy
opa eval -i input.json -d policies/example.rego "data.example.allow"

# Run tests
opa test policies/ -v

Kubernetes Admission Control with Gatekeeper

Gatekeeper brings OPA to Kubernetes as an admission controller.

Installing Gatekeeper

# Install Gatekeeper
kubectl apply -f https://raw.githubusercontent.com/open-policy-agent/gatekeeper/release-3.14/deploy/gatekeeper.yaml

# Verify installation
kubectl get pods -n gatekeeper-system

Constraint Templates

Constraint Templates define reusable policies:

# constraint-templates/required-labels.yaml
apiVersion: templates.gatekeeper.sh/v1
kind: ConstraintTemplate
metadata:
  name: k8srequiredlabels
spec:
  crd:
    spec:
      names:
        kind: K8sRequiredLabels
      validation:
        openAPIV3Schema:
          type: object
          properties:
            labels:
              type: array
              items:
                type: string
  targets:
    - 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])
        }

Applying Constraints

# constraints/require-labels.yaml
apiVersion: constraints.gatekeeper.sh/v1beta1
kind: K8sRequiredLabels
metadata:
  name: must-have-required-labels
spec:
  match:
    kinds:
      - apiGroups: [""]
        kinds: ["Namespace"]
      - apiGroups: ["apps"]
        kinds: ["Deployment", "StatefulSet"]
  parameters:
    labels:
      - "app"
      - "environment"
      - "owner"

Production-Ready Kubernetes Policies

1. Container Image Policy

# constraint-templates/allowed-repos.yaml
apiVersion: templates.gatekeeper.sh/v1
kind: ConstraintTemplate
metadata:
  name: k8sallowedrepos
spec:
  crd:
    spec:
      names:
        kind: K8sAllowedRepos
      validation:
        openAPIV3Schema:
          type: object
          properties:
            repos:
              type: array
              items:
                type: string
  targets:
    - target: admission.k8s.gatekeeper.sh
      rego: |
        package k8sallowedrepos
        
        violation[{"msg": msg}] {
          container := input.review.object.spec.containers[_]
          satisfied := [good | repo = input.parameters.repos[_] ; good = startswith(container.image, repo)]
          not any(satisfied)
          msg := sprintf("Container image %v is not from an approved registry", [container.image])
        }
        
        violation[{"msg": msg}] {
          container := input.review.object.spec.initContainers[_]
          satisfied := [good | repo = input.parameters.repos[_] ; good = startswith(container.image, repo)]
          not any(satisfied)
          msg := sprintf("Init container image %v is not from an approved registry", [container.image])
        }

# constraints/allowed-repos.yaml
apiVersion: constraints.gatekeeper.sh/v1beta1
kind: K8sAllowedRepos
metadata:
  name: allowed-container-registries
spec:
  match:
    kinds:
      - apiGroups: [""]
        kinds: ["Pod"]
    namespaces:
      - production
      - staging
  parameters:
    repos:
      - "gcr.io/my-company/"
      - "docker.io/my-company/"
      - "ghcr.io/my-company/"

2. Security Context Policy

# constraint-templates/pod-security.yaml
apiVersion: templates.gatekeeper.sh/v1
kind: ConstraintTemplate
metadata:
  name: k8spodsecurity
spec:
  crd:
    spec:
      names:
        kind: K8sPodSecurity
  targets:
    - target: admission.k8s.gatekeeper.sh
      rego: |
        package k8spodsecurity
        
        # Deny privileged containers
        violation[{"msg": msg}] {
          container := input_containers[_]
          container.securityContext.privileged
          msg := sprintf("Privileged container not allowed: %v", [container.name])
        }
        
        # Require runAsNonRoot
        violation[{"msg": msg}] {
          not input.review.object.spec.securityContext.runAsNonRoot
          msg := "Containers must run as non-root user"
        }
        
        # Require readOnlyRootFilesystem
        violation[{"msg": msg}] {
          container := input_containers[_]
          not container.securityContext.readOnlyRootFilesystem
          msg := sprintf("Container %v must have readOnlyRootFilesystem: true", [container.name])
        }
        
        # Block privilege escalation
        violation[{"msg": msg}] {
          container := input_containers[_]
          container.securityContext.allowPrivilegeEscalation
          msg := sprintf("Container %v must set allowPrivilegeEscalation: false", [container.name])
        }
        
        # Require capabilities drop ALL
        violation[{"msg": msg}] {
          container := input_containers[_]
          not has_drop_all_capabilities(container)
          msg := sprintf("Container %v must drop ALL capabilities", [container.name])
        }
        
        has_drop_all_capabilities(container) {
          container.securityContext.capabilities.drop[_] == "ALL"
        }
        
        input_containers[c] {
          c := input.review.object.spec.containers[_]
        }
        
        input_containers[c] {
          c := input.review.object.spec.initContainers[_]
        }

3. Resource Limits Policy

# constraint-templates/resource-limits.yaml
apiVersion: templates.gatekeeper.sh/v1
kind: ConstraintTemplate
metadata:
  name: k8srequireresources
spec:
  crd:
    spec:
      names:
        kind: K8sRequireResources
      validation:
        openAPIV3Schema:
          type: object
          properties:
            limits:
              type: object
              properties:
                cpu:
                  type: string
                memory:
                  type: string
  targets:
    - target: admission.k8s.gatekeeper.sh
      rego: |
        package k8srequireresources
        
        violation[{"msg": msg}] {
          container := input_containers[_]
          not container.resources.limits.cpu
          msg := sprintf("Container %v missing CPU limit", [container.name])
        }
        
        violation[{"msg": msg}] {
          container := input_containers[_]
          not container.resources.limits.memory
          msg := sprintf("Container %v missing memory limit", [container.name])
        }
        
        violation[{"msg": msg}] {
          container := input_containers[_]
          not container.resources.requests.cpu
          msg := sprintf("Container %v missing CPU request", [container.name])
        }
        
        violation[{"msg": msg}] {
          container := input_containers[_]
          not container.resources.requests.memory
          msg := sprintf("Container %v missing memory request", [container.name])
        }
        
        # Check if limits exceed maximum
        violation[{"msg": msg}] {
          container := input_containers[_]
          cpu_limit := container.resources.limits.cpu
          max_cpu := input.parameters.limits.cpu
          cpu_exceeds_max(cpu_limit, max_cpu)
          msg := sprintf("Container %v CPU limit %v exceeds maximum %v", [container.name, cpu_limit, max_cpu])
        }
        
        violation[{"msg": msg}] {
          container := input_containers[_]
          mem_limit := container.resources.limits.memory
          max_mem := input.parameters.limits.memory
          memory_exceeds_max(mem_limit, max_mem)
          msg := sprintf("Container %v memory limit %v exceeds maximum %v", [container.name, mem_limit, max_mem])
        }
        
        input_containers[c] {
          c := input.review.object.spec.containers[_]
        }
        
        cpu_exceeds_max(limit, max) {
          limit_value := parse_cpu(limit)
          max_value := parse_cpu(max)
          limit_value > max_value
        }
        
        memory_exceeds_max(limit, max) {
          limit_value := parse_memory(limit)
          max_value := parse_memory(max)
          limit_value > max_value
        }
        
        parse_cpu(cpu) = value {
          endswith(cpu, "m")
          value := to_number(trim_suffix(cpu, "m"))
        }
        
        parse_cpu(cpu) = value {
          not endswith(cpu, "m")
          value := to_number(cpu) * 1000
        }
        
        parse_memory(mem) = value {
          endswith(mem, "Gi")
          value := to_number(trim_suffix(mem, "Gi")) * 1024 * 1024 * 1024
        }
        
        parse_memory(mem) = value {
          endswith(mem, "Mi")
          value := to_number(trim_suffix(mem, "Mi")) * 1024 * 1024
        }

CI/CD Policy Enforcement

GitLab CI with OPA

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

variables:
  OPA_VERSION: "0.59.0"

opa_policy_check:
  stage: security
  image: openpolicyagent/opa:${OPA_VERSION}
  script:
    # Validate Kubernetes manifests
    - |
      for manifest in k8s/*.yaml; do
        echo "Validating $manifest..."
        opa eval -d policies/ -i $manifest "data.kubernetes.deny" --format pretty
        
        VIOLATIONS=$(opa eval -d policies/ -i $manifest "data.kubernetes.deny" --format raw | jq '. | length')
        if [ "$VIOLATIONS" -gt 0 ]; then
          echo "❌ Policy violations in $manifest"
          opa eval -d policies/ -i $manifest "data.kubernetes.deny"
          exit 1
        fi
      done
    
    # Run policy tests
    - opa test policies/ -v
  
  artifacts:
    reports:
      junit: opa-test-results.xml

conftest_check:
  stage: security
  image: openpolicyagent/conftest:latest
  script:
    # Check Dockerfiles
    - conftest test Dockerfile --policy policies/docker.rego
    
    # Check Terraform
    - conftest test terraform/ --policy policies/terraform.rego
    
    # Check Kubernetes
    - conftest test k8s/*.yaml --policy policies/kubernetes.rego

Docker Policy Example

# policies/docker.rego
package docker

deny[msg] {
  input[i].Cmd == "from"
  val := input[i].Value
  contains(val[i], "latest")
  msg := "Do not use 'latest' tag for base images"
}

deny[msg] {
  input[i].Cmd == "run"
  val := input[i].Value
  contains(val[_], "curl")
  msg := "Do not install curl in production images"
}

deny[msg] {
  input[i].Cmd == "user"
  val := input[i].Value
  val[_] == "root"
  msg := "Images should not run as root user"
}

deny[msg] {
  not user_defined
  msg := "No USER instruction found - image will run as root"
}

user_defined {
  input[_].Cmd == "user"
}

Terraform Policy Example

# policies/terraform.rego
package terraform

deny[msg] {
  resource := input.resource.aws_s3_bucket[name]
  resource.acl == "public-read"
  msg := sprintf("S3 bucket %v has public-read ACL", [name])
}

deny[msg] {
  resource := input.resource.aws_security_group[name]
  rule := resource.ingress[_]
  rule.cidr_blocks[_] == "0.0.0.0/0"
  rule.from_port == 22
  msg := sprintf("Security group %v allows SSH from anywhere", [name])
}

deny[msg] {
  resource := input.resource.aws_db_instance[name]
  resource.publicly_accessible == true
  msg := sprintf("Database %v is publicly accessible", [name])
}

deny[msg] {
  resource := input.resource.aws_instance[name]
  not resource.monitoring
  msg := sprintf("EC2 instance %v missing monitoring", [name])
}

Policy Testing

Always test your policies!

# policies/kubernetes_test.rego
package kubernetes

test_deployment_without_labels {
  input := {"review": {"object": {
    "kind": "Deployment",
    "metadata": {"name": "test"},
    "spec": {"replicas": 3}
  }}}
  
  result := deny with input as input
  count(result) > 0
}

test_deployment_with_labels {
  input := {"review": {"object": {
    "kind": "Deployment",
    "metadata": {
      "name": "test",
      "labels": {
        "app": "myapp",
        "environment": "production",
        "owner": "team-platform"
      }
    },
    "spec": {"replicas": 3}
  }}}
  
  result := deny with input as input
  count(result) == 0
}

test_privileged_container {
  input := {"review": {"object": {
    "kind": "Pod",
    "spec": {"containers": [{
      "name": "test",
      "securityContext": {"privileged": true}
    }]}
  }}}
  
  result := deny with input as input
  result[_] == "Privileged container not allowed: test"
}

# Run tests
opa test policies/ -v

# Output:
# policies/kubernetes_test.rego:
#   data.kubernetes.test_deployment_without_labels: PASS (1.2ms)
#   data.kubernetes.test_deployment_with_labels: PASS (0.8ms)
#   data.kubernetes.test_privileged_container: PASS (1.1ms)
# PASS: 3/3

Policy Dashboard and Monitoring

Track policy violations and compliance.

# prometheus-rules.yaml
apiVersion: monitoring.coreos.com/v1
kind: PrometheusRule
metadata:
  name: gatekeeper-metrics
spec:
  groups:
    - name: gatekeeper
      interval: 30s
      rules:
        - alert: HighPolicyViolationRate
          expr: rate(gatekeeper_violations_total[5m]) > 10
          for: 5m
          annotations:
            summary: "High rate of policy violations"
            description: "{{ $value }} violations per second"
        
        - alert: PolicyEnforcementFailure
          expr: gatekeeper_constraint_template_ingestion_count{status="error"} > 0
          annotations:
            summary: "Policy enforcement failure"
            description: "Failed to apply constraint template"

// scripts/policy-dashboard.js
const express = require('express');
const k8s = require('@kubernetes/client-node');

const app = express();
const kc = new k8s.KubeConfig();
kc.loadFromDefault();

const customApi = kc.makeApiClient(k8s.CustomObjectsApi);

app.get('/api/violations', async (req, res) => {
  try {
    // Get all constraints
    const constraints = await customApi.listClusterCustomObject(
      'constraints.gatekeeper.sh',
      'v1beta1',
      'k8srequiredlabels'
    );
    
    const violations = [];
    
    for (const constraint of constraints.body.items) {
      if (constraint.status?.violations) {
        violations.push({
          constraint: constraint.metadata.name,
          count: constraint.status.totalViolations,
          violations: constraint.status.violations
        });
      }
    }
    
    res.json({
      total: violations.reduce((sum, v) => sum + v.count, 0),
      by_constraint: violations
    });
  } catch (error) {
    res.status(500).json({ error: error.message });
  }
});

app.listen(3000);

Advanced Policy Patterns

1. Allow List Pattern

package kubernetes

# Default deny
deny[msg] {
  not is_allowed
  msg := "Resource not in allowlist"
}

is_allowed {
  input.review.object.metadata.namespace == "kube-system"
}

is_allowed {
  input.review.object.metadata.labels.approved == "true"
}

2. Exemption Pattern

package kubernetes

deny[msg] {
  not is_exempt
  violates_policy
  msg := "Policy violation"
}

is_exempt {
  input.review.object.metadata.annotations["policy.exempt"] == "true"
  input.review.object.metadata.annotations["policy.exempt.reason"]
  input.review.object.metadata.annotations["policy.exempt.approved-by"]
}

violates_policy {
  # Your policy logic
}

3. Environment-Specific Policies

package kubernetes

deny[msg] {
  is_production
  input.review.object.spec.replicas < 3
  msg := "Production deployments must have at least 3 replicas"
}

is_production {
  input.review.object.metadata.namespace == "production"
}

is_production {
  input.review.object.metadata.labels.environment == "production"
}

Policy as Code Best Practices

1. Version Control Everything

policies/
├── kubernetes/
│   ├── security.rego
│   ├── resources.rego
│   └── networking.rego
├── terraform/
│   └── aws.rego
├── docker/
│   └── security.rego
└── tests/
    ├── kubernetes_test.rego
    └── terraform_test.rego

2. Test Policies Thoroughly

# Every policy should have tests
test_policy_allows_valid_input { ... }
test_policy_denies_invalid_input { ... }
test_policy_edge_cases { ... }

3. Use Descriptive Messages

# ❌ Bad
deny[msg] {
  condition
  msg := "Violation"
}

# ✅ Good
deny[msg] {
  condition
  msg := sprintf("Container %v violates security policy: %v", [container.name, reason])
}

4. Monitor and Alert

# Set up monitoring for:
- Policy violations per namespace
- Failed policy evaluations
- Policy enforcement latency
- Exemption usage

5. Documentation

# policies/kubernetes/security.rego
# METADATA
# title: Kubernetes Security Policies
# description: Enforces security best practices for Kubernetes workloads
# authors:
#   - Security Team <[email protected]>
# related_resources:
#   - https://kubernetes.io/docs/concepts/security/pod-security-standards/

package kubernetes.security

Key Takeaways

✅ Automate policy enforcement - Eliminate manual reviews ✅ Test your policies - Like any code, policies need tests ✅ Version control policies - Track changes, enable rollback ✅ Start simple, iterate - Begin with basic policies, add complexity ✅ Provide clear error messages - Help developers fix violations ✅ Monitor compliance - Track violations and trends ✅ Use exemptions sparingly - Require approval and justification

What's Next

With automated policy enforcement in place, the next step is automating security responses. In the next article, we'll cover Security Automation and Orchestration—automatically triaging vulnerabilities, creating remediation PRs, and building self-healing security systems.

---

Next Article: Security Automation and Orchestration →

---

Part of the DevSecOps 101 Series