flaskpaste/documentation/fuzzing-tools.md

Fuzzing Tools Guide

Quick reference for offensive security testing tools used in FlaskPaste fuzzing.

---

Tool Overview

┌─────────────┬──────────────────────────────────────────────────────────────┐
│ Tool        │ Purpose
├─────────────┼──────────────────────────────────────────────────────────────┤
│ httpx       │ Async HTTP client for high-concurrency testing
│ aiohttp     │ Async HTTP for protocol-level attacks
│ hypothesis  │ Property-based fuzzing with smart input generation
│ requests    │ Synchronous HTTP for sequential tests
│ curl        │ Quick manual endpoint testing
│ radamsa     │ Mutation-based fuzzing (binary corruption)
│ atheris     │ Coverage-guided Python fuzzing (LibFuzzer)
│ sqlmap      │ Automated SQL injection detection
│ ffuf        │ Fast web fuzzer for directory/parameter discovery
└─────────────┴──────────────────────────────────────────────────────────────┘

---

Installation

# Create isolated fuzzing environment
python3 -m venv /tmp/fuzz-venv
source /tmp/fuzz-venv/bin/activate

# Core tools
pip install httpx aiohttp hypothesis requests pytest pytest-asyncio

# Optional advanced tools
pip install atheris  # Coverage-guided fuzzing
# pip install python-afl  # AFL integration (requires AFL++)

---

httpx - Async HTTP Client

Modern async HTTP client for high-concurrency testing.

Basic Usage

import httpx

# Sync client
r = httpx.get("http://target/")
print(r.status_code, r.text[:100])

# With custom headers
r = httpx.post("http://target/",
    data="payload",
    headers={"X-Custom": "value"})

Async Concurrent Requests

import asyncio
import httpx

async def fuzz_concurrent(url, count=100):
    """Send concurrent requests to test race conditions."""
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for _ in range(count)]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        success = sum(1 for r in results if hasattr(r, 'status_code') and r.status_code == 200)
        print(f"Success: {success}/{count}")
        return results

# Run
asyncio.run(fuzz_concurrent("http://target/paste-id", 50))

Timeout and Error Handling

import httpx

client = httpx.Client(timeout=5.0)

try:
    r = client.get("http://target/slow-endpoint")
except httpx.TimeoutException:
    print("Timeout - potential DoS")
except httpx.ConnectError:
    print("Connection failed")

---

hypothesis - Property-Based Testing

Generate smart test inputs automatically based on specifications.

Basic Fuzzing

from hypothesis import given, strategies as st, settings

@settings(max_examples=1000)
@given(st.binary(min_size=0, max_size=10000))
def test_paste_content(content):
    """Fuzz paste content with arbitrary bytes."""
    import requests
    r = requests.post("http://target/", data=content,
                      headers={"X-PoW-Token": "x", "X-PoW-Solution": "0"})
    assert r.status_code in (201, 400, 413, 429)

Text Fuzzing

from hypothesis import given, strategies as st

@given(st.text(min_size=0, max_size=1000, alphabet=st.characters()))
def test_unicode_handling(text):
    """Test Unicode handling in paste content."""
    import requests
    r = requests.post("http://target/", data=text.encode('utf-8'),
                      headers={"X-PoW-Token": "x", "X-PoW-Solution": "0"})
    assert r.status_code != 500  # No server errors

@given(st.text(alphabet="/<>\"'%;()&+"))
def test_special_chars_in_id(paste_id):
    """Test special characters in paste ID."""
    import requests
    r = requests.get(f"http://target/{paste_id}")
    assert r.status_code in (200, 400, 404)

JSON Fuzzing

from hypothesis import given, strategies as st

@given(st.dictionaries(
    st.text(min_size=1, max_size=50),
    st.one_of(st.text(), st.integers(), st.floats(), st.none(), st.booleans()),
    min_size=0, max_size=20
))
def test_json_endpoints(data):
    """Fuzz JSON endpoints with arbitrary structures."""
    import requests
    r = requests.post("http://target/pki/ca",
                      json=data,
                      headers={"Content-Type": "application/json"})
    assert r.status_code != 500

