Spaces:

techprotrade
/

annator-command-center

Running

App Files Files Community

annator-command-center / tests /fuzzing /README.md

techprotrade

Deploy ATOM FastAPI command center runtime (part 8)

aef804e verified 10 days ago

preview code

Raw

History Blame Contribute Delete

24.2 kB

API Fuzzing Infrastructure

Coverage-guided crash discovery for FastAPI endpoints using Atheris

Overview

API fuzzing is a automated testing technique that generates malformed, unexpected, or random inputs to discover crashes, vulnerabilities, and edge cases in API endpoints. Atom uses Atheris (Google's coverage-guided fuzzer for Python) to perform intelligent fuzzing on critical API endpoints.

Why Atheris?

Coverage-guided: Uses code coverage to intelligently mutate inputs and explore new code paths
Python-native: Direct integration with pytest and Python codebase
Crash discovery: Finds memory corruption, assertion failures, and unhandled exceptions
Performance: 10,000+ iterations per endpoint in ~5-10 minutes
CI/CD integration: Automated weekly campaigns with crash deduplication and bug filing

Goal

Discover crashes in parsing, validation, and business logic code before attackers do. Fuzzing complements unit tests, integration tests, and property-based testing by exploring input spaces that human-written tests miss.

Quick Start

Install Dependencies

# Install Atheris (fuzzing engine)
pip install atheris

# Or install with testing dependencies
cd backend
pip install -r requirements-testing.txt

Run Single Fuzz Test

# Quick verification (100 iterations)
FUZZ_ITERATIONS=100 pytest tests/fuzzing/test_auth_api_fuzzing.py::test_login_endpoint_fuzzing -v -m fuzzing

# Full fuzzing campaign (10,000 iterations)
FUZZ_ITERATIONS=10000 pytest tests/fuzzing/test_auth_api_fuzzing.py::test_login_endpoint_fuzzing -v -m fuzzing

Run All Fuzz Tests

# Run all fuzzing tests (quick verification)
FUZZ_ITERATIONS=100 pytest tests/fuzzing/ -v -m fuzzing

# Run all fuzzing tests (full campaign)
FUZZ_ITERATIONS=10000 pytest tests/fuzzing/ -v -m fuzzing

Run Fuzzing Campaign

# Run all campaigns for 60 seconds (quick test)
python3 tests/fuzzing/scripts/run_fuzzing_campaigns.py --duration 60

# Run all campaigns for 1 hour (full campaign)
python3 tests/fuzzing/scripts/run_fuzzing_campaigns.py --duration 3600

# Run single campaign
python3 tests/fuzzing/scripts/run_fuzzing_campaigns.py --duration 3600 --campaign "POST /api/auth/login"

Fuzzing Infrastructure

FuzzingOrchestrator

Location: tests/fuzzing/campaigns/fuzzing_orchestrator.py

Service for managing fuzzing campaigns:

from tests.fuzzing.campaigns.fuzzing_orchestrator import FuzzingOrchestrator

orchestrator = FuzzingOrchestrator(github_token="...", github_repository="owner/repo")

# Start campaign
result = orchestrator.start_campaign(
    target_endpoint="POST /api/auth/login",
    test_file="tests/fuzzing/test_auth_api_fuzzing.py::test_login_endpoint_fuzzing",
    duration_seconds=3600
)

# Monitor campaign
stats = orchestrator.monitor_campaign(result["campaign_id"])

# Stop campaign
orchestrator.stop_campaign(result["campaign_id"])

Features:

Campaign lifecycle management (start, stop, monitor)
Subprocess management with graceful shutdown (SIGTERM → SIGKILL)
Crash deduplication integration
Automated bug filing via BugFilingService
Campaign statistics tracking (executions, crashes, coverage)

CrashDeduplicator

Location: tests/fuzzing/campaigns/crash_deduplicator.py

Crash deduplication using SHA256 hashing of error signatures:

from tests.fuzzing.campaigns.crash_deduplicator import CrashDeduplicator

deduplicator = CrashDeduplicator()
crashes_by_signature = deduplicator.deduplicate_crashes(crash_dir)

# Get unique crashes
unique_crashes = deduplicator.get_unique_crashes(crashes_by_signature)

# Get signature summary
summary = deduplicator.get_signature_summary(crashes_by_signature)

Features:

SHA256-based crash deduplication
Stack trace extraction from crash logs
Normalized error signatures (line numbers removed for stable hashing)
Unique crash identification for bug filing

BugFilingService

Location: tests/bug_discovery/bug_filing_service.py (from Phase 236)

Automated GitHub issue filing for discovered crashes:

from tests.bug_discovery.bug_filing_service import BugFilingService

bug_service = BugFilingService(github_token, github_repository)

bug_result = bug_service.file_bug(
    test_name="fuzzing_POST_api_auth_login",
    error_message="Crash in POST /api/auth/login: AssertionError...",
    metadata={
        "test_type": "fuzzing",
        "target_endpoint": "POST /api/auth/login",
        "crash_input": "deadbeef...",
        "crash_log": "Traceback...",
        "signature_hash": "abc123..."
    }
)

Features:

Duplicate bug detection (via existing issues)
Automatic labeling (bug-discovery, fuzzing)
Crash metadata attachment (input hex, stack trace, signature)
CI/CD integration (automatic filing on crashes)

Corpus Management

Location: tests/fuzzing/campaigns/corpus/

Corpus files contain interesting inputs discovered during fuzzing:

# Corpus directory structure
tests/fuzzing/campaigns/corpus/
├── auth/
│   ├── login_valid.input
│   ├── login_sql_injection.input
│   └── login_null_bytes.input
├── agent/
│   ├── agent_run_malformed.input
│   └── agent_run_huge_payload.input
└── workflow/
    ├── workflow_create_cyclic.input
    └── workflow_create_xss.input

Purpose:

Re-seeding: Use corpus to speed up future campaigns
Coverage: Corpus inputs explore interesting code paths
Reproduction: Each input is reproducible crash trigger

Usage:

# Re-seed campaign with corpus
FUZZ_CORPUS_DIR="tests/fuzzing/campaigns/corpus/auth" \
FUZZ_ITERATIONS=10000 \
pytest tests/fuzzing/test_auth_api_fuzzing.py -v -m fuzzing

Fuzzing Test Structure

Test File Pattern

tests/fuzzing/test_<endpoint>_fuzzing.py

Examples:

test_auth_api_fuzzing.py - Authentication endpoints
test_agent_api_fuzzing.py - Agent execution endpoints
test_canvas_presentation_fuzzing.py - Canvas presentation endpoints
test_workflow_api_fuzzing.py - Workflow management endpoints
test_skill_installation_fuzzing.py - Skill import endpoints

Pytest Markers

All fuzzing tests use three markers:

@pytest.mark.fuzzing      # Identifies fuzzing tests
@pytest.mark.slow         # Indicates long-running tests
@pytest.mark.timeout(300) # 5-minute timeout per test

Usage:

# Run only fuzzing tests
pytest tests/fuzzing/ -v -m fuzzing

# Run fuzzing + chaos tests
pytest tests/ -v -m "fuzzing or chaos"

# Exclude fuzzing from quick test runs
pytest tests/ -v -m "not fuzzing"

Fixtures

Fuzzing tests reuse fixtures from tests/e2e_ui/fixtures/:

from tests.e2e_ui.fixtures.database_fixtures import db_session
from tests.e2e_ui.fixtures.auth_fixtures import authenticated_user, test_user

Benefits:

No duplication (reuse existing fixtures)
Worker-based database isolation
API-first auth (10-100x faster than UI login)
Transaction rollback for cleanup

Example:

@pytest.mark.fuzzing
@pytest.mark.slow
@pytest.mark.timeout(300)
def test_login_endpoint_fuzzing(db_session, authenticated_user):
    """Fuzz POST /api/auth/login endpoint."""
    # db_session: Isolated database session
    # authenticated_user: Pre-created user with valid credentials
    pass

Crash Artifacts

Crash artifacts are saved to FUZZ_CRASH_DIR environment variable:

export FUZZ_CRASH_DIR=/tmp/fuzz_crashes

# Crashes saved to:
# - /tmp/fuzz_crashes/login_*.input (crashing input)
# - /tmp/fuzz_crashes/login_*.log (stack trace)

Artifact Format:

*.input - Binary input that triggered crash
*.log - Python stack trace and error message

Usage in CI:

- name: Upload crash artifacts
  uses: actions/upload-artifact@v3
  with:
    name: fuzzing-crashes
    path: backend/tests/fuzzing/campaigns/crashes/
    retention-days: 90

Coverage

Authentication Endpoints (FUZZ-03)

Test File: test_auth_api_fuzzing.py

✅ POST /api/auth/login - Login with email/password
✅ POST /api/auth/signup - User registration
✅ POST /api/auth/mobile/login - Mobile login with device token

Test File: test_jwt_validation_fuzzing.py

✅ GET /api/agents - Protected endpoint (JWT validation)
✅ JWT token parsing (header, payload, signature)
✅ JWT expiry validation (timestamp handling)
✅ JWT signature validation (crypto verification)
✅ Authorization header format (Bearer prefix, spacing)

Test File: test_password_reset_fuzzing.py

✅ POST /api/auth/reset-password/request - Password reset request
✅ POST /api/auth/reset-password/confirm - Password reset confirmation
✅ Reset token validation (token format, expiry)
✅ Password strength validation (security payloads)
✅ Token replay attack detection

Agent Endpoints (FUZZ-04)

Test File: test_agent_api_fuzzing.py

✅ POST /api/agents/{id}/run - Agent execution
✅ GET /api/agents/{id}/status - Agent status polling
✅ DELETE /api/agents/{id} - Agent deletion

Test File: test_agent_streaming_fuzzing.py

✅ WebSocket /api/agents/{id}/stream - Agent streaming responses
✅ SSE /api/agents/{id}/events - Agent events streaming

Canvas & Workflow Endpoints (FUZZ-05)

Test File: test_canvas_presentation_fuzzing.py

✅ POST /api/canvas/present - Canvas presentation
✅ POST /api/canvas/{id}/submit - Form submission
✅ POST /api/canvas/{id}/execute - Canvas command execution

Test File: test_workflow_api_fuzzing.py

✅ POST /api/workflows - Workflow creation
✅ PUT /api/workflows/{id} - Workflow update
✅ POST /api/workflows/{id}/trigger - Workflow trigger
✅ POST /api/workflows/{id}/schedule - Workflow scheduling

Test File: test_skill_installation_fuzzing.py

✅ POST /api/skills/import - Skill import (npm + Python)
✅ POST /api/skills/{id}/execute - Skill execution
✅ POST /api/skills/{id}/promote - Skill promotion

Test File: test_trigger_execution_fuzzing.py

✅ POST /api/triggers/validate - Trigger validation
✅ POST /api/triggers/execute - Trigger execution
✅ POST /api/triggers/schedule - Trigger scheduling
✅ POST /api/triggers/webhook - Webhook trigger
✅ POST /api/triggers/events - Event-based triggers

Input Space Coverage

Fuzzed Input Types:

Random strings (100-500 chars)
None/null values
Empty strings
Malformed JSON
Invalid base64
Unicode normalization issues
Huge inputs (1000-10000000 chars)

Security Payloads:

SQL injection: '; DROP TABLE users; --
XSS: <script>alert(1)</script>
Null bytes: \x00
Path traversal: ../../etc/passwd
Command injection: ; rm -rf /
Code injection: __import__('os').system('rm -rf /')

Edge Cases:

Negative timestamps
Huge numbers
Missing required fields
Type mismatches (int instead of str)
Duplicate fields
Malformed JWT structure
Cyclical dependencies (workflows)
Webhook URL validation (javascript:, file://, etc.)

CI/CD Integration

Weekly Pipeline

Workflow: .github/workflows/bug-discovery-weekly.yml

Schedule: Sunday 3 AM UTC (cron 0 3 * * 0)

Duration: 1 hour per campaign (5 campaigns = ~5 hours total)

Jobs:

api-fuzzing (90 min timeout)
- Run fuzzing campaigns: run_fuzzing_campaigns.py --duration 3600
- Upload crash artifacts (90 day retention)
- Upload corpus artifacts (90 day retention)
- Aggregate crash reports: aggregate_crash_reports.py
- Upload fuzzing reports (90 day retention)
- File bugs for crashes: file_bugs_from_artifacts.py
bug-discovery (150 min timeout)
- Run pytest with markers: -m "fuzzing or chaos or browser"
- Upload test results, screenshots, logs
- Generate test report
- File bugs for failures

FUZZ-07 Compliance

Requirement: Fuzzing does NOT run on PR CI pipeline

Rationale: Fuzzing is resource-intensive (1-5 hours) and would slow down PR feedback. Fast PR tests (<10 min) enable rapid development, while weekly fuzzing provides comprehensive crash discovery.

Implementation:

✅ Fuzzing in weekly pipeline only (bug-discovery-weekly.yml)
✅ NOT in PR pipeline (ci.yml, pr.yml)
✅ Manual trigger support (workflow_dispatch)

Verification:

# Verify fuzzing not in PR workflows
grep -L "fuzzing" .github/workflows/*.yml | grep -E "(pr|pull|check)"

Manual Trigger

# Trigger weekly workflow manually (via GitHub UI)
# https://github.com/owner/repo/actions/workflows/bug-discovery-weekly.yml
# Click "Run workflow" button

Corpus Management

What is Corpus?

Corpus files contain interesting inputs discovered during fuzzing that exercise specific code paths or trigger edge cases.

Example:

tests/fuzzing/campaigns/corpus/auth/login_sql_injection.input

Content (binary):

admin' OR '1'='1'--

Corpus Directory Structure

tests/fuzzing/campaigns/corpus/
├── auth/              # Authentication corpus
├── agent/             # Agent execution corpus
├── workflow/          # Workflow management corpus
├── canvas/            # Canvas presentation corpus
└── skill/             # Skill installation corpus

Re-seeding Campaigns

Use corpus to speed up future campaigns:

# Set corpus directory
export FUZZ_CORPUS_DIR="tests/fuzzing/campaigns/corpus/auth"

# Run campaign with corpus
FUZZ_ITERATIONS=10000 pytest tests/fuzzing/test_auth_api_fuzzing.py -v -m fuzzing

Benefits:

Faster coverage (corpus inputs explore interesting paths immediately)
Reproducible crashes (corpus inputs are reproducible test cases)
Regression testing (corpus grows with discovered bugs)

Corpus Management Best Practices

Add interesting inputs: When crashes are discovered, add *.input files to corpus
Organize by endpoint: Use subdirectories (auth/, agent/, workflow/)
Commit to git: Corpus is version-controlled with tests
Update regularly: Add new inputs after each fuzzing campaign

Crash Analysis

Crash Artifacts

Location: tests/fuzzing/campaigns/crashes/

Format:

tests/fuzzing/campaigns/crashes/POST_api_auth_login_2026-03-24T12-34-56/
├── crash-00001.input  # Binary input that triggered crash
├── crash-00001.log    # Stack trace and error message
├── crash-00002.input
├── crash-00002.log
└── ...

Crash Deduplication

Tool: CrashDeduplicator in tests/fuzzing/campaigns/crash_deduplicator.py

Method: SHA256 hash of normalized error signature

# Normalize stack trace (remove line numbers)
normalized_trace = re.sub(r' line \d+', ' line N', stack_trace)

# Generate SHA256 hash
signature_hash = hashlib.sha256(normalized_trace.encode('utf-8')).hexdigest()

Benefits:

Groups duplicate crashes (same root cause)
Reduces bug filing noise
Identifies high-frequency crashes

Bug Filing

Tool: BugFilingService in tests/bug_discovery/bug_filing_service.py

Trigger: Automatic on crash discovery (via FuzzingOrchestrator)

Bug Template:

## Fuzzing Crash: POST /api/auth/login

**Test Type**: Fuzzing
**Target Endpoint**: POST /api/auth/login
**Signature Hash**: `abc123...`

### Crash Metadata
- **Crash Input**: `deadbeef...` (hex)
- **Stack Trace**: ```
Traceback (most recent call last):
  File "...", line 123, in test_login_endpoint_fuzzing
    ...
AssertionError: Expected status 200, got 500

Reproduction

FUZZ_ITERATIONS=1 pytest tests/fuzzing/test_auth_api_fuzzing.py::test_login_endpoint_fuzzing -v

Environment

Commit: abc123...
Date: 2026-03-24T12:34:56Z
Campaign ID: POST_api_auth_login_2026-03-24T12-34-56


### Crash Reports

**Tool**: `aggregate_crash_reports.py`

**Usage**:
```bash
python3 tests/fuzzing/scripts/aggregate_crash_reports.py \
  --crash-dirs "tests/fuzzing/campaigns/crashes/*" \
  --output fuzzing-report.md

Report Sections:

Summary: Total crashes, unique crashes, affected endpoints
Crashes by Endpoint: Grouped by API endpoint
Top Crash Signatures: Sorted by frequency
Bug Filing Status: Filed, duplicate, skipped
Trend Analysis: Comparison with previous week

Best Practices

1. Start Small, Scale Up

Quick verification (100 iterations, ~10 seconds):

FUZZ_ITERATIONS=100 pytest tests/fuzzing/test_auth_api_fuzzing.py -v -m fuzzing

Full campaign (10000 iterations, ~5-10 minutes):

FUZZ_ITERATIONS=10000 pytest tests/fuzzing/test_auth_api_fuzzing.py -v -m fuzzing

Scale based on crash discovery:

If crashes found: Increase iterations to 100000+
If no crashes: Decrease iterations to 1000 (speed up)

2. Use TestClient Pattern

Correct (fast, no network overhead):

from fastapi.testclient import TestClient

app.dependency_overrides[get_db] = lambda: db_session
client = TestClient(app)
response = client.post("/api/auth/login", json={...})
assert response.status_code in [200, 400, 401, 422]

Incorrect (slow, network overhead):

import httpx

response = httpx.post("http://localhost:8000/api/auth/login", json={...})
# Don't do this - network overhead is 10-100x slower

Exception: Use httpx for streaming endpoints (WebSocket, SSE)

3. Reuse Fixtures from e2e_ui

Correct (no duplication, 10-100x faster):

from tests.e2e_ui.fixtures.database_fixtures import db_session
from tests.e2e_ui.fixtures.auth_fixtures import authenticated_user

def test_login_endpoint_fuzzing(db_session, authenticated_user):
    # db_session: Isolated database session
    # authenticated_user: Pre-created user with valid credentials
    pass

Incorrect (duplication, slower):

def test_login_endpoint_fuzzing():
    # Don't create fixtures inline - duplication!
    db_session = SessionLocal()
    user = User(email="test@example.com", password="...")
    db_session.add(user)
    db_session.commit()
    # ...

4. NEVER Fuzz Production URLs

Correct (TestClient with database override):

app.dependency_overrides[get_db] = lambda: db_session
client = TestClient(app)
response = client.post("/api/auth/login", json={...})

Incorrect (production URL, dangerous):

import httpx

response = httpx.post("https://api.atom.ai/api/auth/login", json={...})
# NEVER DO THIS - fuzzing production is dangerous!

5. Set Reasonable Timeouts

@pytest.mark.timeout(300)  # 5 minutes per test
def test_login_endpoint_fuzzing(db_session, authenticated_user):
    pass

Reasoning:

Prevents infinite loops (10000 iterations)
Allows CI to detect hangs
5 minutes is sufficient for 10000 iterations

6. Test Status Codes, Not Just Crashes

# Good: Test all expected status codes
assert response.status_code in [200, 400, 401, 409, 422]

# Bad: Only test for success (misses 500 errors)
assert response.status_code == 200

Reasoning:

Crashes are not the only bugs
500 errors indicate unhandled exceptions
422 errors indicate validation issues
400/401/409 errors indicate business logic bugs

Troubleshooting

Fuzzing Hangs

Symptom: Test runs forever, never completes

Solutions:

Reduce iterations: FUZZ_ITERATIONS=100
Check timeout config: @pytest.mark.timeout(300)
Check for infinite loops in endpoint code
Use pytest --timeout to enforce global timeout

Example:

# Add 10-minute global timeout
pytest --timeout=600 tests/fuzzing/test_auth_api_fuzzing.py -v -m fuzzing

No Crashes Found

Symptom: Fuzzing completes with 0 crashes

Solutions:

Increase iterations: FUZZ_ITERATIONS=100000
Expand input space (fuzz more fields)
Check endpoint code (overly defensive?)
Add security payloads (SQLi, XSS, null bytes)
Check status code assertions (too permissive?)

Example:

# Bad: Too permissive (misses crashes)
assert response.status_code in range(200, 600)

# Good: Explicit status codes
assert response.status_code in [200, 400, 401, 409, 422, 500]

Duplicate Bugs Filed

Symptom: Multiple GitHub issues for same crash

Solutions:

Check deduplication logic: CrashDeduplicator.deduplicate_crashes()
Verify signature hash is stable (line numbers removed?)
Check BugFilingService duplicate detection
Review signature normalization regex

Example:

# Check normalization logic
normalized_trace = re.sub(r' line \d+', ' line N', stack_trace)
# Should remove line numbers for stable hashing

CI Timeout

Symptom: CI workflow fails with timeout

Solutions:

Reduce campaign duration: --duration 1800 (30 min)
Run fewer endpoints: --campaign "POST /api/auth/login"
Increase CI timeout: timeout-minutes: 180
Parallelize campaigns (run multiple jobs)

Example:

# Reduce timeout
- name: Run fuzzing campaigns
  run: |
    python3 tests/fuzzing/scripts/run_fuzzing_campaigns.py --duration 1800
  timeout-minutes: 45

Fuzzing Skipped in CI

Symptom: Tests skip with "atheris not installed"

Solutions:

Verify atheris in requirements-testing.txt
Check CI workflow installs atheris: pip install atheris
Verify pytest markers: @pytest.mark.fuzzing
Check test collection: pytest --collect-only tests/fuzzing/

Example:

# Verify CI installs atheris
- name: Install dependencies
  run: |
    pip install atheris  # Fuzzing support

Summary

API fuzzing with Atheris provides coverage-guided crash discovery for FastAPI endpoints, complementing unit tests, integration tests, and property-based tests. Key takeaways:

Start small: 100 iterations for quick verification, 10000 for full campaign
Use TestClient: Fast, no network overhead, database override for isolation
Reuse fixtures: db_session, authenticated_user from e2e_ui (10-100x faster)
NEVER fuzz production: Use TestClient, not httpx with production URLs
CI/CD integration: Weekly campaigns (1 hour), NOT in PR pipeline (FUZZ-07)
Corpus management: Re-seed campaigns with interesting inputs
Crash analysis: Deduplicate by signature, file bugs automatically

Fuzzing is not a silver bullet, but it's a powerful tool for discovering crashes in parsing, validation, and business logic code that human-written tests miss.

API Fuzzing Infrastructure

Overview

Why Atheris?

Goal

Quick Start

Install Dependencies

Run Single Fuzz Test

Run All Fuzz Tests

Run Fuzzing Campaign

Fuzzing Infrastructure

FuzzingOrchestrator

CrashDeduplicator

BugFilingService

Corpus Management

Fuzzing Test Structure

Test File Pattern

Pytest Markers

Fixtures

Crash Artifacts

Coverage

Authentication Endpoints (FUZZ-03)

Agent Endpoints (FUZZ-04)

Canvas & Workflow Endpoints (FUZZ-05)

Input Space Coverage

CI/CD Integration

Weekly Pipeline

FUZZ-07 Compliance

Manual Trigger

Corpus Management

What is Corpus?

Corpus Directory Structure

Re-seeding Campaigns

Corpus Management Best Practices

Crash Analysis

Crash Artifacts

Crash Deduplication

Bug Filing

Reproduction

Environment

Best Practices

1. Start Small, Scale Up

2. Use TestClient Pattern

3. Reuse Fixtures from e2e_ui

4. NEVER Fuzz Production URLs

5. Set Reasonable Timeouts

6. Test Status Codes, Not Just Crashes

Troubleshooting

Fuzzing Hangs

No Crashes Found

Duplicate Bugs Filed

CI Timeout

Fuzzing Skipped in CI

Further Reading

Research Documentation

Related Documentation

External Resources

Summary