AI Agents for Zero-Trust Policy Generation: Where This Is Heading in 2026

import anthropic
import json
import boto3
 
client = anthropic.Anthropic()
 
 
def get_actual_api_calls(role_arn: str, days: int = 14) -> list[dict]:
    """Pull every API call actually made by this role over the observation window."""
    cloudtrail = boto3.client("cloudtrail")
    events = cloudtrail.lookup_events(
        LookupAttributes=[{"AttributeKey": "Username", "AttributeValue": role_arn}],
    )
    return [{"action": e["EventName"], "resource": e.get("Resources", [])} for e in events["Events"]]
 
 
GENERATE_IAM_PROMPT = """Generate a least-privilege IAM policy based on the
actual API calls this role made over a {days}-day observation window.
 
Observed API calls (action + resource):
{observed_calls}
 
Current policy (overly broad, needs tightening):
{current_policy}
 
Generate a new policy that grants ONLY the actions and resources actually
observed, with resource ARNs scoped as tightly as the observed calls allow
(not wildcarded unless the observed calls genuinely touched multiple resources
matching a pattern). Flag any observed call that seems unusual or worth
a human double-checking before removing broader access."""
 
 
def generate_least_privilege_policy(observed_calls: list[dict], current_policy: dict, days: int) -> dict:
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=2000,
        messages=[{
            "role": "user",
            "content": GENERATE_IAM_PROMPT.format(
                days=days, observed_calls=observed_calls, current_policy=json.dumps(current_policy, indent=2)
            )
        }]
    )
    return response.content[0].text

def get_observed_pod_traffic(namespace: str, days: int = 14) -> list[dict]:
    """Pull actual observed connections from a service mesh's telemetry
    (Cilium Hubble, Istio, Linkerd all expose this)."""
    import subprocess
    result = subprocess.run(
        ["hubble", "observe", "--namespace", namespace, "--since", f"{days*24}h", "-o", "json"],
        capture_output=True, text=True
    )
    return [json.loads(line) for line in result.stdout.splitlines() if line]
 
 
def generate_network_policy(pod_label: str, observed_traffic: list[dict]) -> str:
    destinations = list(set(
        f"{flow['destination']['namespace']}/{flow['destination']['pod_labels']}"
        for flow in observed_traffic if flow.get("verdict") == "FORWARDED"
    ))
 
    prompt = f"""Generate a Kubernetes NetworkPolicy for pods labeled "{pod_label}"
based on these observed egress destinations over 14 days:
{destinations}
 
Default-deny everything else. Include comments explaining what each rule
covers based on the observed traffic."""
 
    response = client.messages.create(
        model="claude-sonnet-5", max_tokens=1000,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.content[0].text

# Generated from 14 days of observed traffic for payments-api
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: payments-api-observed-policy
spec:
  podSelector:
    matchLabels:
      app: payments-api
  policyTypes: ["Egress"]
  egress:
    # Observed: 4,200 connections/day to postgres-primary
    - to:
        - podSelector:
            matchLabels:
              app: postgres-primary
      ports:
        - port: 5432
    # Observed: 180 connections/day to notification-service (order confirmations)
    - to:
        - podSelector:
            matchLabels:
              app: notification-service
      ports:
        - port: 8080
    # No other egress observed in 14-day window — everything else denied

# Cilium NetworkPolicy dry-run mode — logs violations without blocking,
# the critical safety net between generation and enforcement
apiVersion: cilium.io/v2
kind: CiliumNetworkPolicy
metadata:
  name: payments-api-dryrun
  annotations:
    policy.cilium.io/audit-mode: "true"    # Log-only, catches gaps before hard enforcement