forked from username/flaskpaste
add hypothesis-based fuzzing test suite
18 property-based tests covering: - Content handling (binary, text, unicode) - Paste ID validation and path traversal - Header fuzzing (auth, proxy, XFF) - JSON endpoint fuzzing - Size limit enforcement - Injection detection (SQLi, SSTI, XSS) - Error handling paths
This commit is contained in:
Username
485
documentation/fuzzing-tools.md
Normal file
485
documentation/fuzzing-tools.md
Normal file
@@ -0,0 +1,485 @@
|
||||
# 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
|
||||
|
||||
```bash
|
||||
# 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
|
||||
|
||||
```python
|
||||
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
|
||||
|
||||
```python
|
||||
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
|
||||
|
||||
```python
|
||||
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
|
||||
|
||||
```python
|
||||
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
|
||||
|
||||
```python
|
||||
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
|
||||
|
||||
```python
|
||||
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
|
||||
|
||||
```python
|
||||
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
|
||||
|
||||
```bash
|
||||
# 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
|
||||
|
||||
```python
|
||||
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
|
||||
|
||||
```python
|
||||
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
|
||||
|
||||
```bash
|
||||
# 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
|
||||
|
||||
```bash
|
||||
# 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
|
||||
|
||||
```bash
|
||||
git clone https://gitlab.com/akihe/radamsa.git
|
||||
cd radamsa && make && sudo make install
|
||||
```
|
||||
|
||||
### Usage
|
||||
|
||||
```bash
|
||||
# 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
|
||||
|
||||
```bash
|
||||
pip install atheris
|
||||
```
|
||||
|
||||
### Fuzzing Harness
|
||||
|
||||
```python
|
||||
#!/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
|
||||
|
||||
```bash
|
||||
# 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
|
||||
|
||||
```bash
|
||||
#!/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
|
||||
└──────────────────────────────────────────────────────────────────────────────┘
|
||||
```
|
||||
Reference in New Issue
Block a user