Build AI-Powered Kubernetes Policy Enforcer with OPA and Claude

# Install Gatekeeper
kubectl apply -f https://raw.githubusercontent.com/open-policy-agent/gatekeeper/master/deploy/gatekeeper.yaml
 
# Verify
kubectl get pods -n gatekeeper-system

# constraint-template-require-limits.yaml
apiVersion: templates.gatekeeper.sh/v1
kind: ConstraintTemplate
metadata:
  name: requireresourcelimits
spec:
  crd:
    spec:
      names:
        kind: RequireResourceLimits
  targets:
  - target: admission.k8s.gatekeeper.sh
    rego: |
      package requireresourcelimits
 
      violation[{"msg": msg}] {
        container := input.review.object.spec.containers[_]
        not container.resources.limits.cpu
        msg := sprintf("Container '%v' in pod '%v' is missing CPU limits. Set resources.limits.cpu",
          [container.name, input.review.object.metadata.name])
      }
 
      violation[{"msg": msg}] {
        container := input.review.object.spec.containers[_]
        not container.resources.limits.memory
        msg := sprintf("Container '%v' in pod '%v' is missing memory limits. Set resources.limits.memory",
          [container.name, input.review.object.metadata.name])
      }
---
apiVersion: constraints.gatekeeper.sh/v1beta1
kind: RequireResourceLimits
metadata:
  name: must-have-resource-limits
spec:
  match:
    kinds:
    - apiGroups: [""]
      kinds: ["Pod"]
    namespaces: ["production", "staging"]

# policy_explainer.py
import anthropic
import json
from flask import Flask, request, jsonify
 
app = Flask(__name__)
client = anthropic.Anthropic()
 
SYSTEM_PROMPT = """You are a Kubernetes policy enforcement expert. When given a policy violation,
explain it clearly for a developer in 2-3 sentences. Then provide the exact YAML fix.
Format your response as:
EXPLANATION: <plain English explanation>
FIX:
<yaml code block with the fix>
Keep the fix minimal — only change what's needed."""
 
 
def explain_violation(resource_yaml: str, violation_message: str) -> dict:
    yaml_block = "```yaml\n" + resource_yaml + "\n```"
    user_content = (
        f"Policy violation occurred:\n\nViolation message: {violation_message}\n\n"
        "Resource that triggered it:\n" + yaml_block + "\n\nExplain what's wrong and show the exact fix."
    )
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": user_content}],
    )
 
    text = response.content[0].text
    parts = text.split("FIX:", 1)
 
    return {
        "explanation": parts[0].replace("EXPLANATION:", "").strip(),
        "fix": parts[1].strip() if len(parts) > 1 else "",
    }
 
 
@app.route("/explain", methods=["POST"])
def explain():
    data = request.json
    result = explain_violation(
        resource_yaml=data.get("resource", ""),
        violation_message=data.get("violation", ""),
    )
    return jsonify(result)
 
 
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

# rego_generator.py
import anthropic
 
client = anthropic.Anthropic()
 
REGO_SYSTEM = """You are a Rego policy expert for OPA Gatekeeper.
Convert plain English policy descriptions into valid Gatekeeper ConstraintTemplate + Constraint YAML.
Always include proper violation messages that help developers understand what to fix.
Output only valid YAML, no explanation."""
 
 
def generate_policy(description: str, constraint_name: str) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        system=REGO_SYSTEM,
        messages=[
            {
                "role": "user",
                "content": f"""Create a Gatekeeper ConstraintTemplate and Constraint for this policy:
 
Policy: {description}
Constraint name: {constraint_name}
 
Apply to: pods in production namespace""",
            }
        ],
    )
    return response.content[0].text
 
 
# Example usage
if __name__ == "__main__":
    policy = generate_policy(
        description="All pods must have a 'team' label with a non-empty value",
        constraint_name="require-team-label",
    )
    print(policy)

# policy-explainer-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ai-policy-explainer
  namespace: gatekeeper-system
spec:
  replicas: 2
  selector:
    matchLabels:
      app: ai-policy-explainer
  template:
    metadata:
      labels:
        app: ai-policy-explainer
    spec:
      containers:
      - name: explainer
        image: your-registry/ai-policy-explainer:latest
        ports:
        - containerPort: 8080
        env:
        - name: ANTHROPIC_API_KEY
          valueFrom:
            secretKeyRef:
              name: anthropic-secret
              key: api-key
        resources:
          requests:
            cpu: 100m
            memory: 128Mi
          limits:
            cpu: 500m
            memory: 512Mi
---
apiVersion: v1
kind: Secret
metadata:
  name: anthropic-secret
  namespace: gatekeeper-system
type: Opaque
stringData:
  api-key: "your-anthropic-api-key"

# kubectl-policy-explain plugin
#!/usr/bin/env python3
# Save as /usr/local/bin/kubectl-policy_explain and chmod +x
 
import sys
import subprocess
import requests
import json
 
def get_last_admission_error():
    """Get the last admission webhook error from events"""
    result = subprocess.run(
        ["kubectl", "get", "events", "--field-selector", "reason=FailedCreate",
         "-o", "json", "--all-namespaces"],
        capture_output=True, text=True
    )
    events = json.loads(result.stdout)
    if events["items"]:
        return events["items"][-1]["message"]
    return None
 
if __name__ == "__main__":
    resource_file = sys.argv[1] if len(sys.argv) > 1 else None
    if not resource_file:
        print("Usage: kubectl policy-explain <resource.yaml>")
        sys.exit(1)
 
    with open(resource_file) as f:
        resource_yaml = f.read()
 
    # Try applying and catch the error
    result = subprocess.run(
        ["kubectl", "apply", "--dry-run=server", "-f", resource_file],
        capture_output=True, text=True
    )
 
    if result.returncode != 0:
        violation = result.stderr
        resp = requests.post(
            "http://ai-policy-explainer.gatekeeper-system.svc/explain",
            json={"resource": resource_yaml, "violation": violation}
        )
        data = resp.json()
        print(f"\n🚫 Policy Violation\n")
        print(f"Explanation: {data['explanation']}\n")
        print(f"Fix:\n{data['fix']}")

# This should fail policy and return AI explanation
cat <<EOF | kubectl apply -f -
apiVersion: v1
kind: Pod
metadata:
  name: test-no-limits
  namespace: production
spec:
  containers:
  - name: nginx
    image: nginx:latest
EOF
 
# With kubectl plugin:
kubectl policy-explain my-pod.yaml
# Output:
# 🚫 Policy Violation
# Explanation: Your container 'nginx' doesn't have CPU or memory limits set.
# Without limits, this pod can consume all available node resources and crash other pods.
# Fix:
# resources:
#   limits:
#     cpu: "500m"
#     memory: "512Mi"