Kubernetes安全事件深度取证分析

ETCDCTL_API=3 etcdctl --endpoints=https://127.0.0.1:2379 \
  --cacert=/etc/kubernetes/pki/etcd/ca.crt \
  --cert=/etc/kubernetes/pki/etcd/server.crt \
  --key=/etc/kubernetes/pki/etcd/server.key \
  snapshot save /evidence/etcd-snapshot-$(date +%Y%m%d%H%M%S).db

ETCDCTL_API=3 etcdctl snapshot status /evidence/etcd-snapshot.db --write-out=table

ETCDCTL_API=3 etcdctl --endpoints=https://127.0.0.1:2379 \
  --cacert=/etc/kubernetes/pki/etcd/ca.crt \
  --cert=/etc/kubernetes/pki/etcd/server.crt \
  --key=/etc/kubernetes/pki/etcd/server.key \
  get /registry/secrets --prefix --keys-only

cat /etc/kubernetes/encryption-config.yaml | grep -A5 'aescbc:\|aesgcm:\|secretbox:'
kubectl get secrets -A -o json | jq -r '.items[] | select(.metadata.namespace != "kube-system") | "\(.metadata.namespace)/\(.metadata.name)"' | while read secret; do
  ns=$(echo $secret | cut -d/ -f1)
  name=$(echo $secret | cut -d/ -f2)
  kubectl get secret -n $ns $name -o jsonpath='{.data}' | base64 -d 2>/dev/null | head -c 200
  echo ""
done

curl -s https://<etcd-ip>:2379/v2/keys/ --insecure 2>/dev/null | head -c 500
curl -s https://<etcd-ip>:2379/v3/kv/range -X POST -d '{"key":"L3JlZ2lzdHJ5Lw=="}' --insecure 2>/dev/null | head -c 500
ss -tlnp | grep 2379
netstat -tlnp | grep 2379

kubectl auth can-i --list --as=system:serviceaccount:default:default
kubectl auth can-i create pods --as=system:serviceaccount:default:default -n kube-system
kubectl auth can-i '*' '*' --as=system:serviceaccount:default:default
kubectl get clusterrolebindings -o json | jq -r '.items[] | select(.subjects != null) | select(.roleRef.name == "cluster-admin") | {binding: .metadata.name, subjects: [.subjects[] | {kind: .kind, name: .name, namespace: .namespace}]}'

kubectl get clusterrolebindings -o json | jq -r '.items[] | select(.roleRef.name == "cluster-admin") | "\(.metadata.name): \([.subjects[]? | "\(.kind)/\(.name)@\(.namespace // "cluster")"] | join(", "))"'
kubectl get rolebindings -A -o json | jq -r '.items[] | select(.roleRef.name == "cluster-admin" or .roleRef.name == "admin" or .roleRef.name == "edit") | "\(.metadata.namespace)/\(.metadata.name): role=\(.roleRef.name) subjects=[\([.subjects[]? | "\(.kind)/\(.name)"] | join(", "))]"'

kubectl get events --field-selector reason=RoleBinding --all-namespaces --sort-by='.lastTimestamp' | tail -20
kubectl get events --field-selector reason=ClusterRoleBinding --all-namespaces --sort-by='.lastTimestamp' | tail -20

kubectl get pods -A -o json | jq -r '.items[] | select(.spec.containers[]?.securityContext.privileged == true or .spec.containers[]?.securityContext.capabilities.add[]? == "SYS_ADMIN" or .spec.hostPID == true or .spec.hostNetwork == true or .spec.hostIPC == true) | "\(.metadata.namespace)/\(.metadata.name): privileged=\(.spec.containers[0].securityContext.privileged // false) hostPID=\(.spec.hostPID // false) hostNetwork=\(.spec.hostNetwork // false) hostIPC=\(.spec.hostIPC // false)"'

kubectl get pods -A -o json | jq -r '.items[] | .spec.volumes[]? | select(.hostPath != null) | "\(.name): \(.hostPath.path) (type: \(.hostPath.type // "default"))"'

journalctl -u containerd --since "2026-07-15 00:00:00" --until "2026-07-16 00:00:00" | grep -i "creat\|start\|stop\|exec\|died\|oom\|kill"
ls -la /var/log/pods/
kubectl logs <pod-name> -n <namespace> --previous --tail=1000

crane manifest <image>:<tag>
crane export <image>:<tag> /evidence/image-export.tar
tar tf /evidence/image-export.tar | head -50
trivy image --severity HIGH,CRITICAL <image>:<tag>
trivy image --scanners vuln,misconfig,secret <image>:<tag>

