深入kube-apiserver审计机制：从策略配置到事件记录的全流程解析

前言

去年处理过一次安全事件：有人删除了生产环境的核心ConfigMap，导致服务中断。事后排查时，我们无法确定是谁在什么时间执行了删除操作——因为apiserver的审计日志没有开启。

这次事件让我深刻认识到审计（Audit）的重要性。审计日志是K8s安全运营的基石，它记录了谁在什么时间做了什么操作，是事后追溯和安全分析的关键证据。

今天就带大家深入源码，看看kube-apiserver的审计机制是如何实现的。

什么是审计？

K8s审计功能提供了按时间顺序排列的安全相关记录集，记录了：

• 每个用户对API的操作
• 使用K8s API的应用的行为
• 控制面自身的活动

审计能回答的问题：

审计架构概览

用户请求
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ HTTP Handler Chain │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ WithAudit Filter │ │
│ │ │ │
│ │ 1. 创建审计事件 │ │
│ │ 2. 根据策略评估审计级别 │ │
│ │ 3. 记录请求接收（StageRequestReceived） │ │
│ │ 4. 包装ResponseWriter │ │
│ │ 5. 记录响应开始（StageResponseStarted） │ │
│ │ 6. 记录响应完成（StageResponseComplete） │ │
│ └─────────────────────────────────────────────────────────┘ │
└─────────────────────────┬───────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ Audit Backend │
│ │
│ ┌─────────────────────┐ ┌─────────────────────┐ │
│ │ Log Backend │ │ Webhook Backend │ │
│ │ │ │ │ │
│ │ –audit-log-path │ │ –audit-webhook- │ │
│ │ │ │ config-file │ │
│ └─────────────────────┘ └─────────────────────┘ │
│ │ │ │
│ └──────────┬───────────────┘ │
│ ▼ │
│ Union Backend │
└─────────────────────────────────────────────────────────────────┘

审计策略的4个级别

审计策略定义了记录什么内容，有4个级别：

策略配置示例：

# audit-policy.yaml
apiVersion: audit.k8s.io/v1
kind: Policy
rules:
# 不记录健康检查
– level: None
nonResourceURLs:
– /healthz
– /livez
– /readyz

# 记录ConfigMap的变更，包含请求体
– level: Request
resources:
– group: ""
resources: ["configmaps"]
verbs: ["create", "update", "patch", "delete"]

# 记录Secret的所有操作，包含请求和响应
– level: RequestResponse
resources:
– group: ""
resources: ["secrets"]

# 默认只记录元数据
– level: Metadata

源码解析：审计的初始化

初始化入口

审计的初始化在buildGenericConfig中完成：

// cmd/kube-apiserver/app/server.go
func buildGenericConfig(s *options.ServerRunOptions, …) (genericConfig, …) {
// … 其他配置

// 应用审计配置
if lastErr = s.Audit.ApplyTo(genericConfig); lastErr != nil {
return
}
}

ApplyTo方法：构建审计后端

// pkg/kubeapiserver/options/audit.go

func (o *AuditOptions) ApplyTo(c *server.Config) error {
// 1. 构建策略评估器（根据audit-policy-file）
evaluator, err := o.newPolicyRuleEvaluator()
if err != nil {
return err
}

// 2. 构建日志后端（–audit-log-path）
var logBackend audit.Backend
w, err := o.LogOptions.getWriter()
if err != nil {
return err
}

if w != nil {
if evaluator == nil {
klog.V(2).Info("No audit policy file provided, no events will be recorded for log backend")
} else {
logBackend = o.LogOptions.newBackend(w)
}
}

// 3. 构建webhook后端（–audit-webhook-config-file）
var webhookBackend audit.Backend
if o.WebhookOptions.enabled() {
if evaluator == nil {
klog.V(2).Info("No audit policy file provided, no events will be recorded for webhook backend")
} else {
webhookBackend, err = o.WebhookOptions.newUntruncatedBackend(egressDialer)
if err != nil {
return err
}
}
}

// 4. 封装为动态后端（支持截断）
var dynamicBackend audit.Backend
if webhookBackend != nil {
dynamicBackend = o.WebhookOptions.TruncateOptions.wrapBackend(webhookBackend, groupVersion)
}

// 5. 设置策略评估器
c.AuditPolicyRuleEvaluator = evaluator

// 6. 合并所有后端
c.AuditBackend = appendBackend(logBackend, dynamicBackend)

return nil
}

