Skip to content

blacklanternsecurity/blasthttp

Repository files navigation

blasthttp

License: GPL v3 Rust 2024 Crates.io Python 3.10+ PyPI version Rust Tests Python Tests

Offensive-first HTTP library written in Rust with Python bindings. Built for BBOT.

Installation

# Python
pip install blasthttp

# Rust
cargo add blasthttp

Key Advantages

  • Batch connection reuse — connections are pooled and reused within a batch, dramatically reducing overhead when scanning many URLs on the same hosts
  • Rust performance — async I/O, zero-copy where possible, and native concurrency give significant speed improvements over pure Python HTTP clients
  • SSL cert info on every request — extracts CN, SANs, emails, issuer, validity dates, and fingerprint during the TLS handshake that's already happening, eliminating the need for a separate connection to gather this information
  • All TLS ciphers available by default — custom-compiled OpenSSL 3.3.2 with legacy provider baked in (RC4, 3DES, export ciphers, SSLv3) so you can connect to anything
  • No cert validation by default — offensive-first: connects to self-signed, expired, and misconfigured TLS without extra config
  • HTTP/2 support — automatic via ALPN negotiation, falls back to HTTP/1.1
  • Low-level primitivesRawConnection for byte-level TCP/TLS I/O (bypasses HTTP framing entirely) and blasthttp.h2 for manual H2 frame construction (includes a permissive HPACK encoder that emits bytes a strict encoder refuses, the building block for H2 smuggling / CRLF-injection tooling)
  • Response hashing built-in — MD5, SHA256, and MurmurHash3 computed in Rust for both body and headers, ready for fingerprinting

CLI Usage

# Single request
blasthttp https://example.com

# POST with headers and body
blasthttp https://example.com -X POST -H "Content-Type: application/json" -d '{"key":"value"}'

# Batch mode — read URLs from a file, 100 concurrent
blasthttp -l urls.txt -c 100

# Follow redirects
blasthttp https://example.com -L

# Through a proxy
blasthttp https://example.com -x http://127.0.0.1:8080

# ...but connect directly to certain hosts (host, *.suffix, IP, or CIDR)
blasthttp https://example.com -x http://127.0.0.1:8080 --no-proxy '*.internal.corp' --no-proxy 10.0.0.0/8

# Force specific TLS versions/ciphers
blasthttp https://legacy-server.com --min-tls 1.0 --ciphers "RC4-SHA"

# Verbose output (pretty JSON + debug info, -vv includes body)
blasthttp https://example.com -v

Output is JSON (one object per response), including status, headers, redirect chain with per-hop peer IP, TLS cert info, content hashes, parsed cookies, and the actual TCP peer IP for the final hop:

{
  "url": "https://example.com/",
  "status": 200,
  "request_url": "https://example.com",
  "request_method": "GET",
  "headers": [["content-type", "text/html"], ...],
  "raw_headers": "content-type: text/html\r\n...",
  "body": "<!doctype html>...",
  "cookies": {"session": "abc123"},
  "elapsed_ms": 120,
  "peer_ip": "93.184.215.14",
  "redirect_chain": [
    {"url": "http://example.com/", "status": 301, "peer_ip": "93.184.215.14"}
  ],
  "cert_info": {
    "common_name": "example.com",
    "sans": ["example.com", "www.example.com"],
    "emails": [],
    "issuer": "DigiCert Global G2",
    "not_before": "2024-01-15T00:00:00Z",
    "not_after": "2025-02-15T23:59:59Z",
    "fingerprint_sha256": "a0b1c2..."
  },
  "hash": {
    "body_md5": "...",
    "body_mmh3": 1234567,
    "body_sha256": "...",
    "header_md5": "...",
    "header_mmh3": -987654,
    "header_sha256": "..."
  }
}

Options

