Open Policy Agent (OPA) Security Testing: Rego Policy Bypass, Kubernetes Admission Webhook, and Bundle Tampering

Gatekeeper Constraint Template Testing

# Gatekeeper uses ConstraintTemplates + Constraints to define policies
# Common issues: dryrun enforcement mode, missing parameters, incomplete coverage

# Check enforcement actions (audit vs dryrun vs deny)
kubectl get constraints --all-namespaces -o json | \
  python3 -c "
import json,sys
cs = json.load(sys.stdin)['items']
for c in cs:
    meta = c['metadata']
    spec = c.get('spec', {})
    enforcement = spec.get('enforcementAction', 'deny')
    kinds = spec.get('match', {}).get('kinds', [])
    print(f\"{meta['name']}: enforcementAction={enforcement}\")
    if enforcement in ['audit', 'dryrun']:
        print(f\"  [!] {enforcement}: violations are LOGGED but not BLOCKED\")
    print(f\"  Applies to: {kinds}\")
"

# Check Gatekeeper audit results — violations in audit mode mean real policy breaches
kubectl get constraint -o json | \
  python3 -c "
import json,sys
cs = json.load(sys.stdin)['items']
for c in cs:
    violations = c.get('status', {}).get('violations', [])
    if violations:
        print(f\"{c['metadata']['name']}: {len(violations)} violations\")
        for v in violations[:3]:
            print(f\"  Namespace: {v.get('namespace','?')} Name: {v.get('name','?')}\")
            print(f\"  Message: {v.get('message','?')}\")
"

# Test for missing constraint coverage
# Example: if there's a template for container privileges but no matching constraint,
# the template is never enforced

# Get all ConstraintTemplates
templates=$(kubectl get constrainttemplates -o jsonpath='{.items[*].metadata.name}')

# For each template, check if a Constraint exists
for tmpl in $templates; do
    kind=$(kubectl get constrainttemplate "$tmpl" -o jsonpath='{.spec.crd.spec.names.kind}')
    count=$(kubectl get "$kind" 2>/dev/null | wc -l)
    [ "$count" -le 1 ] && echo "Template $tmpl has no deployed constraint → not enforced"
done

OPA Bundle Security and Signing

# OPA bundles carry policy and data — tampering with bundle source = policy bypass
# Bundles should be signed; signature verification must be enforced

# Check if OPA is configured to verify bundle signatures
kubectl get configmap opa-config -n opa-system -o yaml | grep -A 10 "bundle\|signing"
# Look for: keys, scope, signing configuration

# If bundle signing is NOT enabled, test by serving a modified bundle from a rogue server
# This requires MITM or DNS poisoning to bundle_url, but tests the signing requirement

# Check OPA startup flags for signature verification
kubectl get deploy opa -n opa-system -o json | \
  python3 -c "
import json,sys
deploy = json.load(sys.stdin)
for c in deploy['spec']['template']['spec']['containers']:
    if 'opa' in c['name']:
        args = c.get('args', [])
        print('OPA args:', args)
        if any('bundle-signing-key' in str(a) for a in args):
            print('  [OK] Bundle signing key configured')
        else:
            print('  [!] No bundle signing key — bundles not verified')
"

# Check OPA bundle configuration
# Secure bundle config with signature:
# services:
#   bundle-service:
#     url: https://bundle.example.com
# bundles:
#   main:
#     resource: /bundles/policy.tar.gz
#     signing:
#       keyid: my-signing-key
#       scope: write
# keys:
#   my-signing-key:
#     algorithm: RS256
#     key: |
#       -----BEGIN PUBLIC KEY-----
#       ...

# Test OPA management API exposure
kubectl port-forward svc/opa -n opa-system 8181:8181 &
# Check if management API is accessible without authentication
curl -s http://localhost:8181/v1/policies | head -20
# Returns policies → OPA management API is unauthenticated

OPA API Authorization Bypass

# OPA is widely used for API authorization — policy gaps allow access to protected endpoints

# Test OPA policy with unexpected input shapes
# If policy expects input.user.roles but JWT claim is input.user.role (singular):
curl -s -X POST http://opa-endpoint:8181/v1/data/authz/allow \
  -H "Content-Type: application/json" \
  -d '{
    "input": {
      "user": {
        "role": "admin"    # singular, policy checks .roles (plural)
      },
      "path": "/api/admin",
      "method": "GET"
    }
  }' | python3 -c "import json,sys; print(json.load(sys.stdin))"
# If result: {"result": false} — correct denial
# If result: {} or undefined — policy has field mismatch that evaluates to undefined = allow

# Test for wildcard/glob matching bypass in path policies
# Policy using glob.match("/api/admin*", [], input.path)
# may match /api/adminextra or /api/admin%2fetc

# Test method normalization
curl -s -X POST http://opa-endpoint:8181/v1/data/authz/allow \
  -d '{"input": {"method": "get", "path": "/api/admin"}}' | \
  python3 -c "import json,sys; d=json.load(sys.stdin); print('Allow:', d.get('result'))"
# If policy only checks upper(input.method) but receives lowercase → bypass

# Enumerate OPA policy endpoints
curl -s http://opa-endpoint:8181/v1/policies | \
  python3 -c "import json,sys; [print(p['id']) for p in json.load(sys.stdin).get('result',[])]"

OPA Data API Exposure

# OPA's data API stores context used by policies — users, groups, allowed paths
# An exposed data API lets attackers read authorization state

# Check if OPA data API is externally accessible
kubectl get svc opa -n opa-system -o json | \
  python3 -c "
import json,sys
svc = json.load(sys.stdin)
print('Type:', svc['spec']['type'])
ports = svc['spec']['ports']
for p in ports:
    print(f\"Port: {p.get('port')} → {p.get('targetPort')} ({p.get('protocol','TCP')})\")
# ClusterIP = only internal (OK)
# LoadBalancer or NodePort = exposed externally
"

# Test data API enumeration
curl -s http://opa-endpoint:8181/v1/data | \
  python3 -c "
import json,sys
d = json.load(sys.stdin).get('result', {})
def print_keys(obj, indent=0):
    if isinstance(obj, dict):
        for k in obj:
            print(' '*indent + str(k))
            print_keys(obj[k], indent+2)
print_keys(d)
"
# If policy data includes user → role mappings, allowed endpoints, etc.
# this is sensitive information an attacker can use to understand the policy

# OPA decision logging — check if logs go somewhere secure
kubectl get deploy opa -n opa-system -o json | \
  python3 -c "
import json,sys
d = json.load(sys.stdin)
for c in d['spec']['template']['spec']['containers']:
    if 'opa' in c['name']:
        print('Args:', c.get('args', []))
        # Look for: --set=decision_logs.console=true (logs to stdout only)
        # vs --set=decision_logs.service=... (ships logs to storage)
"

OPA Hardening