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 拒绝、敏感资源删除）。审计的关键不是"记录得全"，是"能快速找到需要的信息"。
