mirror of
https://github.com/FuzzingLabs/fuzzforge_ai.git
synced 2026-02-13 16:32:47 +00:00
fe50d4ef72
This commit implements a complete Python fuzzing workflow using Atheris: ## Python Worker (workers/python/) - Dockerfile with Python 3.11, Atheris, and build tools - Generic worker.py for dynamic workflow discovery - requirements.txt with temporalio, boto3, atheris dependencies - Added to docker-compose.temporal.yaml with dedicated cache volume ## AtherisFuzzer Module (backend/toolbox/modules/fuzzer/) - Reusable module extending BaseModule - Auto-discovers fuzz targets (fuzz_*.py, *_fuzz.py, fuzz_target.py) - Recursive search to find targets in nested directories - Dynamically loads TestOneInput() function - Configurable max_iterations and timeout - Real-time stats callback support for live monitoring - Returns findings as ModuleFinding objects ## Atheris Fuzzing Workflow (backend/toolbox/workflows/atheris_fuzzing/) - Temporal workflow for orchestrating fuzzing - Downloads user code from MinIO - Executes AtherisFuzzer module - Uploads results to MinIO - Cleans up cache after execution - metadata.yaml with vertical: python for routing ## Test Project (test_projects/python_fuzz_waterfall/) - Demonstrates stateful waterfall vulnerability - main.py with check_secret() that leaks progress - fuzz_target.py with Atheris TestOneInput() harness - Complete README with usage instructions ## Backend Fixes - Fixed parameter merging in REST API endpoints (workflows.py) - Changed workflow parameter passing from positional args to kwargs (manager.py) - Default parameters now properly merged with user parameters ## Testing ✅ Worker discovered AtherisFuzzingWorkflow ✅ Workflow executed end-to-end successfully ✅ Fuzz target auto-discovered in nested directories ✅ Atheris ran 100,000 iterations ✅ Results uploaded and cache cleaned
97 lines
2.7 KiB
Python
97 lines
2.7 KiB
Python
"""
|
|
Example application with a stateful vulnerability.
|
|
|
|
This simulates a password checking system that leaks state information
|
|
through a global progress variable - a classic waterfall vulnerability.
|
|
"""
|
|
|
|
# Global state - simulates session state
|
|
progress = 0
|
|
SECRET = "FUZZINGLABS" # 11 characters
|
|
|
|
|
|
def check_secret(input_data: bytes) -> bool:
|
|
"""
|
|
Vulnerable function: checks secret character by character.
|
|
|
|
This is a waterfall vulnerability - state leaks through the progress variable.
|
|
|
|
Real-world analogy:
|
|
- Timing attacks on password checkers
|
|
- Protocol state machines with sequential validation
|
|
- Multi-step authentication flows
|
|
|
|
Args:
|
|
input_data: Input bytes to check
|
|
|
|
Returns:
|
|
True if progress was made, False otherwise
|
|
|
|
Raises:
|
|
SystemError: When complete secret is discovered (vulnerability trigger)
|
|
"""
|
|
global progress
|
|
|
|
if len(input_data) > progress:
|
|
if input_data[progress] == ord(SECRET[progress]):
|
|
progress += 1
|
|
|
|
# Progress indicator (useful for monitoring during fuzzing)
|
|
if progress % 2 == 0: # Every 2 characters
|
|
print(f"[DEBUG] Progress: {progress}/{len(SECRET)} characters matched")
|
|
|
|
# VULNERABILITY: Crashes when complete secret found
|
|
if progress == len(SECRET):
|
|
raise SystemError(f"SECRET COMPROMISED: {SECRET}")
|
|
|
|
return True
|
|
else:
|
|
# Wrong character - reset progress
|
|
progress = 0
|
|
return False
|
|
|
|
return False
|
|
|
|
|
|
def reset_state():
|
|
"""Reset the global state (useful for testing)"""
|
|
global progress
|
|
progress = 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
"""Example usage showing the vulnerability"""
|
|
print("=" * 60)
|
|
print("Waterfall Vulnerability Demonstration")
|
|
print("=" * 60)
|
|
print(f"Secret: {SECRET}")
|
|
print(f"Secret length: {len(SECRET)} characters")
|
|
print()
|
|
|
|
# Test inputs showing progressive discovery
|
|
test_inputs = [
|
|
b"F", # First char correct
|
|
b"FU", # First two chars correct
|
|
b"FUZ", # First three chars correct
|
|
b"WRONG", # Wrong - resets progress
|
|
b"FUZZINGLABS", # Complete secret - triggers crash!
|
|
]
|
|
|
|
for test in test_inputs:
|
|
reset_state() # Start fresh for each test
|
|
print(f"Testing input: {test.decode(errors='ignore')!r}")
|
|
|
|
try:
|
|
result = check_secret(test)
|
|
print(f" Result: {result}, Progress: {progress}/{len(SECRET)}")
|
|
except SystemError as e:
|
|
print(f" 💥 CRASH: {e}")
|
|
|
|
print()
|
|
|
|
print("=" * 60)
|
|
print("To fuzz this vulnerability with FuzzForge:")
|
|
print(" ff init")
|
|
print(" ff workflow run atheris_fuzzing .")
|
|
print("=" * 60)
|