# Nested structures
@given(st.recursive(
    st.text() | st.integers(),
    lambda children: st.lists(children) | st.dictionaries(st.text(), children),
    max_leaves=50
))
def test_nested_json(data):
    """Test deeply nested JSON structures."""
    import requests
    try:
        r = requests.post("http://target/pki/issue", json={"data": data})
        assert r.status_code != 500
    except Exception:
        pass  # Connection errors ok

Header Fuzzing

from hypothesis import given, strategies as st

@given(st.dictionaries(st.text(max_size=100), st.text(max_size=500), max_size=20))
def test_arbitrary_headers(headers):
    """Test with arbitrary HTTP headers."""
    import requests

    # Filter dangerous headers
    safe = {k: v for k, v in headers.items()
            if k.lower() not in ('host', 'content-length', 'transfer-encoding')}

    try:
        r = requests.get("http://target/", headers=safe, timeout=5)
    except Exception:
        pass

Running Hypothesis Tests

# Run with pytest
pytest test_fuzz.py -v --hypothesis-show-statistics

# Increase examples
pytest test_fuzz.py --hypothesis-seed=0 -x

# Save failing examples
pytest test_fuzz.py --hypothesis-database=.hypothesis

---

aiohttp - Low-Level Async HTTP

For protocol-level attacks and custom request crafting.

HTTP Smuggling Test

import aiohttp
import asyncio

async def test_http_smuggling():
    """Test for HTTP request smuggling."""
    # CL.TE smuggling attempt
    payload = (
        b"POST / HTTP/1.1\r\n"
        b"Host: target\r\n"
        b"Content-Length: 13\r\n"
        b"Transfer-Encoding: chunked\r\n"
        b"\r\n"
        b"0\r\n"
        b"\r\n"
        b"SMUGGLED"
    )

    reader, writer = await asyncio.open_connection('127.0.0.1', 5000)
    writer.write(payload)
    await writer.drain()

    response = await reader.read(1024)
    print(response.decode(errors='replace'))

    writer.close()
    await writer.wait_closed()

asyncio.run(test_http_smuggling())

Slowloris Attack Test

import asyncio

async def slowloris_test(host, port, connections=50):
    """Test resistance to slowloris attacks."""
    async def slow_connection():
        try:
            reader, writer = await asyncio.open_connection(host, port)
            writer.write(b"GET / HTTP/1.1\r\nHost: target\r\n")
            await writer.drain()

            # Send headers slowly
            for i in range(100):
                writer.write(f"X-Header-{i}: {'A'*100}\r\n".encode())
                await writer.drain()
                await asyncio.sleep(1)
        except Exception as e:
            return str(e)
        return "kept alive"

    tasks = [slow_connection() for _ in range(connections)]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    print(f"Connections maintained: {sum(1 for r in results if r == 'kept alive')}")

asyncio.run(slowloris_test('127.0.0.1', 5000))

---

curl - Quick Manual Testing

Cheatsheet

# Basic requests
curl -v http://target/                    # Verbose GET
curl -X POST http://target/ -d "data"     # POST with data
curl -X PUT http://target/id -d "update"  # PUT request

# Headers
curl -H "X-Custom: value" http://target/
curl -H "Host: evil.com" http://target/   # Host header injection
curl -H "X-Forwarded-For: 127.0.0.1" http://target/

# Response info
curl -s -o /dev/null -w "%{http_code}\n" http://target/
curl -s -o /dev/null -w "%{time_total}\n" http://target/  # Timing

# Binary data
curl -X POST http://target/ --data-binary @file.bin
dd if=/dev/urandom bs=1024 count=100 | curl -X POST http://target/ --data-binary @-

# Follow redirects
curl -L http://target/

# With authentication
curl --cert client.crt --key client.key https://target/

# Parallel requests (race condition)
seq 1 100 | xargs -P50 -I{} curl -s http://target/paste-id

Injection Testing

# SQL injection
curl "http://target/1'%20OR%20'1'='1"
curl "http://target/1%3B%20DROP%20TABLE%20pastes"

# Path traversal
curl "http://target/../../../etc/passwd"
curl "http://target/..%2f..%2f..%2fetc/passwd"

# Command injection (timing)
time curl -X POST http://target/ -d "; sleep 5"

# SSTI
curl -X POST http://target/ -d "{{7*7}}"
curl -X POST http://target/ -d '${7*7}'

---

radamsa - Mutation Fuzzer

Generate corrupted inputs from valid samples.

Installation