apiVersion: audit.k8s.io/v1
kind: Policy
rules:
  - level: None
    resources:
      - group: ""
        resources: ["endpoints", "services/status"]
    users: ["system:kube-proxy"]
  - level: None
    resources:
      - group: ""
        resources: ["nodes", "nodes/status"]
    verbs: ["get"]
    users: ["system:node-problem-detector"]
  - level: Metadata
    resources:
      - group: ""
        resources: ["secrets", "configmaps"]
  - level: RequestResponse
    resources:
      - group: "rbac.authorization.k8s.io"
    omitStages:
      - "RequestReceived"
  - level: Metadata
    omitStages:
      - "RequestReceived"

cat /var/log/k8s-audit/audit.log | jq 'select(.verb == "create" or .verb == "patch" or .verb == "update" or .verb == "delete") | select(.objectRef.resource == "clusterrolebindings" or .objectRef.resource == "rolebindings" or .objectRef.resource == "secrets" or .objectRef.resource == "pods") | {timestamp: .requestReceivedTimestamp, user: .user.username, verb: .verb, resource: .objectRef.resource, namespace: .objectRef.namespace, name: .objectRef.name, code: .responseStatus.code, sourceIP: .sourceIPs[0]}'

cat /var/log/k8s-audit/audit.log | jq 'select(.user.username == "system:anonymous" or (.user.username | startswith("system:serviceaccount:")) and .verb == "list" and .objectRef.resource == "secrets") | {timestamp: .requestReceivedTimestamp, user: .user.username, verb: .verb, resource: .objectRef.resource, namespace: .objectRef.namespace, code: .responseStatus.code}'

TOKEN=$(cat /var/run/secrets/kubernetes.io/serviceaccount/token)
kubectl auth can-i --list --token=$TOKEN
kubectl auth can-i --list --token=$TOKEN -n default
kubectl auth can-i get secrets --token=$TOKEN -A
kubectl auth can-i create pods --token=$TOKEN -n kube-system

kubectl get pods -A -o json | jq -r '.items[] | select(.spec.automountServiceAccountToken != false) | select(.metadata.namespace != "kube-system") | "\(.metadata.namespace)/\(.metadata.name): sa=\(.spec.serviceAccountName // "default") automount=\(.spec.automountServiceAccountToken // true)"'

kubectl get pods -A -o json | jq -r '.items[] | select(.spec.automountServiceAccountToken != false) | "\(.metadata.namespace)/\(.metadata.name) sa=\(.spec.serviceAccountName // "default")"'
kubectl get serviceaccounts -A -o json | jq -r '.items[] | select(.automountServiceAccountToken != false) | "\(.metadata.namespace)/\(.metadata.name) automount=\(.automountServiceAccountToken // true)"'

kubectl get networkpolicies -A -o json | jq -r '.items[] | "\(.metadata.namespace)/\(.metadata.name): ingress=\(.spec.ingress != null) egress=\(.spec.egress != null) podSelector=\(.spec.podSelector)"'
kubectl get pods -A -o json | jq -r '.items[] | "\(.metadata.namespace)/\(.metadata.name)"' | while read pod; do
  ns=$(echo $pod | cut -d/ -f1)
  name=$(echo $pod | cut -d/ -f2)
  np=$(kubectl get networkpolicies -n $ns -o json | jq -r --arg ns "$ns" --arg name "$name" '.items[] | select(.spec.podSelector.matchLabels as $ml | $ml | to_entries | all(. as $k | .key as $k | $k as $k | .value as $v | true)) | .metadata.name')
  if [ -z "$np" ]; then
    echo "NO POLICY: $pod"
  fi
done

kubectl get pods -A -o wide -o json | jq -r '.items[] | "\(.metadata.namespace)/\(.metadata.name) node=\(.spec.nodeName) ip=\(.status.podIP)"'
kubectl exec -n <namespace> <pod> -- ss -tlnp
kubectl exec -n <namespace> <pod> -- cat /etc/resolv.conf

trivy image --scanners vuln,misconfig,secret,license --severity HIGH,CRITICAL <image>:<tag>
trivy fs --scanners vuln,misconfig,secret --severity HIGH,CRITICAL /path/to/dockerfile
grype <image>:<tag> --fail-on high
docker scout cves <image>:<tag> --only-severity critical,high

crane digest <image>:<tag>
crane manifest <image>:<tag> | jq '.config'
crane config <image>:<tag> | jq '.config.Env'
crane config <image>:<tag> | jq '.config.Entrypoint'
crane config <image>:<tag> | jq '.config.Cmd'

