K8s 审计日志分析：从海量事件中提取安全与排障线索
一、安全团队问"谁在凌晨 3 点删了 production namespace 的 Secret"，你翻了 20 分钟日志没找到
K8s 审计日志（Audit Log ）记录了集群中所有 API 请求——谁（User/ServiceAccount）、什么时间、做了什么操作（create/update/delete）、操作了哪个资源、结果是什么（成功/拒绝/错误）。这些日志是安全事故溯源和合规审计的最底层数据源。
但审计日志的量极大。一个中等规模的集群（100+ Pods、20+ Services）每天产生的审计日志可以超过 10GB。要从这 10GB 里找到"凌晨 3 点删了 Secret 的元凶"，靠 kubectl logs 或 grep 是不可能的——你需要在日志写入时就做结构化索引和过滤。
审计策略的关键是"只记录你需要的事件"。K8s 支持四级审计策略：None（不记录）、Metadata（只记录请求元数据不含 body）、Request（记录请求 body）、RequestResponse（记录请求和响应 body）。全部开 RequestResponse 会把磁盘写满——必须分层记录。
二、底层机制与原理剖析
审计策略的分层设计：
Metadata 级别（建议对所有 API 请求启用）：记录请求的基本信息——谁、什么操作、什么资源、什么时间、成功/失败。不记录请求和响应的 body，因此存储开销可控。90% 的排障和安全审计场景只需要这些信息。
Request 级别（敏感资源）：专门记录 Secret、ConfigMap、ServiceAccount 等安全敏感资源的请求 body。原因：需要知道"删了什么 Secret 的内容"或"创建了什么 ServiceAccount"。
RequestResponse 级别（极限调试）：记录请求和响应的完整内容。仅在极少数场景使用——如排查某个奇怪的 API 行为。因为响应 body 可能包含大量数据（如 list 操作返回几千个 Pod 的信息）。
三、生产级代码实现
# k8s/audit-policy.yaml
# 分层审计策略：按资源类型和操作设定不同级别
apiVersion: audit.k8s.io/v1
kind: Policy
rules:
# =========================================================
# 级别 1: RequestResponse —— 安全关键操作
# =========================================================
# Secret 的所有操作
- level: RequestResponse
resources:
- group: ""
resources: ["secrets"]
# ServiceAccount 的创建/删除/修改
- level: RequestResponse
resources:
- group: ""
resources: ["serviceaccounts"]
verbs: ["create", "delete", "update", "patch"]
# RBAC 相关（ClusterRole/Role/ClusterRoleBinding/RoleBinding）
- level: RequestResponse
resources:
- group: "rbac.authorization.k8s.io"
resources: ["clusterroles", "roles", "clusterrolebindings", "rolebindings"]
# =========================================================
# 级别 2: Request —— 重要资源操作
# =========================================================
# ConfigMap 的修改
- level: Request
resources:
- group: ""
resources: ["configmaps"]
verbs: ["update", "patch"]
# Deployment/DaemonSet/StatefulSet 的创建和删除
- level: Request
resources:
- group: "apps"
resources: ["deployments", "daemonsets", "statefulsets"]
verbs: ["create", "delete"]
# Pod exec 操作（安全敏感）
- level: Request
resources:
- group: ""
resources: ["pods/exec"]
# =========================================================
# 级别 3: Metadata —— 所有其余操作
# =========================================================
# 不记录只读操作（get/list/watch），避免刷屏
- level: Metadata
verbs: ["create", "update", "patch", "delete"]
# 记录所有非资源 URL 请求（如 /healthz）
- level: Metadata
nonResourceURLs:
- "/healthz*"
- "/metrics"
- "/version"
# =========================================================
# 级别 4: None —— 不记录（排除）
# =========================================================
# 排除 kubelet 和 system: 组件的大量心跳请求
- level: None
users: ["system:kube-proxy", "system:kubelet"]
verbs: ["watch"]
- level: None
userGroups: ["system:nodes"]
verbs: ["get", "update"]
# 排除对 /healthz 的只读请求
- level: None
nonResourceURLs:
- "/healthz*"
- "/readyz*"
verbs: ["get"]
# audit-log-analyzer.py
"""
K8s 审计日志分析器
从 Elasticsearch 查询审计日志并做异常检测
"""
import logging
from typing import List, Dict, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class AuditAlert:
"""审计告警"""
severity: str # critical / high / medium / low
title: str
description: str
user: str
resource: str
verb: str
timestamp: str
namespace: str = ""
class AuditAnalyzer:
"""
审计日志分析器
分析维度：
1. 异常时间操作（非工作时间的关键操作）
2. 权限异常（大量 403 拒绝）
3. 资源异常（非预期地删除生产资源）
"""
# 工作时间窗口（9:00-18:00，周一到周五）
WORK_HOURS_START = 9
WORK_HOURS_END = 18
# 关键操作（触发告警阈值更低）
CRITICAL_VERBS = {"delete", "create"}
CRITICAL_RESOURCES = {"secrets", "serviceaccounts", "clusterroles", "clusterrolebindings"}
# 403 拒绝阈值（短时间内超过此值视为异常）
FORBIDDEN_THRESHOLD = 10 # 5 分钟内超过 10 次 403
def analyze_batch(self, events: List[dict]) -> List[AuditAlert]:
"""批量分析审计事件"""
alerts = []
# 分组分析
alerts.extend(self._detect_off_hours_ops(events))
alerts.extend(self._detect_permission_anomaly(events))
alerts.extend(self._detect_sensitive_deletion(events))
return alerts
def _detect_off_hours_ops(self, events: List[dict]) -> List[AuditAlert]:
"""检测非工作时间的高危操作"""
alerts = []
for event in events:
ts = self._parse_timestamp(event.get("stageTimestamp", ""))
if not ts:
continue
# 工作时间外的操作
hour = ts.hour
weekday = ts.weekday() # 0=Monday
is_off_hours = (
hour < self.WORK_HOURS_START or
hour >= self.WORK_HOURS_END or
weekday >= 5 # 周末
)
if not is_off_hours:
continue
verb = event.get("verb", "")
resource = event.get("objectRef", {}).get("resource", "")
# 只关注关键操作
if verb not in self.CRITICAL_VERBS:
continue
if resource not in self.CRITICAL_RESOURCES:
continue
user = event.get("user", {}).get("username", "unknown")
namespace = event.get("objectRef", {}).get("namespace", "")
alerts.append(AuditAlert(
severity="high",
title=f"非工作时间 {verb} {resource}",
description=f"用户 {user} 在 {ts.strftime('%H:%M')} 执行了 {verb} {resource}（非工作时间）",
user=user,
resource=resource,
verb=verb,
timestamp=ts.isoformat(),
namespace=namespace,
))
return alerts
def _detect_permission_anomaly(self, events: List[dict]) -> List[AuditAlert]:
"""检测权限异常（大量 403）"""
alerts = []
# 按用户聚合 403 计数
forbidden_by_user: Dict[str, List[dict]] = {}
now = datetime.utcnow()
window = timedelta(minutes=5)
for event in events:
if event.get("responseStatus", {}).get("code") != 403:
continue
ts = self._parse_timestamp(event.get("stageTimestamp", ""))
if not ts or (now - ts.replace(tzinfo=None)) > window:
continue
user = event.get("user", {}).get("username", "unknown")
if user not in forbidden_by_user:
forbidden_by_user[user] = []
forbidden_by_user[user].append(event)
for user, user_events in forbidden_by_user.items():
if len(user_events) >= self.FORBIDDEN_THRESHOLD:
resources = set(
e.get("objectRef", {}).get("resource", "unknown")
for e in user_events
)
alerts.append(AuditAlert(
severity="medium",
title=f"大量 403 拒绝",
description=(
f"用户 {user} 在过去 5 分钟内收到 {len(user_events)} 次 403 拒绝，"
f"涉及资源: {', '.join(resources)}"
),
user=user,
resource=", ".join(resources),
verb="多种",
timestamp=now.isoformat(),
))
return alerts
def _detect_sensitive_deletion(self, events: List[dict]) -> List[AuditAlert]:
"""检测敏感资源的删除操作"""
alerts = []
for event in events:
obj_ref = event.get("objectRef", {})
resource = obj_ref.get("resource", "")
verb = event.get("verb", "")
# 删除敏感资源
if verb != "delete":
continue
if resource not in {"secrets", "serviceaccounts", "persistentvolumeclaims"}:
continue
user = event.get("user", {}).get("username", "unknown")
namespace = obj_ref.get("namespace", "")
name = obj_ref.get("name", "unknown")
ts = self._parse_timestamp(event.get("stageTimestamp", ""))
# 检查是否是系统组件（如 garbage collector）
if user.startswith("system:"):
continue
alerts.append(AuditAlert(
severity="critical",
title=f"删除敏感资源: {resource}",
description=f"用户 {user} 删除了 {namespace}/{resource}/{name}",
user=user,
resource=f"{namespace}/{resource}/{name}",
verb=verb,
timestamp=ts.isoformat() if ts else "",
namespace=namespace,
))
return alerts
@staticmethod
def _parse_timestamp(ts_str: str) -> Optional[datetime]:
"""解析日志时间戳"""
try:
return datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
except (ValueError, AttributeError):
return None
四、边界分析与架构权衡
审计日志的存储成本：
全量 RequestResponse 审计日志的存储成本是 Metadata 级别的 5-10 倍
建议：只对安全敏感资源（Secret、RBAC）开 RequestResponse，其余资源 Metadata 足够
日志分析的延迟：
Elasticsearch 的索引写入和查询有一定延迟（通常 2-5 秒）。对于需要"实时"的安全告警，用 webhook 后端直接推送事件到告警引擎
Filebeat/Fluentd 的 tail 模式也有一些延迟（取决于 refresh_interval 配置）
什么操作不该记录：
kubelet 的心跳请求（每秒数百次）——如果记录会让日志膨胀到不可管理
只读操作（get/list/watch）——除非你需要审计"谁看了你的 Secret"
健康检查端点（/healthz、/readyz）
五、总结
K8s 审计日志是安全事故溯源的最后一道防线。审计策略的分层设计是核心——安全敏感资源开 RequestResponse（记录完整内容），一般资源开 Metadata（记录操作元数据），系统心跳开 None（不记录）。配合 Elasticsearch + Kibana 做结构化索引和可视化 ，配合自定义分析器做异常检测（非工作时间高危操作、大量 403 拒绝、敏感资源删除）。审计的关键不是"记录得全"，是"能快速找到需要的信息"。