Flag Description Default
URL Target URL (omit when using -l)
-X, --method HTTP method GET
-H, --header Custom header (repeatable)
-d, --data Request body
-l, --list File of URLs for batch mode
-c, --concurrency Max concurrent requests (batch) 50
--rate-limit Requests per second (batch mode) unlimited
-L, --follow-redirects Follow redirects off
--max-redirects Max redirect hops 10
-t, --timeout Request timeout (seconds) 10
--max-body-size Max response body (bytes) 10 MB
--verify Enable TLS cert validation off
-x, --proxy HTTP/SOCKS proxy URL
--no-proxy Host that bypasses the proxy: host, *.suffix, IP, CIDR, or * (repeatable)
--ciphers OpenSSL cipher string all
--min-tls Minimum TLS version (1.0–1.3)
--max-tls Maximum TLS version (1.0–1.3)
-v, --verbose Verbose output (-vv includes body)

Python API

All request methods are async — they return native Python coroutines via pyo3-async-runtimes.

import asyncio
import blasthttp

async def main():
    client = blasthttp.BlastHTTP()

    # Single request — Response is httpx-style
    r = await client.request("https://example.com")
    print(r.status_code, r.is_success)            # 200, True
    print(r.headers["Content-Type"])              # case-insensitive
    print(r.peer_ip)                              # actual TCP peer IP
    print(r.text[:80])                            # UTF-8 body (lazy)
    print(r.hash.body_md5)                        # md5/sha256/mmh3 (lazy)
    print(r.cookies)                              # Set-Cookie parsed (lazy)
    print(r.request.url, r.request.method)        # original (pre-redirect)
    r.raise_for_status()                          # raises HTTPStatusError on 4xx/5xx

    # Batch requests — full result list at the end
    configs = [
        blasthttp.BatchConfig("https://a.com"),
        blasthttp.BatchConfig("https://b.com", method="POST", body="data"),
    ]
    results = await client.request_batch(configs, concurrency=50)
    for r in results:
        if r.success:
            print(r.url, r.response.status_code, r.response.peer_ip)

    # Streaming batch — process results as they complete
    async for batch in client.request_batch_stream(configs, concurrency=50):
        for r in batch:
            if r.success:
                print(r.url, r.response.status_code)

    # Download to file
    await client.download("https://example.com/file.zip", "/tmp/file.zip")

asyncio.run(main())

Response API

Response follows the httpx convention so it works as a drop-in for httpx-flavored consumers:

Attribute Type Notes
status / status_code int HTTP status code
is_success bool True for 2xx-3xx
url str final URL (after redirects)
text / body str UTF-8 decoded body (lazy)
content / body_bytes bytes raw body
headers Headers case-insensitive, mutable; headers["Content-Type"] works either case
cookies dict[str, str] parsed Set-Cookie (lazy)
raw_headers str canonical Name: Value\r\n… form (lazy)
hash ResponseHash md5 / sha256 / mmh3 of body and headers (lazy)
cert_info CertInfo | None TLS cert details from the handshake
peer_ip str | None actual IP of the final hop's TCP connection; None when proxied
redirect_chain list[RedirectHop] each hop carries its own peer_ip
elapsed_ms / elapsed int / timedelta total request time
request Request original request.url and request.method
debug_log list[str] internal debug/timing lines
json() callable json.loads(self.text)
raise_for_status() callable raises HTTPStatusError on 4xx/5xx

body, raw_headers, cookies, and hash are lazy — first access computes, subsequent accesses are free. Hashing a 10 MB body is real work; batch jobs that filter on status and never look at the body don't pay for it.

headers is a Headers instance (case-insensitive view), not a list of tuples. Iterate with .items() to get (name, value) pairs preserving original case and duplicate names (e.g. multiple Set-Cookie). Headers is registered with collections.abc.MutableMapping, so libraries that special-case mappings (DeepDiff, dataclasses, etc.) recognize it.

raise_for_status() raises blasthttp.HTTPStatusError on 4xx/5xx — import it directly:

from blasthttp import HTTPStatusError
try:
    r = await client.request("https://example.com/404")
    r.raise_for_status()