审计后端详解

1. 日志后端（Log Backend）

日志后端将审计事件写入本地文件。

// staging/src/k8s.io/apiserver/plugin/pkg/audit/log/backend.go

type backend struct {
out io.Writer // 输出流
format string // 格式：legacy或json
encoder runtime.Encoder
}

// 创建日志后端
func (o *AuditLogOptions) newBackend(w io.Writer) audit.Backend {
return &backend{
out: w,
format: o.Format,
}
}

// 处理审计事件
func (b *backend) ProcessEvents(events …*auditinternal.Event) bool {
success := true
for _, ev := range events {
success = b.logEvent(ev) && success
}
return success
}

func (b *backend) logEvent(ev *auditinternal.Event) bool {
line := ""
switch b.format {
case FormatLegacy:
line = audit.EventString(ev) + "\\n"
case FormatJson:
bs, err := runtime.Encode(b.encoder, ev)
if err != nil {
audit.HandlePluginError(PluginName, err, ev)
return false
}
line = string(bs[:])
}

// 写入日志
if _, err := fmt.Fprint(b.out, line); err != nil {
audit.HandlePluginError(PluginName, err, ev)
return false
}
return true
}

日志轮转：使用lumberjack库实现自动轮转

import "gopkg.in/natefinch/lumberjack.v2"

return &lumberjack.Logger{
Filename: o.Path, // 日志文件路径
MaxAge: o.MaxAge, // 最大保留天数
MaxBackups: o.MaxBackups, // 最大备份数
MaxSize: o.MaxSize, // 单个文件最大大小（MB）
Compress: o.Compress, // 是否压缩
}, nil

2. Webhook后端

Webhook后端将审计事件发送到远程HTTP服务。

// staging/src/k8s.io/apiserver/plugin/pkg/audit/webhook/webhook.go

func (o *AuditWebhookOptions) newUntruncatedBackend(egressDialer utilnet.DialFunc) (audit.Backend, error) {
// 创建REST客户端
webhookClient, err := o.webhookClient(egressDialer)
if err != nil {
return nil, err
}

return &backend{
webhookClient: webhookClient,
}, nil
}

// 发送审计事件到webhook
func (b *backend) ProcessEvents(events …*auditinternal.Event) bool {
success := true
for _, ev := range events {
success = b.sendEvent(ev) && success
}
return success
}

func (b *backend) sendEvent(ev *auditinternal.Event) bool {
// 发送HTTP POST请求
result := b.webhookClient.Create()
if err := result.Error(); err != nil {
audit.HandlePluginError(PluginName, err, ev)
return false
}
return true
}

Webhook配置：

# webhook-config.yaml
apiVersion: v1
kind: Config
clusters:
– name: audit–server
cluster:
certificate-authority: /path/to/ca.crt
server: https://audit.example.com/webhook

users:
– name: apiserver
user:
client-certificate: /path/to/client.crt
client-key: /path/to/client.key

current-context: webhook
contexts:
– context:
cluster: audit–server
user: apiserver
name: webhook

3. Union后端：多后端组合

// staging/src/k8s.io/apiserver/pkg/audit/union.go

// Union将多个后端组合成一个
type unionBackend struct {
backends []audit.Backend
}

func Union(backends …audit.Backend) audit.Backend {
return unionBackend{backends: backends}
}

func (u unionBackend) ProcessEvents(events …*auditinternal.Event) bool {
success := true
for _, backend := range u.backends {
success = backend.ProcessEvents(events…) && success
}
return success
}

HTTP处理链中的审计

