Build an AI Deployment Validator with Claude API and OPA

pip install anthropic opa-python-client pyyaml
# Or install OPA binary directly:
curl -L -o opa https://openpolicyagent.org/downloads/latest/opa_linux_amd64_static
chmod +x opa && mv opa /usr/local/bin/

# policies/k8s_security.rego
package k8s.security
 
import future.keywords.if
import future.keywords.in
 
# Deny privileged containers
deny[msg] if {
    container := input.spec.template.spec.containers[_]
    container.securityContext.privileged == true
    msg := sprintf("Container '%v' runs as privileged — this grants root-level host access", [container.name])
}
 
# Deny containers without resource limits
deny[msg] if {
    container := input.spec.template.spec.containers[_]
    not container.resources.limits
    msg := sprintf("Container '%v' has no resource limits — it can consume unbounded CPU/memory and cause node OOM", [container.name])
}
 
# Deny latest tag
deny[msg] if {
    container := input.spec.template.spec.containers[_]
    endswith(container.image, ":latest")
    msg := sprintf("Container '%v' uses :latest tag — prevents reproducible deployments and rollbacks", [container.name])
}
 
# Deny running as root
deny[msg] if {
    container := input.spec.template.spec.containers[_]
    container.securityContext.runAsUser == 0
    msg := sprintf("Container '%v' runs as root (UID 0) — violates least-privilege principle", [container.name])
}
 
# Warn on missing readiness probe
warn[msg] if {
    container := input.spec.template.spec.containers[_]
    not container.readinessProbe
    msg := sprintf("Container '%v' has no readiness probe — traffic may reach pods before they are ready", [container.name])
}
 
# Deny hostNetwork
deny[msg] if {
    input.spec.template.spec.hostNetwork == true
    msg := "hostNetwork: true exposes host network namespace to the pod — severe security risk"
}

import anthropic
import json
import subprocess
import sys
import yaml
from pathlib import Path
 
 
OPA_POLICY_DIR = "policies/"
FAIL_ON_WARN = False
 
 
def run_opa_eval(manifest: dict, policy_file: str) -> dict:
    """Run OPA evaluation against a manifest."""
    manifest_json = json.dumps(manifest)
 
    result = subprocess.run(
        [
            "opa", "eval",
            "--data", policy_file,
            "--input", "/dev/stdin",
            "--format", "json",
            "data.k8s.security"
        ],
        input=manifest_json,
        capture_output=True,
        text=True,
        timeout=30
    )
 
    if result.returncode != 0:
        return {"error": result.stderr, "deny": [], "warn": []}
 
    output = json.loads(result.stdout)
    results = output.get("result", [{}])[0].get("expressions", [{}])[0].get("value", {})
 
    return {
        "deny": results.get("deny", []),
        "warn": results.get("warn", [])
    }
 
 
