在 Kubernetes 集群里做安全巡检很多人第一反应是装一个商业级的云安全平台或者用 kube-bench、kube-hunter 这类现成工具跑一遍。说实话这些我都用过但实际维护多个集群的时候总会遇到几类问题工具太重、策略太死、报告太长没人看、又或者只是想知道“我自己集群里有没有明显配置错误”不想为这个专门搭一套重型平台。后来我抽了个周末用 Python 写了一个轻量级的 K8s 漏洞扫描器专门用来自动发现集群内的配置错误。这篇文章就是把整个思路和关键实现拆开来讲包括我怎么设计检查项、怎么写采集层、以及踩到的一些坑。这套扫描器不是什么黑科技本质上就是把 kubectl get 和 Python Client 能拿到的资源数据汇总起来再用一堆规则去判断“这样配置到底合不合理”。核心价值在于规则可自己维护、输出可定制、能直接集成进自己的巡检脚本或者 CI 流程里。适合正在维护 K8s 集群但又不想引入重平台的运维、SRE以及想入门云原生安全开发的 Python 开发者。1. 整体设计思路先想清楚要查什么而不是先写代码我见过不少人写这类工具时上来就写client.CoreV1().list_pod_for_all_namespaces()把 pod 列出来然后输出一堆“已发现 N 个异常”但到最后也不知道异常到底该怎么判断。在动手之前应该先问自己一个问题一个配置错误的 K8s 集群长什么样1.1 配置错误不等于漏洞但往往是漏洞的入口K8s 的绝大多数安全事件都不是靠 0day 打进来的而是某个配置项大意了。比如 Deployment 里的容器用了 root 用户启动且没有设置allowPrivilegeEscalation: false比如某个 Service 把数据库端口暴露成 NodePort 而且没有加来源 IP 限制再比如 etcd、kubelet 的证书权限过大某个低权限的 ServiceAccount 居然能 list 所有 secret。这些单独看都不是“漏洞”但组合起来就是攻击路径。所以这个扫描器设计的第一原则是不追求发现真实攻击链只做配置基线检查。换句话说它要回答的是“你的集群目前有哪些配置不符合安全最佳实践”而不是“你是否已经被入侵”。这个定位非常关键决定了整个项目的复杂度和可维护性。基于这个定位我把检查项分成五大类认证与授权问题RBAC 角色过大、ServiceAccount 权限异常、匿名访问是否开启工作负载安全容器是否以 root 运行、特权模式、hostNetwork、hostPID 是否被滥用网络安全配置NetworkPolicy 是否缺失、Service 暴露方式是否合理数据保护配置Secret 是否明文存储、是否被普通账号可读资源与稳定性风险缺少 resource limit、缺少 liveness/readiness 探针等每类检查项内部再拆成多个子检查函数每个函数接收资源对象返回发现的问题列表。这个架构的好处是后续加规则非常方便不需要动框架只需要新增一个函数然后在注册表里登记一下。1.2 为什么选择 Python 而不是 Go说实话用 Go 写 K8s 的工具是“正统做法”client-go 性能好、编译成二进制方便分发。但我选 Python 有三个很实际的理由第一Python 生态里的 kubernetes client 库是官方维护的方法命名和 kubectl 基本一致熟悉 kubectl 的人上手几乎没有成本。第二规则引擎用 Python 表达非常自然比如判断一个 Deployment 是否配置了 resource limit直接用container.resources.limits是否存在就可以代码极其直观。第三也是最现实的这类巡检脚本的运行频率不会太高一天一次甚至一周一次性能根本不是瓶颈。用 Python 可以把开发时间压缩到极短周末两天基本就能完成第一版。如果你是个人开发者或者小团队维护这个时间成本优势是很实用的。当然扫描器占用的资源和被扫描集群的大小也有关系。单个集群几千个 pod 以内Python 完全够用。我之前在大概 2000 个 pod 的集群上实测全量扫描一遍大概要 2~3 分钟对于一个离线巡检任务来说完全可接受。2. 核心依赖与采集层搭好数据管道扫描器能不能得出有意义的结果前提是能拿到准确、完整的集群数据。所以采集层是整个项目的地基我会先讲清楚怎么搭。2.1 环境准备与认证方式项目依赖非常少只有两个核心包pip install kubernetes pyyaml这里kubernetes是官方 Python SDKpyyaml用来解析 YAML 配置。如果后续要做更复杂的分析可以再加pandas但我实际用下来第一版完全没必要纯 Python 的字典操作就够了。认证我推荐直接用本地 kubeconfig和 kubectl 保持一致。这样你在一台能操作集群的机器上跑脚本不需要额外处理 token 或证书from kubernetes import client, config def load_cluster(): try: config.load_incluster_config() except Exception: config.load_kube_config() return client.CoreV1Api(), client.AppsV1Api(), client.RbacAuthorizationV1Api()注意这里的异常处理顺序。如果你打算让这个脚本既能放在集群里以 Pod 方式运行又能本地执行就用这种先后顺序先尝试load_incluster_config失败再load_kube_config。用 Pod 方式跑的时候需要给 ServiceAccount 绑定合适的权限这个后面会讲。我实际开发时还加了一个多集群支持。思路很简单接收一个--context参数然后用load_kube_config(contextargs.context)加载指定集群的配置。如果你维护多个集群这个功能几乎是必须的。2.2 资源数据采集用 SDK 代替 kubectl 命令采集层要拿到的核心资源包括NamespacePodDeployment / StatefulSet / DaemonSetServiceNetworkPolicyRole / ClusterRole / RoleBinding / ClusterRoleBindingSecret注意只拿元数据不拿 value这里有一个很重要的设计决策我刻意不去拉取 Secret 的具体值。因为巡检配置错误不需要知道明文只需要知道哪些 Secret 被哪些主体可读、是否用stringData明文写入了配置。拉取明文会增加脚本本身的安全风险——你不希望一个安全工具反过来变成泄露入口。采集代码并不复杂以 Deployment 为例def collect_deployments(apps_v1): deps apps_v1.list_deployment_for_all_namespaces() return deps.itemsPod 的采集需要注意不要直接拿全部字段很多字段用不到。我习惯是把 pod 转换为 Python dict 后再精简只保留分析需要的字段这样内存占用会小很多而且在后面做判断时直观。下面是一个精简方法def extract_pod_info(pod): containers [] for c in pod.spec.containers: containers.append({ name: c.name, securityContext: c.security_context, resources: c.resources, image: c.image, imagePullPolicy: c.image_pull_policy, ports: [p.container_port for p in c.ports] if c.ports else [], }) return { name: pod.metadata.name, namespace: pod.metadata.namespace, hostNetwork: pod.spec.host_network, hostPID: pod.spec.host_pid, containers: containers, nodeName: pod.spec.node_name, labels: pod.metadata.labels or {} }采集完的数据会缓存到一个全局 dict 里所有检查函数共享这批数据。这样做有个额外好处如果以后要支持多集群聚合分析这个结构可以直接复用。2.3 配置检查框架实现每个检查函数都遵循统一的签名这样后续写新规则时不需要理解整体框架。我定义了一个规则注册表checks [] def register_check(category, severity, description, handler): checks.append({ category: category, severity: severity, description: description, handler: handler }) def run_checks(collector): findings [] for check in checks: try: result check[handler](collector) for item in result: findings.append({ category: check[category], severity: check[severity], rule: check[description], resource: item.get(resource, ), namespace: item.get(namespace, ), message: item.get(message, ), suggestion: item.get(suggestion, ) }) except Exception as e: findings.append({ category: framework, severity: medium, rule: fcheck execution error: {check[description]}, message: str(e) }) return findings这里每个检查函数返回的是一个 list即使一项都没发现也返回空 list 而不是 None。这个约定很重要可以避免上层做判空处理。另外每个检查函数内部都有 try-except单个规则挂了不会影响整体运行对巡检类工具来说是刚需。实际跑的时候你会发现后端 API 偶尔会超时某个资源反序列化异常都是常态容错比什么都重要。3. 检查项的实现从基本配置到深度风险这一节列几个我当时实现的核心规则既有比较基础的也有相对隐晦的配置点。每一条我都会说明判断逻辑和代码。3.1 高危RBAC 权限过大RBAC 权限问题是 K8s 安全里最经典也是最容易被忽视的一类。尤其是ClusterRoleBinding绑定了cluster-admin或者某个 Role 对secrets有get/list权限影响面都很大。我用两种方式检查第一种是直接用 SDK 列出所有 ClusterRole 和 ClusterRoleBinding统计哪些主体绑定了 admin 类角色def check_admin_binding(rbac_v1): findings [] crbs rbac_v1.list_cluster_role_binding().items for crb in crbs: if not crb.role_ref: continue role_name crb.role_ref.name if role_name in (cluster-admin, admin, root) or admin in role_name.lower(): for subject in crb.subjects or []: if subject.kind User or subject.kind Group: findings.append({ resource: fClusterRoleBinding/{crb.metadata.name}, namespace: crb.metadata.namespace or all, message: fsubject {subject.name} binds to role {role_name}, suggestion: 确认该账户是否真的需要集群管理员权限尽量缩小为 namespace 级 Role }) return findings第二种是更动态一点的分析方式利用reviewAPI 来模拟某个主体对某类资源的权限。这个比较重量级但结果非常精确。我当时只在排查特定应用出现 403 时才用这个功能日常巡检用第一种静态分析就够了。注意ClusterRole本身带admin关键字并不一定是坏事比如一些日志组件确实需要大范围读取权限。所以规则里把级别设成 high 而不是 critical留给使用者人工复核的空间。3.2 高危Pod 以内 root 或特权模式运行容器以 root 运行是最常见的高危项之一。K8s 的默认行为是只要镜像里没有显式指定用户容器内进程就会以 root 身份跑严格来说是镜像 USER 指令决定但很多基础镜像默认就是 root。如果容器被攻破攻击者直接就是 root对节点的影响就大了。这项检查实现起来很直接def check_root_user(collector): findings [] for pod in collector.pods: for container in pod[containers]: sc container.get(securityContext) or {} run_as_non_root sc.get(runAsNonRoot) if run_as_non_root is not True: # 还需要看镜像本身是否明确指定了 USER findings.append({ resource: fPod/{pod[name]}/container/{container[name]}, namespace: pod[namespace], message: container may run as root (runAsNonRoot not set to true), suggestion: 在 securityContext 中设置 runAsNonRoot: true并确保镜像 USER 指定非 root }) return findings这里有个坑runAsNonRoot: true只在镜像中指定了非 root 用户时才能真正生效如果镜像本身还是 USER rootK8s 会直接拒绝启动 Pod。所以准确的做法是先判断runAsNonRoot如果没设置就看镜像里的USER——这个信息 SDK 拿不到只能通过正则去分析镜像 Dockerfile 或者直接用镜像仓库的 metadata。我当时第一版只做了前者已经能筛出大部分问题。特权模式privileged: true的判断类似但严重程度更高因为这是容器逃逸最常见的前置条件。另外还要判断hostNetwork、hostPID是否打开这类 pod 如果被入侵攻击面会放大到整个节点。3.3 中危缺少资源配额和 LimitRange从“漏洞扫描器”这个视角看资源配额好像不太像漏洞但实际生产事故里很多 P0 都是因为没有限制某个 pod 的 CPU 或内存导致节点资源耗尽。它属于“配置错误”里非常典型的一类。检查项主要有两个层次第一层检查 namespace 是否配置了 ResourceQuota 和 LimitRangedef check_resource_quota(collector): findings [] quota_namespaces {q.metadata.namespace for q in collector.quotas} limit_range_namespaces {lr.metadata.namespace for lr in collector.limit_ranges} for ns in collector.namespaces: if ns not in quota_namespaces and ns not in limit_range_namespaces: findings.append({ resource: fNamespace/{ns}, namespace: ns, message: namespace has no ResourceQuota or LimitRange, suggestion: 为命名空间创建 ResourceQuota避免单个应用耗尽集群资源 }) return findings第二层检查 deployment 里的容器是否配置了resources.limits。注意这里我会排除几个系统命名空间比如 kube-system 里很多组件kube-proxy、metrics-server 等确实可以不设 limits强设反而有问题。这个排除列表可以做成配置文件用户在跑不同集群时按需调整。3.4 中危NetworkPolicy 缺失与 Service 暴露默认情况下 K8s 的网络是“全互通”的对于一个多租户或环境隔离严格的集群来说这本身就是巨大的配置风险。我没法靠脚本自动判断你的业务是否需要隔离但我可以检查“你有没有定义 NetworkPolicy”以及“依赖你的服务的命名空间是不是来自可疑来源”。NetworkPolicy 的检查很简单统计每个 namespace 的 NetworkPolicy 数量如果为 0则提示该 namespace 没有网络隔离策略。但在生产环境要注意有些网络插件比如 Flannel 的某些版本不支持 NetworkPolicy这时候检查出来也没用。所以这个规则我在报告里统一标注为 medium建议人工根据网络插件决定是否处理。Service 暴露的检查稍微复杂一些。我重点检测type: NodePort或type: LoadBalancer的 Service如果它的 targetPort 对应的是数据库类端口5432、3306、6379 等就提示风险。这里其实没有绝对的对错测试环境经常需要这类暴露所以报告里会附带 namespace 和 service 信息方便使用者判断。3.5 中危Secret 明文与权限扩散Secret 这个点容易起争议。先说结论Secret 默认就是以 base64 编码存储的不是加密。这个结论决定了检查方向。我做了两个判断第一个是检测 Secret 是否通过stringData创建如果存在说明有人用明文方式提交过 Secret 的 YAML这个过程可能把明文泄露到 Git 或其他配置管理系统里def check_secret_stringdata(collector): findings [] # collector.secrets 里保存的是精简后的 secret 对象 for sec in collector.secrets: if sec.get(stringData): findings.append({ resource: fSecret/{sec[name]}, namespace: sec[namespace], message: secret uses stringData, plaintext may have been committed, suggestion: 改用 external secrets 或 sealed secrets 此类工具 }) return findings第二个是权限扩散检查列出所有 Role/ClusterRole过滤出对secrets资源有get/list/watch权限的角色再找到哪些 ServiceAccount 绑定了这些角色。如果是一个 Web 应用的服务账户能读所有 Secret说明权限过大需要收紧。这个检查是 3.1 的一个变体但更细化对企业合规审计很有价值。4. 报告输出与常见问题排查检查项写完之后输出层直接影响工具是否真的有人愿意用。我之前见过太多工具检查结果做成一个巨大的 JSON 或 HTML 表格几十列字段最后根本没人看。4.1 报告输出默认控制台摘要 可选 JSON/CSV第一版我只做控制台输出后来发现多人协作时需要分享结果就又加了 JSON 和 CSV 两种导出方式。控制台输出做一个简单的分级摘要High: 2 | Medium: 7 | Low: 4 | Passed: 0 [High] ClusterRoleBinding/test:admin binds to role cluster-admin - namespace: all, suggest: 请确认是否必要这个摘要格式要克制不需要高亮颜色方便直接留存和后续处理。CSV 导出用来给非技术同事看字段包括分类、级别、资源名、消息、建议这五列Excel 打开直接筛选就行。代码层面上 ctrlc 的处理也要留意。巡检工具很容易有人中途想终止但半途中断可能导致报告残缺让人误以为“查过了没事”。我加了一个atexit钩子强制退出时也尝试输出已有的 findings——至少保留了这次扫描已经发现的结论。4.2 常见问题与排查技巧实录写这个工具的过程中我确实踩了不少坑记录几个比较典型的。第一个坑版本兼容性。Kubernetes Python Client 的版本和集群 API 版本必须匹配否则某些字段解析会出现问题。我用 1.27 集群配 18.x client 是没问题的但如果你的集群是 1.24 甚至更老建议先把 client 降到对应版本否则部分新字段拿不到检查结果会失真。**第二个坑SDK 返回字段可能为 None。**K8s 资源对象的很多可选字段默认是 None比如securityContext、resources.limits新手经常直接container.resources.limits[cpu]然后就TypeError了。在每个检查函数里都要加容错我在框架层做了统一 try-except但最好在函数内部就防御。**第三个坑权限不匹配导致误报。**使用load_kube_config时当前用户只对部分 namespace 有权限那么list_pod_for_all_namespaces()会直接抛 403导致整个脚本中断。我后来改成如果 403 就降级为只扫描有权限的 namespace并生成一条 “scan throttled” 的提示避免用户看到空报告误以为集群很安全。**第四个坑Secret 数量多导致扫描缓慢。**全量拉取 Secret 时如果集群里有几千个 Secret内存占用会明显上升。解决方法是分页拉取SDK 的list_*方法支持limit和_continue参数。系列工具一般不会出大问题但一旦遇到大集群这就是刚需。第五个坑不要只依赖 SDK 的默认字段。container.ports在没有定义时返回 None但有些镜像跑起来也会监听端口所以不能以“没有定义端口”来判断网络暴露。类似的逻辑都需要结合业务去判断扫描器只能做辅助。4.3 与 Kubectl 插件结合从脚本到日常工具整个工具开发完成后我还做了一个小优化把它封装成 kubectl 插件使用方式变成kubectl config-scan --context prod-cluster --output json scan.json原理非常简单就是写一个名为kubectl-config_scan的脚本放到 PATH 里内部再调用 Python 入口。这样可以充分利用 kubectl 的上下文切换逻辑不用在 Python 里管理多集群配置。习惯用 kubectl 的同事接受度会高很多。这个插件化思路适合任何 Python 写的 K8s 小工具成本极低但让工具在团队里推广起来容易很多。5. 后续可以怎么扩展我在这个工具上后续准备做的方向有三个可以参考。第一个是把检测规则做成 YAML 配置这样不用改代码就能调整阈值比如 namespace 排除列表、风险级别、端口黑名单等。这个对非 Python 背景的运维同学很友好也更接近企业里边审计边迭代的需求。第二个是增加基线差异对比。这次扫描发现的问题记录下来生成 baseline下次扫描只报告新增问题。这一点对“持续追踪整改进度”的场景价值非常大否则每次都是一堆同样的问题久了就没人看了。第三个是接入事件采集。配置错误只是静态快照K8s 里真正的问题往往出现在运行中比如 Pod 反复重启、探针失效等。如果扫描器能把事件流一起采集下来再结合规则分析就能发现更多动态风险。不过我自己的经验是第一版不要想太多。先把配置检查跑起来让团队形成习惯再逐步加功能。一个没人跑的工具功能再多也没意义。最后再分享一个我实际使用中的体会安全巡检类工具最大的价值不是发现多少高危项而是让团队养成一种“每次变更后都自动检查一遍”的习惯。把它接进 CI/CD 里的方式很简单扫描结果里只要 high 以上问题数大于 0 就构建失败。不用一开始就做太重先从每天定时跑一次开始等大家发现报告确实能指出问题自然会有人来推动整改。