审计是在HTTP handler chain中通过WithAudit中间件实现的。

WithAudit中间件

// staging/src/k8s.io/apiserver/pkg/endpoints/filters/audit.go

func WithAudit(
handler http.Handler,
sink audit.Sink,
policy audit.PolicyRuleEvaluator,
) http.Handler {
if sink == nil || policy == nil {
return handler
}

return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
// 1. 创建审计事件并附加到context
req, ev, omitStages, err := createAuditEventAndAttachToContext(req, policy)
if err != nil {
responsewriters.InternalError(w, req, errors.New("failed to create audit event"))
return
}

// 2. 如果没有事件（策略为None），直接处理
if ev == nil {
handler.ServeHTTP(w, req)
return
}

ctx := req.Context()

// 3. 记录请求接收阶段
ev.Stage = auditinternal.StageRequestReceived
processAuditEvent(ctx, sink, ev, omitStages)

// 4. 包装ResponseWriter以拦截响应
respWriter := decorateResponseWriter(ctx, w, ev, sink, omitStages)

// 5. 使用defer确保响应完成阶段被记录
defer func() {
if r := recover(); r != nil {
// 记录panic
ev.Stage = auditinternal.StagePanic
ev.ResponseStatus = &metav1.Status{
Code: http.StatusInternalServerError,
Status: metav1.StatusFailure,
}
processAuditEvent(ctx, sink, ev, omitStages)
panic(r)
}

// 记录响应完成
ev.Stage = auditinternal.StageResponseComplete
processAuditEvent(ctx, sink, ev, omitStages)
}()

// 6. 处理请求
handler.ServeHTTP(respWriter, req)
})
}

审计事件的3个阶段

请求处理时间线
─────────────────────────────────────────────────────────────►

│ │ │
│ │ │
▼ ▼ ▼
RequestReceived ResponseStarted ResponseComplete
（请求接收） （响应开始） （响应完成）

│────────────────────│─────────────────────│
长运行请求 响应发送

创建审计事件

func createAuditEventAndAttachToContext(
req *http.Request,
policy audit.PolicyRuleEvaluator,
) (*http.Request, *auditinternal.Event, []auditinternal.Stage, error) {

// 获取请求信息
ctx := req.Context()
attribs, err := GetAuthorizerAttributes(ctx)

// 评估审计级别
level, omitStages := policy.LevelAndStages(attribs)
if level == auditinternal.LevelNone {
return req, nil, nil, nil // 不记录
}

// 创建审计事件
ev := &auditinternal.Event{
Timestamp: metav1.NowMicro(),
AuditID: types.UID(uuid.New().String()),
Level: level,
Verb: attribs.GetVerb(),
RequestURI: req.URL.RequestURI(),
User: attribs.GetUser(),
SourceIPs: sourceIPs(req),
ObjectRef: objectRef(attribs),
}

// 根据级别记录请求体
if level >= auditinternal.LevelRequest {
ev.RequestObject = recordRequestObject(req, level)
}

// 将事件附加到context
ctx = audit.WithAuditContext(ctx, ev)
req = req.WithContext(ctx)

return req, ev, omitStages, nil
}

配置审计

基本配置

kube-apiserver \\
–audit-policy-file=/etc/kubernetes/audit-policy.yaml \\
–audit-log-path=/var/log/kubernetes/audit.log \\
–audit-log-format=json \\
–audit-log-maxsize=100 \\
–audit-log-maxbackup=10 \\
–audit-log-maxage=30

高级配置：Webhook

kube-apiserver \\
–audit-policy-file=/etc/kubernetes/audit-policy.yaml \\
–audit-webhook-config-file=/etc/kubernetes/audit-webhook.yaml \\
–audit-webhook-mode=batch \\
–audit-webhook-batch-max-size=100 \\
–audit-webhook-batch-max-wait=1s

Webhook模式：

• blocking：同步发送，可能影响API响应时间
• batching：批量异步发送，性能更好

审计日志分析

日志示例