git clone https://gitlab.com/akihe/radamsa.git
cd radamsa && make && sudo make install

Usage

# Generate mutations from seed
echo "normal input" > seed.txt
radamsa seed.txt                          # Single mutation
radamsa -n 100 seed.txt > mutations.txt   # 100 mutations

# Fuzz endpoint
for i in $(seq 1 1000); do
    radamsa seed.txt | curl -s -X POST http://target/ \
        --data-binary @- -o /dev/null -w "%{http_code}\n"
done | sort | uniq -c

# JSON mutations
echo '{"key": "value"}' > seed.json
radamsa seed.json | curl -s -X POST http://target/api \
    -H "Content-Type: application/json" --data-binary @-

---

atheris - Coverage-Guided Fuzzing

LibFuzzer-based Python fuzzing with code coverage feedback.

Installation

pip install atheris

Fuzzing Harness

#!/usr/bin/env python3
import atheris
import sys

with atheris.instrument_imports():
    # Import after instrumentation
    from app import create_app

app = create_app()

def TestOneInput(data):
    """Fuzz test function called by atheris."""
    fdp = atheris.FuzzedDataProvider(data)

    with app.test_client() as client:
        # Generate fuzzed request
        method = fdp.PickValueInList(['GET', 'POST', 'PUT', 'DELETE'])
        path = '/' + fdp.ConsumeUnicodeNoSurrogates(100)
        body = fdp.ConsumeBytes(fdp.remaining_bytes())

        try:
            if method == 'GET':
                client.get(path)
            elif method == 'POST':
                client.post(path, data=body)
            elif method == 'PUT':
                client.put(path, data=body)
            elif method == 'DELETE':
                client.delete(path)
        except Exception:
            pass  # Catch app exceptions

if __name__ == "__main__":
    atheris.Setup(sys.argv, TestOneInput)
    atheris.Fuzz()

Running

# Basic run
python fuzz_harness.py

# With options
python fuzz_harness.py -max_len=10000 -runs=100000 -jobs=4

# With corpus
mkdir corpus
python fuzz_harness.py corpus/ -max_len=10000

---

Workflow: Complete Fuzzing Session

#!/bin/bash
# Complete fuzzing workflow

export TARGET="http://127.0.0.1:5000"
export FUZZ_DIR="/tmp/flaskpaste-fuzz"

# 1. Setup
mkdir -p "$FUZZ_DIR"/{results,crashes,corpus}
source /tmp/fuzz-venv/bin/activate

# 2. Reconnaissance
echo "=== Recon ===" | tee "$FUZZ_DIR/results/recon.txt"
for path in / /health /challenge /pastes /pki /audit; do
    curl -s -o /dev/null -w "%s %s\n" "$TARGET$path" "$path" | tee -a "$FUZZ_DIR/results/recon.txt"
done

# 3. Injection tests
echo "=== Injection ===" | tee "$FUZZ_DIR/results/injection.txt"
python3 -c "
import requests
for payload in [\"' OR 1=1--\", \"{{7*7}}\", \"; sleep 5\"]:
    r = requests.post('$TARGET/', data=payload,
                      headers={'X-PoW-Token':'x','X-PoW-Solution':'0'})
    print(f'{r.status_code} {payload[:20]}')
" | tee -a "$FUZZ_DIR/results/injection.txt"

# 4. Hypothesis fuzzing
pytest tests/fuzz/ -v --hypothesis-show-statistics 2>&1 | tee "$FUZZ_DIR/results/hypothesis.txt"

# 5. Generate report
echo "=== Results ==="
cat "$FUZZ_DIR/results/"*.txt

---

Quick Reference

┌──────────────────────────────────────────────────────────────────────────────┐
│ COMMAND CHEATSHEET
├──────────────────────────────────────────────────────────────────────────────┤
│ curl -s -o /dev/null -w "%{http_code}" URL    Status code only
│ curl -w "%{time_total}" URL                   Response timing
│ seq 1 100 | xargs -P50 curl -s URL            Parallel requests
│
│ pytest test.py --hypothesis-show-statistics   Run with stats
│ radamsa seed.txt | curl --data-binary @-      Mutation fuzzing
│
│ python fuzz.py -runs=10000 -jobs=4            Atheris parallel
│ python fuzz.py -max_len=10000                 Max input size
└──────────────────────────────────────────────────────────────────────────────┘