Analyzing Kubernetes Audit Logs
When to Use
- When investigating security incidents that require analyzing kubernetes audit logs
- When building detection rules or threat hunting queries for this domain
- When SOC analysts need structured procedures for this analysis type
- When validating security monitoring coverage for related attack techniques
Detection Gaps & Validation
level: Metadatahides the payload: at Metadata level therequestObjectis dropped, so privileged-pod detection (securityContext.privileged,hostPID, capability adds) and secret values are invisible. Uselevel: RequestResponseforpodscreate/update and forclusterrolebindingsto see the escalation body.- Subresources are separate:
pods/execandpods/attachare distinct frompods- a rule matching onlyresource == "pods"misses every shell-in. Match the subresource explicitly. omitStages: ["RequestReceived"]combined with a low default level can silently discard events; confirm your policy emits theResponseCompletestage for sensitive verbs.- Anonymous access: match
user.username == "system:anonymous"or groupsystem:unauthenticated, not just missing users. - Secret enumeration at Metadata shows the verb/name but not which keys were read - escalate to Request level for the secrets you care about.
- How to validate the rule fires: run
kubectl exec -it <pod> -- shandkubectl get secret <name> -o yamlagainst a test namespace, then grep the audit log forpods/execandsecretsevents with your username. If nothing appears, the audit policy level is too low or the stage is omitted.
Prerequisites
- Familiarity with container security concepts and tools
- Access to a test or lab environment for safe execution
- Python 3.8+ with required dependencies installed
- Appropriate authorization for any testing activities
Instructions
Parse Kubernetes audit log files (JSON lines format) to detect security-relevant events including unauthorized access, privilege escalation, and data exfiltration.
import json
with open("/var/log/kubernetes/audit.log") as f:
for line in f:
event = json.loads(line)
verb = event.get("verb")
resource = event.get("objectRef", {}).get("resource")
user = event.get("user", {}).get("username")
if verb == "create" and resource == "pods/exec":
print(f"Pod exec by {user}")
Key events to detect:
- pods/exec and pods/attach (shell into containers)
- secrets access (get/list/watch)
- clusterrolebindings creation (RBAC escalation)
- Privileged pod creation
- Anonymous or system:unauthenticated access
Examples
# Detect secret enumeration
if verb in ("get", "list") and resource == "secrets":
print(f"Secret access: {user} -> {event['objectRef'].get('name')}")