# analyzing-kubernetes-audit-logs

Parses Kubernetes API server audit logs (JSON lines) to detect exec-into-pod, secret access, RBAC modifications, privileged pod creation, and anonymous API access, and builds SIEM detection rules from the event patterns. Use when investigating a suspected cluster compromise, reconstructing what an attacker did through the API server, or writing Kubernetes-specific detection content. Keywords: audit policy, audit log, kube-apiserver, exec into pod, RBAC change, anonymous access, detection rules. Do not use for syscall-level detection inside a running container - use detecting-container-runtime-threats-with-falco.
'

- **Kind:** skill
- **Source:** https://github.com/mukul975/Anthropic-Cybersecurity-Skills
- **Page:** https://forefy.com/skills/70ca9707-9fe5-46bc-8206-34b9d4aaacbe
- **API (JSON + files):** https://forefy.com/api/asr/70ca9707-9fe5-46bc-8206-34b9d4aaacbe

---

## LICENSE

```

```

## SKILL.md

---
name: analyzing-kubernetes-audit-logs
description: >-
  Parses Kubernetes API server audit logs (JSON lines) to detect exec-into-pod, secret access,
  RBAC modifications, privileged pod creation, and anonymous API access, and builds SIEM
  detection rules from the event patterns. Use when investigating a suspected cluster
  compromise, reconstructing what an attacker did through the API server, or writing
  Kubernetes-specific detection content. Keywords: audit policy, audit log, kube-apiserver,
  exec into pod, RBAC change, anonymous access, detection rules. Do not use for syscall-level
  detection inside a running container - use detecting-container-runtime-threats-with-falco.

  '
domain: cybersecurity
subdomain: container-security
tags:
- kubernetes-security
- container-security
- audit-log-analysis
- rbac
- privilege-escalation
- k8s-api-server
- threat-detection
version: '1.0'
author: mahipal
license: Apache-2.0
nist_csf:
- PR.PS-01
- PR.IR-01
- ID.AM-08
- DE.CM-01
mitre_attack:
- T1610
- T1613
- T1078
- T1552.007
---

# 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

## 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.

```python
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:
1. pods/exec and pods/attach (shell into containers)
2. secrets access (get/list/watch)
3. clusterrolebindings creation (RBAC escalation)
4. Privileged pod creation
5. Anonymous or system:unauthenticated access

## Examples

```python
# Detect secret enumeration
if verb in ("get", "list") and resource == "secrets":
    print(f"Secret access: {user} -> {event['objectRef'].get('name')}")
```

## references

```

```

## references/api-reference.md

# API Reference: Analyzing Kubernetes Audit Logs

## Audit Log Format (JSON Lines)

```json
{
  "kind": "Event",
  "apiVersion": "audit.k8s.io/v1",
  "level": "RequestResponse",
  "verb": "create",
  "user": {"username": "admin", "groups": ["system:masters"]},
  "sourceIPs": ["10.0.0.5"],
  "objectRef": {
    "resource": "pods",
    "subresource": "exec",
    "namespace": "default",
    "name": "web-pod"
  },
  "responseStatus": {"code": 200},
  "requestReceivedTimestamp": "2025-03-15T14:00:00Z"
}
```

## Security-Critical Audit Events

| Event | objectRef | Severity |
|-------|-----------|----------|
| Pod exec | `resource: pods, subresource: exec` | HIGH |
| Secret access | `resource: secrets, verb: get/list` | HIGH |
| RBAC change | `resource: clusterrolebindings` | CRITICAL |
| Privileged pod | `requestObject.spec.containers[].securityContext.privileged` | CRITICAL |
| Anonymous access | `user.username: system:anonymous` | CRITICAL |

## Audit Policy Levels

| Level | Captures |
|-------|----------|
| None | No logging |
| Metadata | Timestamp, user, verb, resource |
| Request | Metadata + request body |
| RequestResponse | Request + response body |

## Python Parsing

```python
import json
with open("audit.log") as f:
    for line in f:
        event = json.loads(line)
        print(event["verb"], event["objectRef"]["resource"])
```

### References

- K8s Auditing: https://kubernetes.io/docs/tasks/debug/debug-cluster/audit/
- Audit policy: https://kubernetes.io/docs/reference/config-api/apiserver-audit.v1/
- Datadog k8s audit: https://www.datadoghq.com/blog/monitor-kubernetes-audit-logs/

## scripts

```

```

## scripts/agent.py

```python
#!/usr/bin/env python3
"""Agent for analyzing Kubernetes audit logs for security threats."""

import json
import argparse
from collections import defaultdict
from datetime import datetime


def parse_audit_log(log_path):
    """Parse Kubernetes audit log file (JSON lines format)."""
    events = []
    with open(log_path) as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                events.append(json.loads(line))
            except json.JSONDecodeError:
                continue
    return events


def detect_pod_exec(events):
    """Detect kubectl exec and attach events (shell access to pods)."""
    findings = []
    for event in events:
        obj_ref = event.get("objectRef", {})
        subresource = obj_ref.get("subresource", "")
        if subresource in ("exec", "attach"):
            findings.append({
                "timestamp": event.get("requestReceivedTimestamp", ""),
                "user": event.get("user", {}).get("username", ""),
                "groups": event.get("user", {}).get("groups", []),
                "verb": event.get("verb", ""),
                "namespace": obj_ref.get("namespace", ""),
                "pod": obj_ref.get("name", ""),
                "subresource": subresource,
                "source_ip": event.get("sourceIPs", [""])[0],
                "severity": "HIGH",
            })
    return findings


