Offensive-first HTTP library written in Rust with Python bindings. Built for BBOT.
# Python
pip install blasthttp
# Rust
cargo add blasthttp- 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 primitives —
RawConnectionfor byte-level TCP/TLS I/O (bypasses HTTP framing entirely) andblasthttp.h2for 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
# 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 -vOutput 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": "..."
}
}| 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) |
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 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 attachedResponse, 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.
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.
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).
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.
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(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 |
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 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, andcorpare 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).
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.
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.
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 likeallow_invalid_name,allow_invalid_value,huffman_name,huffman_value,indexing,force_static_index,length_bloat_name,length_bloat_valueencode_headers(headers) -> bytes— HPACK header-block-fragmentDecoder()— 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 RustProbeOptsexposed as a keyword argumentPREFACE— 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.
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:
urlacceptsstr(exact) orre.Pattern(.search()semantics) orNone(any).methodis case-insensitive.match_headers={...}andmatch_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 raiseTimeoutExceptionor any other exception to simulate failures. ReturningMockResponseis most ergonomic; returning ablasthttp.Responsedirectly also works. - Pass-through: pass a
real_clientand ashould_mock_fn(host) -> boolpredicate to forward selected requests to a realBlastHTTPinstance. Common use case: mock everything except127.0.0.1so 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_clientPytest 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()- Rust (2024 edition) — install via rustup
- Python 3.10+ (for Python bindings)
- Standard C build tools (
build-essential/gcc,make,perl) curlorwget(for OpenSSL download)
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.shThis 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.
cargo build --releaseThe 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.
Requires maturin:
pip install maturin
maturin develop --releaseThis 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.
.cargo/config.tomlsetsOPENSSL_DIR(relative path tovendor/openssl/install/) andOPENSSL_STATIC=1so theopenssl-syscrate links against the custom build staticallybuild.rsruns before compilation and verifies the custom OpenSSL headers exist, failing fast with an actionable error if they don't- The
[features] pythongate meanscargo buildproduces a pure Rust binary, whilematurin buildactivates PyO3 and produces a Python-loadable.so