except HTTPStatusError as e:
    print(e.response.status_code)   # the failing Response is attached

Response, BatchResult, RedirectHop, and Headers all have Python constructors so test code can synthesize them from canned data — blasthttp.Response(url=..., status=200, headers=[...], body=b"...", ...). Useful when building fixtures by hand; the blasthttp.mock submodule below uses them under the hood.

Batch vs. streaming batch

Two ways to issue the same workload — pick whichever matches how you want to consume the results:

Returns Consume with When to use
request_batch(configs, ...) list[BatchResult] after every request finishes results = await ...; for r in results: You want the full set in one shot.
request_batch_stream(configs, ...) async iterator of list[BatchResult] chunks, in completion order async for batch in ...: for r in batch: A slow request shouldn't block faster peers behind it; you want to overlap consumer work with in-flight HTTP I/O; partial results are useful before the slowest finishes.

request_batch_stream yields up to 1000 results per chunk, or whatever has accumulated after ~200ms — the timeout flushes partial chunks so the consumer is never starved when results trickle in slowly. Both functions accept the same (configs, concurrency=50, rate_limit=None) arguments and respect set_rate_limit() identically.

request() parameters

All parameters except url are optional:

Parameter Type Notes
url str Target URL
method str HTTP method (default GET)
headers list[tuple[str, str]] Request headers
body str | bytes Request body (mutually exclusive with files)
files dict Multipart form-data (httpx-style files=; see below)
timeout int Request timeout in seconds
follow_redirects bool Follow redirects
max_redirects int Max redirect hops
verify_certs bool Enable TLS cert validation (default False)
proxy str HTTP/SOCKS proxy URL
no_proxy list[str] Hosts that bypass the proxy
cipher_string str OpenSSL cipher string
min_tls_version str Minimum TLS version ("1.0""1.3")
max_tls_version str Maximum TLS version ("1.0""1.3")
retries int Number of retries on failure
retry_wait_min_ms int Minimum backoff between retries (ms)
retry_wait_max_ms int Maximum backoff between retries (ms)
max_body_size int Max response body bytes to download
request_target str Override the HTTP request-line URI
resolve_ip str Connect to this IP (DNS pinning / curl --resolve)

BatchConfig accepts the same parameters (pass them as constructor kwargs).

Multipart file uploads

Pass files to request() or BatchConfig for httpx-style multipart form-data. body and files are mutually exclusive — passing both raises ValueError.

r = await client.request(
    "https://example.com/upload",
    method="POST",
    files={"document": ("report.pdf", pdf_bytes, "application/pdf")},
)

The Content-Type: multipart/form-data boundary header is set automatically.

Retries

request(), download(), and BatchConfig support automatic retries with exponential backoff:

r = await client.request(
    "https://flaky-api.example.com/data",
    retries=3,
    retry_wait_min_ms=100,
    retry_wait_max_ms=5000,
)

download() parameters

download(url, path, ...) saves a response body to a file. It always follows redirects (up to 10 hops).

Parameter Type Notes
url str Target URL
path str Local file path to write
max_size int Max response body bytes
timeout int Request timeout in seconds
verify_certs bool Enable TLS cert validation
proxy str HTTP/SOCKS proxy URL
no_proxy list[str] Hosts that bypass the proxy
headers list[tuple[str, str]] Request headers
retries int Number of retries on failure

DNS Pinning & Request-Line Control

Use resolve_ip to connect to a specific IP while keeping the original hostname for SNI and Host header — like curl --resolve:

# Connect to 93.184.215.14 but use example.com for TLS SNI and Host header
response = await client.request("https://example.com/", resolve_ip="93.184.215.14")

# Virtual host scanning: override Host header while pinning to target IP
response = await client.request(
    "http://target.com/",
    headers=[("Host", "secret-vhost.target.com")],
    resolve_ip="10.0.0.1",
)

Use request_target to override the request-line URI (e.g. for SSRF testing or request smuggling):