def detect_secret_access(events):
    """Detect access to Kubernetes secrets."""
    findings = []
    for event in events:
        obj_ref = event.get("objectRef", {})
        if obj_ref.get("resource") != "secrets":
            continue
        verb = event.get("verb", "")
        if verb not in ("get", "list", "watch", "create", "update", "delete"):
            continue
        findings.append({
            "timestamp": event.get("requestReceivedTimestamp", ""),
            "user": event.get("user", {}).get("username", ""),
            "verb": verb,
            "namespace": obj_ref.get("namespace", ""),
            "secret_name": obj_ref.get("name", ""),
            "source_ip": event.get("sourceIPs", [""])[0],
            "severity": "HIGH" if verb in ("list", "delete") else "MEDIUM",
        })
    return findings


def detect_rbac_changes(events):
    """Detect RBAC role and binding modifications."""
    rbac_resources = {"clusterroles", "clusterrolebindings", "roles", "rolebindings"}
    findings = []
    for event in events:
        obj_ref = event.get("objectRef", {})
        resource = obj_ref.get("resource", "")
        verb = event.get("verb", "")
        if resource in rbac_resources and verb in ("create", "update", "patch", "delete"):
            findings.append({
                "timestamp": event.get("requestReceivedTimestamp", ""),
                "user": event.get("user", {}).get("username", ""),
                "verb": verb,
                "resource": resource,
                "name": obj_ref.get("name", ""),
                "namespace": obj_ref.get("namespace", ""),
                "source_ip": event.get("sourceIPs", [""])[0],
                "severity": "CRITICAL" if "cluster" in resource else "HIGH",
            })
    return findings


def detect_privileged_pods(events):
    """Detect creation of privileged pods."""
    findings = []
    for event in events:
        if event.get("verb") != "create":
            continue
        obj_ref = event.get("objectRef", {})
        if obj_ref.get("resource") != "pods":
            continue
        request_obj = event.get("requestObject", {})
        spec = request_obj.get("spec", {})
        containers = spec.get("containers", [])
        for container in containers:
            sc = container.get("securityContext", {})
            if sc.get("privileged"):
                findings.append({
                    "timestamp": event.get("requestReceivedTimestamp", ""),
                    "user": event.get("user", {}).get("username", ""),
                    "namespace": obj_ref.get("namespace", ""),
                    "pod": obj_ref.get("name", ""),
                    "container": container.get("name", ""),
                    "severity": "CRITICAL",
                })
    return findings


def detect_anonymous_access(events):
    """Detect API access by anonymous or unauthenticated users."""
    findings = []
    anon_users = {"system:anonymous", "system:unauthenticated"}
    for event in events:
        user = event.get("user", {}).get("username", "")
        groups = event.get("user", {}).get("groups", [])
        if user in anon_users or "system:unauthenticated" in groups:
            status_code = event.get("responseStatus", {}).get("code", 0)
            if status_code < 400:
                findings.append({
                    "timestamp": event.get("requestReceivedTimestamp", ""),
                    "user": user,
                    "verb": event.get("verb", ""),
                    "resource": event.get("objectRef", {}).get("resource", ""),
                    "source_ip": event.get("sourceIPs", [""])[0],
                    "status_code": status_code,
                    "severity": "CRITICAL",
                })
    return findings


def detect_forbidden_surge(events, threshold=20):
    """Detect 403 surges indicating enumeration or brute force."""
    user_forbidden = defaultdict(int)
    for event in events:
        if event.get("responseStatus", {}).get("code") == 403:
            user = event.get("user", {}).get("username", "")
            user_forbidden[user] += 1
    surges = []
    for user, count in user_forbidden.items():
        if count >= threshold:
            surges.append({"user": user, "forbidden_count": count, "severity": "MEDIUM"})
    return sorted(surges, key=lambda x: x["forbidden_count"], reverse=True)


def main():
    parser = argparse.ArgumentParser(description="Kubernetes Audit Log Analyzer")
    parser.add_argument("--audit-log", required=True, help="Path to audit log file")
    parser.add_argument("--output", default="k8s_audit_report.json")
    parser.add_argument("--action", choices=[
        "exec", "secrets", "rbac", "privileged", "anonymous", "full_analysis"
    ], default="full_analysis")
    args = parser.parse_args()

    events = parse_audit_log(args.audit_log)
    report = {"log_file": args.audit_log, "total_events": len(events),
              "generated_at": datetime.utcnow().isoformat(), "findings": {}}
    print(f"[+] Parsed {len(events)} audit events")

    if args.action in ("exec", "full_analysis"):
        findings = detect_pod_exec(events)
        report["findings"]["pod_exec"] = findings
        print(f"[+] Pod exec/attach events: {len(findings)}")

    if args.action in ("secrets", "full_analysis"):
        findings = detect_secret_access(events)
        report["findings"]["secret_access"] = findings
        print(f"[+] Secret access events: {len(findings)}")

    if args.action in ("rbac", "full_analysis"):
        findings = detect_rbac_changes(events)
        report["findings"]["rbac_changes"] = findings
        print(f"[+] RBAC changes: {len(findings)}")

    if args.action in ("privileged", "full_analysis"):
        findings = detect_privileged_pods(events)
        report["findings"]["privileged_pods"] = findings
        print(f"[+] Privileged pod creation: {len(findings)}")

    if args.action in ("anonymous", "full_analysis"):
        findings = detect_anonymous_access(events)
        report["findings"]["anonymous_access"] = findings
        print(f"[+] Anonymous access events: {len(findings)}")

    forbidden = detect_forbidden_surge(events)
    report["findings"]["forbidden_surges"] = forbidden
    print(f"[+] 403 surges: {len(forbidden)}")

    with open(args.output, "w") as f:
        json.dump(report, f, indent=2, default=str)
    print(f"[+] Report saved to {args.output}")


if __name__ == "__main__":
    main()
```