cosign verify --key cosign.pub <registry>/<image>:<tag>
cosign verify --key cosign.pub --annotations "author=<expected>" <registry>/<image>:<tag>
cosign inspect <registry>/<image>:<tag>
cosign verify-attestation --key cosign.pub --type spdxjson <registry>/<image>:<tag>
cosign list-tags <registry>/<image>

curl -s -k https://<registry>/api/v2.0/projects | jq '.[] | {name: .name, public: .public, metadata: .metadata}'
curl -s -k -u admin:password https://<registry>/api/v2.0/projects/<project>/repositories | jq '.[] | {name: .name, pull_count: .pull_count}'
curl -s -k -u admin:password https://<registry>/api/v2.0/projects/<project>/webhook/policies | jq '.[] | {name: .name, enabled: .enabled, event_types: .event_types}'

title: Kubernetes RBAC Privilege Escalation via ClusterRoleBinding Creation
id: k8s-rbac-privilege-escalation-clusterrolebinding
status: experimental
description: Detects creation of ClusterRoleBinding that grants cluster-admin or equivalent privileges, which may indicate privilege escalation
references:
  - https://kubernetes.io/docs/reference/access-authn-authz/rbac/
author: Security Operations
date: 2026-07-16
tags:
  - attack.privilege_escalation
  - attack.t1098
  - kubernetes
logsource:
  product: kubernetes
  service: audit
detection:
  selection:
    objectRef.resource: clusterrolebindings
    verb:
      - create
      - patch
      - update
    responseStatus.code: 200
  filter_legitimate:
    user.username|contains:
      - 'system:serviceaccount:kube-system:'
      - 'system:kube-controller-manager'
  condition: selection and not filter_legitimate
level: high
falsepositives:
  - Legitimate cluster-admin provisioning by infrastructure automation

#!/bin/bash
echo "=========================================="
echo "Kubernetes Security Audit Script"
echo "=========================================="
echo ""
echo "[*] Checking for privileged pods..."
kubectl get pods -A -o json | jq -r '.items[] | select(.spec.containers[]?.securityContext.privileged == true) | "ALERT: Privileged Pod - \(.metadata.namespace)/\(.metadata.name) image=\(.spec.containers[0].image)"'
echo ""
echo "[*] Checking for hostPID/hostNetwork/hostIPC..."
kubectl get pods -A -o json | jq -r '.items[] | select(.spec.hostPID == true or .spec.hostNetwork == true or .spec.hostIPC == true) | "ALERT: Host Namespace Pod - \(.metadata.namespace)/\(.metadata.name) hostPID=\(.spec.hostPID // false) hostNetwork=\(.spec.hostNetwork // false) hostIPC=\(.spec.hostIPC // false)"'
echo ""
echo "[*] Checking for hostPath mounts..."
kubectl get pods -A -o json | jq -r '.items[] | .spec.volumes[]? | select(.hostPath != null) | select(.hostPath.path == "/etc" or .hostPath.path == "/var/run/docker.sock" or .hostPath.path == "/" or .hostPath.path == "/proc" or .hostPath.path == "/root") | "ALERT: Sensitive hostPath mount found"'
echo ""
echo "[*] Checking cluster-admin bindings..."
kubectl get clusterrolebindings -o json | jq -r '.items[] | select(.roleRef.name == "cluster-admin") | "ALERT: cluster-admin binding - \(.metadata.name) -> [\([.subjects[]? | "\(.kind)/\(.name)"] | join(", "))]"'
echo ""
echo "[*] Checking for automountServiceAccountToken..."
kubectl get pods -A -o json | jq -r '.items[] | select(.spec.automountServiceAccountToken != false) | select(.metadata.namespace != "kube-system") | "WARNING: SA Token automounted - \(.metadata.namespace)/\(.metadata.name) sa=\(.spec.serviceAccountName // "default")"'
echo ""
echo "[*] Checking for unnamed/anonymous users with RBAC bindings..."
kubectl get rolebindings -A -o json | jq -r '.items[] | select(.subjects[]? | select(.name == "system:anonymous" or .name == "system:unauthenticated")) | "ALERT: Anonymous binding - \(.metadata.namespace)/\(.metadata.name) role=\(.roleRef.name)"'
echo ""
echo "[*] Checking exposed Services..."
kubectl get svc -A -o json | jq -r '.items[] | select(.spec.type == "NodePort" or .spec.type == "LoadBalancer") | "WARNING: Exposed Service - \(.metadata.namespace)/\(.metadata.name) type=\(.spec.type) ports=[\([.spec.ports[]? | "\(.port)/\(.protocol)"] | join(", "))]"'
echo ""
echo "[*] Checking for container images from public registries..."
kubectl get pods -A -o json | jq -r '.items[] | .spec.containers[].image | select(test("^docker.io/|^gcr.io/|^quay.io/|^k8s.gcr.io/") | not and test("/") | not) | "WARNING: Public registry image - \(.)" | sort -u'
echo ""
echo "[*] Checking for NetworkPolicy coverage..."
ALL_NS=$(kubectl get ns -o jsonpath='{.items[*].metadata.name}')
for ns in $ALL_NS; do
  NP_COUNT=$(kubectl get networkpolicies -n $ns --no-headers 2>/dev/null | wc -l | tr -d ' ')
  POD_COUNT=$(kubectl get pods -n $ns --no-headers 2>/dev/null | wc -l | tr -d ' ')
  if [ "$POD_COUNT" -gt 0 ] && [ "$NP_COUNT" -eq 0 ]; then
    echo "WARNING: Namespace $ns has $POD_COUNT pods but 0 NetworkPolicies"
  fi