{
"kind": "Event",
"apiVersion": "audit.k8s.io/v1",
"level": "Request",
"auditID": "c5d4e6f7-a8b9-4c0d-1e2f-3a4b5c6d7e8f",
"stage": "ResponseComplete",
"requestURI": "/api/v1/namespaces/default/pods/nginx",
"verb": "create",
"user": {
"username": "admin",
"groups": ["system:masters", "system:authenticated"]
},
"sourceIPs": ["192.168.1.100"],
"objectRef": {
"resource": "pods",
"namespace": "default",
"name": "nginx"
},
"responseStatus": {
"code": 201
},
"requestObject": {
"apiVersion": "v1",
"kind": "Pod",
"metadata": {
"name": "nginx",
"namespace": "default"
},
"spec": {
"containers": [{
"name": "nginx",
"image": "nginx:1.19"
}]
}
},
"timestamp": "2024-01-15T10:30:00.123456Z"
}

实用查询

# 查找删除操作
jq 'select(.verb == "delete")' /var/log/kubernetes/audit.log

# 查找特定用户的操作
jq 'select(.user.username == "admin")' /var/log/kubernetes/audit.log

# 查找失败的操作
jq 'select(.responseStatus.code >= 400)' /var/log/kubernetes/audit.log

# 统计各用户的操作次数
jq -r '.user.username' /var/log/kubernetes/audit.log | sort | uniq -c | sort -rn

踩坑实录：审计常见问题

坑1：审计日志文件过大

现象：磁盘被审计日志占满

解决方案：

# 配置日志轮转和压缩
kube-apiserver \\
–audit-log-maxsize=100 \\ # 单个文件100MB
–audit-log-maxbackup=10 \\ # 保留10个备份
–audit-log-maxage=30 \\ # 保留30天
–audit-log-compress=true # 压缩备份

坑2：审计影响性能

现象：开启审计后API响应变慢

解决方案：

# 1. 使用较宽松的策略
# 对高频读操作使用Metadata级别

# 2. 使用Webhook batch模式
–audit-webhook-mode=batch
–audit-webhook-batch-max-size=100
–audit-webhook-batch-max-wait=1s

# 3. 异步后端
–audit-log-mode=async

坑3：审计事件丢失

现象：高负载时部分审计事件没有记录

根因：后端处理不过来，事件被丢弃

解决方案：

# 增加缓冲区大小
–audit-webhook-truncate-max-batch-size=10000
–audit-webhook-truncate-max-event-size=102400

坑4：敏感信息泄露

现象：审计日志中包含Secret的明文内容

解决方案：

# 对Secret使用Metadata级别
apiVersion: audit.k8s.io/v1
kind: Policy
rules:
– level: Metadata
resources:
– group: ""
resources: ["secrets"]

审计最佳实践

1. 分层审计策略

apiVersion: audit.k8s.io/v1
kind: Policy
rules:
# 不记录系统组件的读操作
– level: None
users: ["system:kube-proxy", "system:kubelet"]
verbs: ["get", "list", "watch"]

# 详细记录敏感资源
– level: RequestResponse
resources:
– group: "rbac.authorization.k8s.io"
resources: ["roles", "rolebindings", "clusterroles", "clusterrolebindings"]

# 记录默认
– level: Metadata

2. 集中化审计日志

# 使用Webhook将审计日志发送到集中存储
kube-apiserver \\
–audit-webhook-config-file=/etc/kubernetes/audit-webhook.yaml \\
–audit-webhook-mode=batch

3. 定期审计分析

# 查找异常操作
jq 'select(.responseStatus.code >= 400)' audit.log | jq -s 'group_by(.user.username) | map({user: .[0].user.username, count: length})'

# 查找权限提升操作
jq 'select(.verb == "create" or .verb == "update") | select(.objectRef.resource | contains("role"))' audit.log

总结

通过今天的分析，我们深入理解了kube-apiserver的审计机制：

审计是K8s安全运营的基础设施，正确配置审计对事后追溯和合规非常重要。