def analyze_with_claude(manifest: dict, opa_results: dict) -> dict:
    """Get Claude's security and configuration analysis."""
    client = anthropic.Anthropic()
 
    manifest_yaml = yaml.dump(manifest, default_flow_style=False)
    opa_violations = json.dumps(opa_results, indent=2)
 
    prompt = f"""You are a Kubernetes security expert reviewing a deployment manifest.
 
## Manifest
```yaml
{manifest_yaml}
```
 
## OPA Policy Violations Found
{opa_violations}
 
Analyze this manifest and identify:
1. Security issues NOT caught by OPA (beyond the violations listed above)
2. Reliability risks (single replica, no disruption budget, etc.)
3. Operational best practices violations (missing labels, annotations, etc.)
4. The 3 most critical issues overall (OPA + your analysis combined)
5. Specific code fixes for each critical issue
 
Return JSON:
{{
  "additional_security_issues": ["..."],
  "reliability_risks": ["..."],
  "best_practice_violations": ["..."],
  "critical_issues": [
    {{
      "issue": "description",
      "severity": "critical|high|medium",
      "fix": "exact YAML or command to fix this"
    }}
  ],
  "overall_risk": "high|medium|low",
  "recommendation": "approve|approve_with_changes|reject"
}}"""
 
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=2000,
        messages=[{"role": "user", "content": prompt}]
    )
 
    text = response.content[0].text.strip()
    if "```" in text:
        text = text.split("```")[1]
        if text.startswith("json"):
            text = text[4:]
 
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return {"error": "Could not parse Claude response", "raw": text}
 
 
def validate_manifest(manifest_path: str) -> bool:
    """Main validation function. Returns True if manifest passes, False if it fails."""
    with open(manifest_path) as f:
        manifest = yaml.safe_load(f)
 
    resource_name = f"{manifest.get('kind', 'Unknown')}/{manifest.get('metadata', {}).get('name', 'unknown')}"
    print(f"\n{'='*60}")
    print(f"Validating: {resource_name}")
    print("="*60)
 
    # Step 1: OPA evaluation
    print("\n[1/2] Running OPA policy checks...")
    opa_results = {"deny": [], "warn": []}
    for policy_file in Path(OPA_POLICY_DIR).glob("*.rego"):
        result = run_opa_eval(manifest, str(policy_file))
        opa_results["deny"].extend(result.get("deny", []))
        opa_results["warn"].extend(result.get("warn", []))
 
    # Step 2: Claude analysis
    print("[2/2] Running AI security analysis...")
    claude_analysis = analyze_with_claude(manifest, opa_results)
 
    # Display results
    passed = True
 
    if opa_results["deny"]:
        print(f"\n POLICY VIOLATIONS ({len(opa_results['deny'])} found):")
        for violation in opa_results["deny"]:
            print(f"  ✗ {violation}")
        passed = False
 
    if opa_results["warn"]:
        print(f"\n WARNINGS ({len(opa_results['warn'])} found):")
        for warning in opa_results["warn"]:
            print(f"  ⚠ {warning}")
        if FAIL_ON_WARN:
            passed = False
 
    critical = claude_analysis.get("critical_issues", [])
    if critical:
        print(f"\n AI ANALYSIS — CRITICAL ISSUES ({len(critical)} found):")
        for issue in critical:
            severity = issue.get("severity", "unknown").upper()
            print(f"\n  [{severity}] {issue.get('issue', '')}")
            fix = issue.get("fix", "")
            if fix:
                print(f"  Fix:\n{chr(10).join('    ' + line for line in fix.split(chr(10)))}")
 
        # Fail if AI finds critical issues
        critical_count = sum(1 for i in critical if i.get("severity") == "critical")
        if critical_count > 0:
            passed = False
 
    recommendation = claude_analysis.get("recommendation", "unknown")
    risk = claude_analysis.get("overall_risk", "unknown")
    print(f"\n Risk Level: {risk.upper()}")
    print(f" Recommendation: {recommendation.upper().replace('_', ' ')}")
 
    if passed:
        print("\n VALIDATION PASSED — Safe to deploy")
    else:
        print("\n VALIDATION FAILED — Fix issues before deploying")
 
    return passed
 
 
if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python validator.py <manifest.yaml> [manifest2.yaml ...]")
        sys.exit(1)
 
    all_passed = True
    for manifest_file in sys.argv[1:]:
        passed = validate_manifest(manifest_file)
        if not passed:
            all_passed = False
 
    sys.exit(0 if all_passed else 1)

name: Validate Kubernetes Manifests
 
on:
  pull_request:
    paths:
      - 'k8s/**/*.yaml'
      - 'k8s/**/*.yml'
 
jobs:
  validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
 
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'
 
      - name: Install dependencies
        run: pip install anthropic pyyaml
 
      - name: Install OPA
        run: |
          curl -L -o opa https://openpolicyagent.org/downloads/latest/opa_linux_amd64_static
          chmod +x opa && sudo mv opa /usr/local/bin/
 
      - name: Validate changed manifests
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        run: |
          # Get list of changed YAML files
          CHANGED=$(git diff --name-only origin/main...HEAD | grep -E '\.ya?ml$' | grep 'k8s/')
          if [ -z "$CHANGED" ]; then
            echo "No Kubernetes manifests changed"
            exit 0
          fi
          python validator.py $CHANGED