done
echo ""
echo "[*] Checking for images with known critical vulnerabilities..."
kubectl get pods -A -o json | jq -r '.items[] | "\(.metadata.namespace)/\(.metadata.name) \(.spec.containers[0].image)"' | head -20 | while read line; do
  image=$(echo $line | awk '{print $2}')
  result=$(trivy image --severity CRITICAL --quiet --format json $image 2>/dev/null | jq '.Results[]?.Vulnerabilities[]? | .VulnerabilityID' 2>/dev/null | wc -l | tr -d ' ')
  if [ "$result" -gt 0 ]; then
    echo "CRITICAL: $line has $result critical vulnerabilities"
  fi
done
echo ""
echo "=========================================="
echo "Audit completed at $(date)"
echo "=========================================="

#!/usr/bin/env python3
import json
import subprocess
import sys
from datetime import datetime, timedelta

def run_kubectl(args):
    result = subprocess.run(["kubectl"] + args, capture_output=True, text=True, timeout=60)
    if result.returncode != 0:
        return None
    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        return result.stdout

def check_privileged_pods():
    findings = []
    data = run_kubectl(["get", "pods", "-A", "-o", "json"])
    if not data:
        return findings
    for pod in data.get("items", []):
        ns = pod["metadata"]["namespace"]
        name = pod["metadata"]["name"]
        for c in pod.get("spec", {}).get("containers", []):
            sc = c.get("securityContext", {})
            if sc.get("privileged") is True:
                findings.append({
                    "severity": "CRITICAL",
                    "type": "Privileged Container",
                    "namespace": ns,
                    "name": name,
                    "image": c.get("image", "unknown"),
                    "detail": "Container runs in privileged mode"
                })
            caps = sc.get("capabilities", {}).get("add", [])
            dangerous_caps = [cap for cap in caps if cap in ["SYS_ADMIN", "SYS_PTRACE", "SYS_MODULE", "NET_ADMIN", "SYS_RAWIO"]]
            if dangerous_caps:
                findings.append({
                    "severity": "HIGH",
                    "type": "Dangerous Capabilities",
                    "namespace": ns,
                    "name": name,
                    "image": c.get("image", "unknown"),
                    "detail": f"Dangerous capabilities: {', '.join(dangerous_caps)}"
                })
        spec = pod.get("spec", {})
        if spec.get("hostPID") is True:
            findings.append({
                "severity": "HIGH",
                "type": "Host PID Namespace",
                "namespace": ns,
                "name": name,
                "image": "",
                "detail": "Pod uses host PID namespace"
            })
        if spec.get("hostNetwork") is True:
            findings.append({
                "severity": "HIGH",
                "type": "Host Network",
                "namespace": ns,
                "name": name,
                "image": "",
                "detail": "Pod uses host network namespace"
            })
        for vol in spec.get("volumes", []):
            hp = vol.get("hostPath", {})
            path = hp.get("path", "")
            sensitive_paths = ["/etc", "/var/run/docker.sock", "/proc", "/sys", "/root", "/var/log"]
            for sp in sensitive_paths:
                if path == sp or path.startswith(sp + "/"):
                    findings.append({
                        "severity": "HIGH",
                        "type": "Sensitive HostPath Mount",
                        "namespace": ns,
                        "name": name,
                        "image": "",
                        "detail": f"Volume '{vol.get('name')}' mounts sensitive path: {path}"
                    })
    return findings