# Send "GET http://internal.server/admin HTTP/1.1" on the wire
response = await client.request(
    "http://proxy.target.com/",
    request_target="http://internal.server/admin",
)

Proxy exclusions (no_proxy)

proxy routes a request through an HTTP or SOCKS5 proxy; no_proxy is a per-request list of hosts that bypass it and connect directly — the NO_PROXY equivalent. It's accepted by request(), download(), raw_connect(), and BatchConfig, and as the repeatable --no-proxy CLI flag.

# Proxy everything except internal hosts and the loopback range.
r = await client.request(
    "https://example.com/",
    proxy="http://127.0.0.1:8080",
    no_proxy=["localhost", "*.internal.corp", "10.0.0.0/8"],
)

Each entry matches case-insensitively as:

  • an exact hostname (elastic.corp);
  • a domain suffix — *.corp, .corp, and corp are equivalent and match the domain itself plus any subdomain;
  • a single IP (127.0.0.1, ::1);
  • a CIDR range (10.0.0.0/8, fd00::/8), matched only against IP hosts;
  • or * to bypass the proxy for every host.

The proxy/no_proxy decision is re-evaluated on every redirect hop, not just the first URL: if a request starts on an excluded (direct) host and is redirected onto a non-excluded host, the later hop goes through the proxy — and vice versa.

no_proxy only does anything when a proxy is set, so passing it without one is rejected up front (the request errors rather than silently ignoring the exclusions).

Global Rate Limiting

Set a client-level rate limit (requests per second) that applies to all request methods — request(), request_batch(), request_batch_stream(), and download():

client = blasthttp.BlastHTTP()
client.set_rate_limit(50)  # 50 requests/sec across all callers

# All of these respect the 50 rps limit:
await client.request("https://example.com")
await client.request_batch(configs, concurrency=100)
await client.download("https://example.com/file", "/tmp/file")

# Disable rate limiting
client.set_rate_limit(0)
# or
client.set_rate_limit(None)

When multiple callers share the same BlastHTTP instance, the rate limiter is global — two concurrent request_batch() (or request_batch_stream()) calls will collectively stay under the limit.

The client-level rate limit takes precedence over the per-call rate_limit parameter on request_batch() / request_batch_stream().

RawConnection handles returned from raw_connect() inherit the client's rate limiter: every send_bytes / read_raw call on that handle also consumes one token, so a caller that pipelines many ops on a single connection can't burst past the limit.

RawConnection — byte-level TCP/TLS I/O

raw_connect() returns a RawConnection handle that exposes the TCP or TLS stream directly. No HTTP parsing, no framing, no re-emission — you write bytes in and read bytes out. Use it to send hand-crafted requests (malformed HTTP/1, raw HTTP/2 frames, non-HTTP protocols) or to peek at raw server responses the way a protocol fuzzer or smuggling detector needs.

import asyncio
import blasthttp

async def main():
    client = blasthttp.BlastHTTP()

    # Open a TLS connection, negotiating h2 via ALPN.
    conn = await client.raw_connect(
        "https://example.com/",
        alpn_protocols=["h2", "http/1.1"],
    )
    print("ALPN negotiated:", conn.negotiated_alpn)
    print("TLS cert CN:", conn.cert_info.common_name if conn.cert_info else None)

    # Send arbitrary bytes — no Content-Length / framing added for you.
    await conn.send_bytes(
        b"GET / HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n\r\n"
    )

    # Read with a per-call deadline. Empty bytes = timeout or peer close.
    data = await conn.read_raw(max_bytes=65536, timeout_ms=2000)
    print(data[:200])

    await conn.close()

asyncio.run(main())

raw_connect() takes all the same TLS knobs as request()verify_certs, cipher_string, min_tls_version, max_tls_version, resolve_ip, proxy, no_proxy. alpn_protocols is a list of byte-strings (commonly ["h2", "http/1.1"]) used in the TLS ALPN extension. After the handshake, conn.negotiated_alpn reports which one the server picked (or None if no ALPN was negotiated, including all plain-HTTP connections).

conn.peer_ip returns the actual IP address the OS connected to, same as Response.peer_ip. None when the connection went through a proxy.

HTTP/2 primitives — blasthttp.h2

blasthttp.h2 is a minimal, permissive HTTP/2 toolkit for callers who need to emit custom H2 frames over a RawConnection. The encoder deliberately lets you produce bytes a strict implementation refuses (CRLF in header values, invalid header names, forced Huffman on/off, custom indexing choices) — the exact knobs protocol fuzzers and H2 smuggling detectors depend on.

from blasthttp import h2

# Encode a header block with HPACK.
block = h2.encode_headers([
    h2.Header(":method", "GET"),
    h2.Header(":path", "/"),
    h2.Header(":authority", "example.com"),
    h2.Header(":scheme", "https"),
    # Permissive escape hatch: emit CRLF inside a value for
    # CRLF-injection testing against H2-to-H1 downgrades.
    h2.Header(
        "x-injected", "bogus\r\nX-Smuggled: yes",
        allow_invalid_value=True, huffman_value=False,
    ),
])

# Build a full probe: preface + SETTINGS + HEADERS (+ optional DATA).
probe = h2.build_probe(
    [
        h2.Header(":method", "POST"),
        h2.Header(":path", "/"),
        h2.Header(":authority", "example.com"),
        h2.Header(":scheme", "https"),
        h2.Header("content-length", "5"),
    ],
    body=b"hello",
)

# Or assemble frames individually for finer control.
settings = h2.build_settings_frame()
headers = h2.build_headers_frame(block, stream_id=1, end_stream=False)
data = h2.build_data_frame(b"hello", stream_id=1, end_stream=True)

# Decode response frames with a stateful decoder (tracks HPACK
# dynamic-table state across calls).
dec = h2.Decoder()
response_headers = dec.decode(block)  # -> [(name: bytes, value: bytes), ...]

blasthttp.h2 exposes:

  • Header(name, value, **permissiveness) — a single header-field pair, with flags like allow_invalid_name, allow_invalid_value, huffman_name, huffman_value, indexing, force_static_index, length_bloat_name, length_bloat_value
  • encode_headers(headers) -> bytes — HPACK header-block-fragment
  • Decoder() — stateful HPACK decoder (dynamic-table-aware); decode(block) -> [(name, value), ...]
  • build_settings_frame(settings=None, ack=False), build_headers_frame(block, stream_id, end_stream, end_headers, ...), build_data_frame(data, stream_id, end_stream, ...), build_continuation_frame, build_rst_stream_frame, build_goaway_frame, build_ping_frame, build_priority_frame, build_window_update_frame, build_raw_frame (escape hatch for malformed frames)
  • build_probe(headers, body=None, ...) — one-shot preface + SETTINGS + HEADERS + optional DATA builder with every knob from the underlying Rust ProbeOpts exposed as a keyword argument
  • PREFACE — the H2 connection preface bytes
  • Frame-type + flag constants: FRAME_HEADERS, FRAME_DATA, FRAME_SETTINGS, FRAME_CONTINUATION, FRAME_RST_STREAM, FRAME_GOAWAY, FRAME_PING, FRAME_PRIORITY, FRAME_WINDOW_UPDATE, FLAG_END_HEADERS, FLAG_END_STREAM, FLAG_ACK, FLAG_PADDED, FLAG_PRIORITY

This isn't a full H2 client — no stream state machine, no flow control, no HTTP-level response parsing. It's the bytes in, bytes out. Pair with RawConnection for end-to-end control.

Test fixture mock — blasthttp.mock

blasthttp.mock.BlasthttpMock is a drop-in replacement for BlastHTTP that returns canned responses or invokes registered callbacks instead of hitting the network. Useful for unit tests that exercise code which calls into a BlastHTTP client.

import asyncio
import re
import blasthttp
from blasthttp.mock import BlasthttpMock, MockResponse