def check_rbac_bindings():
    findings = []
    data = run_kubectl(["get", "clusterrolebindings", "-o", "json"])
    if not data:
        return findings
    for binding in data.get("items", []):
        if binding.get("roleRef", {}).get("name") == "cluster-admin":
            subjects = binding.get("subjects", []) or []
            for s in subjects:
                if s.get("kind") == "User" and not s.get("name", "").startswith("system:"):
                    findings.append({
                        "severity": "CRITICAL",
                        "type": "ClusterAdmin Binding",
                        "resource": binding["metadata"]["name"],
                        "detail": f"User '{s.get('name')}' has cluster-admin role"
                    })
                elif s.get("kind") == "ServiceAccount" and s.get("namespace") != "kube-system":
                    findings.append({
                        "severity": "CRITICAL",
                        "type": "ClusterAdmin SA Binding",
                        "resource": binding["metadata"]["name"],
                        "detail": f"SA '{s.get('namespace')}/{s.get('name')}' has cluster-admin role"
                    })
    return findings

def check_audit_log_anomalies(log_path):
    findings = []
    try:
        with open(log_path, 'r') as f:
            for line in f:
                try:
                    entry = json.loads(line.strip())
                except json.JSONDecodeError:
                    continue
                user = entry.get("user", {}).get("username", "")
                verb = entry.get("verb", "")
                resource = entry.get("objectRef", {}).get("resource", "")
                code = entry.get("responseStatus", {}).get("code", 0)
                if user == "system:anonymous" and code == 200:
                    findings.append({
                        "severity": "CRITICAL",
                        "type": "Anonymous API Access",
                        "detail": f"Anonymous user successfully {verb} {resource}",
                        "timestamp": entry.get("requestReceivedTimestamp", "")
                    })
                if verb in ["create", "patch", "update"] and resource in ["clusterrolebindings", "rolebindings"]:
                    findings.append({
                        "severity": "HIGH",
                        "type": "RBAC Modification",
                        "detail": f"User '{user}' {verb} {resource}",
                        "timestamp": entry.get("requestReceivedTimestamp", "")
                    })
                if verb == "create" and resource == "pods/exec":
                    findings.append({
                        "severity": "HIGH",
                        "type": "Pod Exec",
                        "detail": f"User '{user}' exec into pod in {entry.get('objectRef', {}).get('namespace', 'unknown')}",
                        "timestamp": entry.get("requestReceivedTimestamp", "")
                    })
    except FileNotFoundError:
        print(f"[!] Audit log not found: {log_path}")
    return findings

def main():
    print("=" * 60)
    print("Kubernetes Security Audit - Automated Detection")
    print(f"Scan Time: {datetime.now().isoformat()}")
    print("=" * 60)
    all_findings = []
    print("\n[*] Checking privileged pods and dangerous capabilities...")
    findings = check_privileged_pods()
    all_findings.extend(findings)
    for f in findings:
        print(f"  [{f['severity']}] {f['type']}: {f.get('namespace', '')}/{f.get('name', '')} - {f['detail']}")
    print(f"\n[*] Checking RBAC bindings...")
    findings = check_rbac_bindings()
    all_findings.extend(findings)
    for f in findings:
        print(f"  [{f['severity']}] {f['type']}: {f['detail']}")
    audit_log = "/var/log/k8s-audit/audit.log"
    print(f"\n[*] Analyzing audit log: {audit_log}")
    findings = check_audit_log_anomalies(audit_log)
    all_findings.extend(findings)
    for f in findings:
        print(f"  [{f['severity']}] {f['type']}: {f['detail']} @ {f.get('timestamp', 'N/A')}")
    print("\n" + "=" * 60)
    critical = sum(1 for f in all_findings if f["severity"] == "CRITICAL")
    high = sum(1 for f in all_findings if f["severity"] == "HIGH")
    medium = sum(1 for f in all_findings if f["severity"] == "MEDIUM")
    print(f"Summary: {critical} CRITICAL, {high} HIGH, {medium} MEDIUM")
    print(f"Total findings: {len(all_findings)}")
    print("=" * 60)
    report = {
        "scan_time": datetime.now().isoformat(),
        "total_findings": len(all_findings),
        "critical": critical,
        "high": high,
        "medium": medium,
        "findings": all_findings
    }
    with open("/evidence/k8s-audit-report.json", "w") as f:
        json.dump(report, f, indent=2, ensure_ascii=False)
    print(f"\nReport saved to /evidence/k8s-audit-report.json")
    if critical > 0:
        sys.exit(2)
    elif high > 0:
        sys.exit(1)
    else:
        sys.exit(0)

if __name__ == "__main__":
    main()