async def main():
    mock = BlasthttpMock()

    # Static responses — declarative
    mock.add_response(url="https://api.example.com/users", json={"users": []})
    mock.add_response(url=re.compile(r"/v\d+/health"), text="ok")

    # Programmatic — callback receives a MockRequest, returns a MockResponse
    def echo(req):
        return MockResponse(status_code=201, json={"echoed": req.method})
    mock.add_callback(echo, url="https://api.example.com/echo")

    # Use it like a real client
    r = await mock.request("https://api.example.com/users")
    print(r.json())                    # {"users": []}
    print(isinstance(r, blasthttp.Response))   # True

    # Batch streaming works the same as the real client
    configs = [blasthttp.BatchConfig(f"https://api.example.com/users/{i}") for i in range(5)]
    mock.add_response(url=re.compile(r"/users/\d+"), json={"id": 0})
    async for r in mock.request_batch_stream(configs):
        ...

asyncio.run(main())

Key behaviors:

  • Matchers: url accepts str (exact) or re.Pattern (.search() semantics) or None (any). method is case-insensitive. match_headers={...} and match_json={...} add subset predicates against the request headers / decoded JSON body.
  • FIFO + recycle: handlers are matched in registration order. Once consumed, a handler is moved to a recycle queue so subsequent requests can re-match it — no need to re-register for repeated calls.
  • Callbacks: sync and async (async def) both supported. Callbacks may raise TimeoutException or any other exception to simulate failures. Returning MockResponse is most ergonomic; returning a blasthttp.Response directly also works.
  • Pass-through: pass a real_client and a should_mock_fn(host) -> bool predicate to forward selected requests to a real BlastHTTP instance. Common use case: mock everything except 127.0.0.1 so a fixture HTTP server stays exercised.
real = blasthttp.BlastHTTP()
mock = BlasthttpMock(real_client=real, should_mock_fn=lambda h: h != "127.0.0.1")
mock.add_response(url="https://example.com/", text="mocked")
# https://example.com/ → mocked, http://127.0.0.1/* → real_client

Pytest integration is intentionally left to the consumer — wrap the mock in a fixture in your conftest.py:

import pytest
from blasthttp.mock import BlasthttpMock

@pytest.fixture
def http_mock():
    return BlasthttpMock()

Building

Prerequisites

  • Rust (2024 edition) — install via rustup
  • Python 3.10+ (for Python bindings)
  • Standard C build tools (build-essential / gcc, make, perl)
  • curl or wget (for OpenSSL download)

1. Build custom OpenSSL

blasthttp ships a script that downloads OpenSSL 3.3.2, compiles it with weak cipher support (RC4, DES, 3DES, export ciphers, SSLv3), and installs it to vendor/openssl/install/. This only needs to be run once — the result is cached and reused.

./scripts/build-openssl.sh

This produces a static build (libssl.a, libcrypto.a) with the legacy provider baked in. The binary has no runtime dependency on system OpenSSL. Delete vendor/openssl/install/ to force a rebuild.

2. Build the Rust CLI

cargo build --release

The binary is at target/release/blasthttp. If you skip step 1, the build will fail with a clear error telling you to run the OpenSSL script.

3. Build the Python module

Requires maturin:

pip install maturin
maturin develop --release

This compiles the Rust code with Python bindings enabled and installs the blasthttp package into your current Python environment. You can then import blasthttp from Python.

How it fits together

  • .cargo/config.toml sets OPENSSL_DIR (relative path to vendor/openssl/install/) and OPENSSL_STATIC=1 so the openssl-sys crate links against the custom build statically
  • build.rs runs before compilation and verifies the custom OpenSSL headers exist, failing fast with an actionable error if they don't
  • The [features] python gate means cargo build produces a pure Rust binary, while maturin build activates PyO3 and produces a Python-loadable .so

About

A fast offensive-focused HTTP client written in Rust, with Python bindings

Resources

License

Security policy

Stars

Watchers

Forks

Packages

 
 
 

Contributors