Build a Custom REST API Fuzzer
Build a Python fuzzer that parses OpenAPI specs and detects API vulnerabilities.
Introduction
30 Second Summary
Every app you use depends on APIs to move data around. A single unvalidated input to one of those APIs can crash a server, leak private data, or open a door for attackers.
In this project, you will build a custom API fuzzer in Python that automatically discovers security vulnerabilities by bombarding endpoints with malicious inputs and classifying the results.
What You'll Build
A two-component security testing tool: a deliberately vulnerable Flask API with planted bugs, and a Python fuzzer that parses an OpenAPI specification to generate hundreds of mutated test cases.
By the end of this project, you'll have:
- A vulnerable Flask REST API with four endpoints containing deliberate security flaws that your fuzzer will detect.
- A Python fuzzer engine that parses an OpenAPI spec, generates mutated requests using four mutation strategies, and fires them at a target.
- A JSON findings report that classifies discovered issues by severity (crash, unexpected status, schema violation).
- Secret Mission: Add authentication bypass detection to your fuzzer to catch broken access control.
Are there any prerequisites?
You should be comfortable with Python (classes, HTTP requests, and file I/O). A basic understanding of REST APIs and how they handle requests is helpful.
Before We Start
You are about to build a custom API fuzzer from scratch. Fuzzing is a technique where you throw thousands of unexpected, malformed, or boundary-case inputs at an application to see where it breaks.
Before you can build the fuzzer and its target API, you need a clean project directory with the right dependencies installed. In this step, you will create the project structure, set up a virtual environment, and install the three Python packages the project depends on.
In this step, get ready to:
- Create the project directory and a Python virtual environment.
- Install Flask, requests, and PyYAML.
- Create the three project files you will work in throughout this project.
Create your project directory and virtual environment
A virtual environment keeps your project's packages isolated from the rest of your system. This prevents version conflicts between projects.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir api-fuzzer
cd api-fuzzer- Create a virtual environment by running this command:
python3 -m venv venvWhat does this command do?
python3 -m venv venv tells Python to run its built-in venv module and create a new virtual environment in a folder called venv/.
This folder contains its own copy of the Python interpreter and a private space for installed packages.
- Activate the virtual environment:
macOS
source venv/bin/activateWindows
venv\Scripts\activateYou should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.
Don't see (venv) in your prompt?
Make sure you ran the activate command that matches your operating system. On macOS, it is source venv/bin/activate and on Windows it is venv\Scripts\activate.
If you see a permissions error on Windows, try running PowerShell as Administrator and executing Set-ExecutionPolicy RemoteSigned first.
help me activate my Python virtual environment
Install the required packages
Your fuzzer needs three packages: a web framework for the target API, an HTTP library for sending requests, and a YAML parser for reading the API specification.
- Install all three packages by running this command:
pip install Flask==3.1.3 requests==2.34.2 PyYAML==6.0.3What are these packages?
- Flask is a lightweight web framework. You will use it to build the deliberately vulnerable target API.
- requests is a popular HTTP library. The fuzzer uses it to fire mutated requests at the target.
- PyYAML parses YAML files. You will use it to read the OpenAPI specification that describes the target API's endpoints.
- Verify the packages installed correctly by running:
pip listYou should see Flask, requests, and PyYAML in the output along with their version numbers.
Seeing an installation error?
Make sure your virtual environment is active. You should see (venv) in your prompt. If it is not active, packages install globally and may fail due to permissions.
help me fix pip install errors in my virtual environment
Create the project files
Your project uses three files. You will create them now as empty files and fill them with code in the following steps.
- Create the three project files by running this command:
macOS
touch target_api.py openapi.yaml fuzzer.pyWindows
New-Item target_api.py, openapi.yaml, fuzzer.pyWhat is each file for?
- target_api.py is the deliberately vulnerable API your fuzzer will attack.
- openapi.yaml is the specification that formally describes the API's endpoints, parameters, and expected responses.
- fuzzer.py is the fuzzer engine that parses the spec, generates mutated requests, and classifies responses.
- Confirm the files exist by running:
lsYou should see target_api.py, openapi.yaml, fuzzer.py, and the venv/ folder in the output.
Your project is ready to go. Next up, you will build the deliberately vulnerable target API that your fuzzer will attack.
Build the Vulnerable Target API
Your environment is ready with all dependencies installed. Now you need something to fuzz.
Every fuzzer needs a target. In this step, you will build a deliberately vulnerable Flask REST API with four endpoints, each containing a different planted security flaw. You will also write an OpenAPI specification that formally describes how the API should behave.
In this step, get ready to:
- Write a Flask API with four deliberately vulnerable endpoints.
- Write an OpenAPI YAML specification describing the API's expected behavior.
- Run the target API and confirm all endpoints respond correctly.
Write the vulnerable Flask API
A good fuzzing target needs realistic vulnerabilities. Your API will have four endpoints, each with a different category of security flaw: a crash on bad input types, missing input validation, simulated SQL injection, and broken access control.
- Switch back to your code editor with the api-fuzzer folder open.
- Open target_api.py and paste in the following code:
from flask import Flask, request, jsonify
app = Flask(__name__)
# In-memory "database"
users_db = {
1: {"id": 1, "name": "Alice", "email": "alice@example.com"},
2: {"id": 2, "name": "Bob", "email": "bob@example.com"},
3: {"id": 3, "name": "Charlie", "email": "charlie@example.com"},
}
admin_data = {
"secret_key": "sk-super-secret-admin-key-12345",
"database_url": "postgres://admin:password@prod-db:5432/main",
"users_count": 3,
}
@app.route("/users/<user_id>", methods=["GET"])
def get_user(user_id):
uid = int(user_id)
user = users_db.get(uid)
if user:
return jsonify(user), 200
return jsonify({"error": "User not found"}), 404
@app.route("/users", methods=["POST"])
def create_user():
data = request.get_json(force=True, silent=True)
if data is None:
return jsonify({"error": "Invalid JSON"}), 400
new_id = max(users_db.keys()) + 1
users_db[new_id] = {
"id": new_id,
"name": data.get("name", ""),
"email": data.get("email", ""),
}
return jsonify(users_db[new_id]), 201
@app.route("/search", methods=["POST"])
def search_users():
data = request.get_json(force=True, silent=True)
if data is None:
return jsonify({"error": "Invalid JSON"}), 400
query = data.get("query", "")
dangerous_chars = ["'", ";", "--", "DROP", "OR 1=1", "UNION"]
for char in dangerous_chars:
if char.lower() in query.lower():
raise Exception(f"Database error: syntax error near '{query}'")
results = [
u for u in users_db.values() if query.lower() in u["name"].lower()
]
return jsonify({"results": results}), 200
@app.route("/admin", methods=["GET"])
def get_admin():
return jsonify(admin_data), 200
if __name__ == "__main__":
app.run(debug=False, port=5000)What does this code do?
This file creates a Flask web server with four endpoints, each containing a different vulnerability your fuzzer will later detect:
- GET /users/<user_id> calls int(user_id) directly. Any non-integer input (like abc or an empty string) will crash the server with a 500 error.
- POST /users accepts any JSON body with zero validation. There are no type checks, length limits, or required-field enforcement.
- POST /search checks for SQL metacharacters and raises an unhandled exception when it finds them. This simulates a backend that breaks on injection attempts.
- GET /admin returns sensitive data to anyone who requests it, with no authentication check at all.
- Save target_api.py.
You will not see output yet. The next substep runs the server and confirms everything works.
Seeing a red underline in your editor?
If your editor highlights flask as unresolved, make sure your virtual environment is still active. Your editor's Python interpreter must point to the venv you created in Step 1.
Help me fix Flask import errors in my editor
Write the OpenAPI specification
An OpenAPI specification is a machine-readable contract that describes your API's endpoints, parameters, request bodies, and expected responses. Your fuzzer will parse this file to know what to attack and what "normal" looks like.
- Open openapi.yaml and paste in the following specification:
openapi: "3.0.3"
info:
title: Vulnerable Demo API
version: "1.0.0"
description: A deliberately vulnerable API for fuzzing practice
servers:
- url: http://127.0.0.1:5000
paths:
/users/{user_id}:
get:
summary: Get a user by ID
parameters:
- name: user_id
in: path
required: true
schema:
type: integer
responses:
"200":
description: User found
"404":
description: User not found
/users:
post:
summary: Create a new user
requestBody:
required: true
content:
application/json:
schema:
type: object
required:
- name
- email
properties:
name:
type: string
minLength: 1
maxLength: 100
email:
type: string
format: email
responses:
"201":
description: User created
"400":
description: Invalid input
/search:
post:
summary: Search for users
requestBody:
required: true
content:
application/json:
schema:
type: object
required:
- query
properties:
query:
type: string
minLength: 1
maxLength: 200
responses:
"200":
description: Search results
"400":
description: Invalid input
/admin:
get:
summary: Get admin panel data
security:
- bearerAuth: []
responses:
"200":
description: Admin data
"401":
description: Unauthorized
"403":
description: Forbidden
components:
securitySchemes:
bearerAuth:
type: http
scheme: bearerWhat does this spec tell the fuzzer?
The spec declares what the API promises to do. The fuzzer will use this to find mismatches between the contract and reality:
- user_id is declared as type: integer. The fuzzer will send non-integers to see if the server enforces this.
- name and email are required with length limits. The fuzzer will omit them and send oversized values.
- The /admin endpoint declares bearerAuth security. The fuzzer can test whether the API actually enforces authentication.
- Save openapi.yaml.
You will not see output yet. The next substep starts the server so you can confirm everything is wired up.
YAML indentation errors later?
YAML is whitespace-sensitive. Make sure you pasted the entire block without extra tabs or mixed spaces. If you get parse errors in a later step, re-paste this file and ensure every indent uses exactly two spaces.
Help me fix YAML indentation errors
✔️ Awesome, I've got everything!
Great. Make sure both files are saved before moving on.
ⓧ I'd like to double check the full code
Here is the full target_api.py:
from flask import Flask, request, jsonify
app = Flask(__name__)
# In-memory "database"
users_db = {
1: {"id": 1, "name": "Alice", "email": "alice@example.com"},
2: {"id": 2, "name": "Bob", "email": "bob@example.com"},
3: {"id": 3, "name": "Charlie", "email": "charlie@example.com"},
}
admin_data = {
"secret_key": "sk-super-secret-admin-key-12345",
"database_url": "postgres://admin:password@prod-db:5432/main",
"users_count": 3,
}
@app.route("/users/<user_id>", methods=["GET"])
def get_user(user_id):
uid = int(user_id)
user = users_db.get(uid)
if user:
return jsonify(user), 200
return jsonify({"error": "User not found"}), 404
@app.route("/users", methods=["POST"])
def create_user():
data = request.get_json(force=True, silent=True)
if data is None:
return jsonify({"error": "Invalid JSON"}), 400
new_id = max(users_db.keys()) + 1
users_db[new_id] = {
"id": new_id,
"name": data.get("name", ""),
"email": data.get("email", ""),
}
return jsonify(users_db[new_id]), 201
@app.route("/search", methods=["POST"])
def search_users():
data = request.get_json(force=True, silent=True)
if data is None:
return jsonify({"error": "Invalid JSON"}), 400
query = data.get("query", "")
dangerous_chars = ["'", ";", "--", "DROP", "OR 1=1", "UNION"]
for char in dangerous_chars:
if char.lower() in query.lower():
raise Exception(f"Database error: syntax error near '{query}'")
results = [
u for u in users_db.values() if query.lower() in u["name"].lower()
]
return jsonify({"results": results}), 200
@app.route("/admin", methods=["GET"])
def get_admin():
return jsonify(admin_data), 200
if __name__ == "__main__":
app.run(debug=False, port=5000)And here is the full openapi.yaml:
openapi: "3.0.3"
info:
title: Vulnerable Demo API
version: "1.0.0"
description: A deliberately vulnerable API for fuzzing practice
servers:
- url: http://127.0.0.1:5000
paths:
/users/{user_id}:
get:
summary: Get a user by ID
parameters:
- name: user_id
in: path
required: true
schema:
type: integer
responses:
"200":
description: User found
"404":
description: User not found
/users:
post:
summary: Create a new user
requestBody:
required: true
content:
application/json:
schema:
type: object
required:
- name
- email
properties:
name:
type: string
minLength: 1
maxLength: 100
email:
type: string
format: email
responses:
"201":
description: User created
"400":
description: Invalid input
/search:
post:
summary: Search for users
requestBody:
required: true
content:
application/json:
schema:
type: object
required:
- query
properties:
query:
type: string
minLength: 1
maxLength: 200
responses:
"200":
description: Search results
"400":
description: Invalid input
/admin:
get:
summary: Get admin panel data
security:
- bearerAuth: []
responses:
"200":
description: Admin data
"401":
description: Unauthorized
"403":
description: Forbidden
components:
securitySchemes:
bearerAuth:
type: http
scheme: bearerStart the target API and test manually
Before you build the fuzzer, confirm the API starts cleanly and responds to normal requests. This gives you a known-good baseline.
- Start the Flask server by running this command in your terminal (make sure you are inside the api-fuzzer directory):
python target_api.pyWhat should I see?
Your terminal should show a line indicating the server is running on http://127.0.0.1:5000. The terminal will stay occupied while the server runs.
- Open a second terminal window or tab (keep the server running in the first one).
- Activate your virtual environment in the new terminal, then test the API by running this command:
curl http://127.0.0.1:5000/users/1You should see a JSON response like this:
{"email":"alice@example.com","id":1,"name":"Alice"}Connection refused?
Make sure python target_api.py is still running in your first terminal. If you see a traceback instead of the server starting, check that you saved target_api.py without any syntax errors.
Help me troubleshoot my Flask connection
Your vulnerable API is running and confirmed working. Next up, you will build the fuzzer engine that parses the OpenAPI spec and generates hundreds of mutated test cases to throw at these endpoints.
Build the Fuzzer - Spec Parser and Mutation Strategies
You have a vulnerable API with a formal specification describing how it should behave. Now you need the tool that reads that spec and automatically generates attack payloads.
In this step, you will build the fuzzer's brain: a spec parser that extracts fuzz targets, and a mutation engine that generates hundreds of malicious inputs tailored to each parameter type.
In this step, get ready to:
- Parse the OpenAPI spec into structured fuzz targets.
- Implement four mutation strategies: boundary values, type confusion, injection payloads, and missing/extra fields.
- Wire the parser and mutation engine together to generate test cases.
Write the OpenAPI spec parser
The fuzzer needs to understand what endpoints exist, what parameters they accept, and what responses are considered normal. Your openapi.yaml file already contains all of this information in a machine-readable format.
- Open fuzzer.py in your code editor and add the imports and spec parser by pasting the following code:
import json
import time
import sys
from datetime import datetime, timezone
import requests
import yaml
# --- OpenAPI Spec Parser ---
def parse_openapi_spec(spec_path):
"""Parse an OpenAPI YAML spec and extract fuzz targets."""
with open(spec_path, "r") as f:
spec = yaml.safe_load(f)
base_url = spec["servers"][0]["url"]
fuzz_targets = []
for path, methods in spec.get("paths", {}).items():
for method, details in methods.items():
if method not in ("get", "post", "put", "delete", "patch"):
continue
target = {
"path": path,
"method": method.upper(),
"base_url": base_url,
"path_params": [],
"body_schema": None,
"expected_statuses": [],
"requires_auth": False,
}
# Extract path parameters
for param in details.get("parameters", []):
if param.get("in") == "path":
target["path_params"].append({
"name": param["name"],
"type": param.get("schema", {}).get("type", "string"),
})
# Extract request body schema
request_body = details.get("requestBody", {})
content = request_body.get("content", {})
json_content = content.get("application/json", {})
if json_content:
target["body_schema"] = json_content.get("schema", {})
# Extract expected status codes
for status_code in details.get("responses", {}).keys():
target["expected_statuses"].append(str(status_code))
# Check if auth is required
if details.get("security"):
target["requires_auth"] = True
fuzz_targets.append(target)
return fuzz_targetsWhat does this code do?
The imports bring in libraries you will use throughout the fuzzer. yaml parses the OpenAPI spec, requests sends HTTP calls, and json handles the findings report later.
The parse_openapi_spec() function reads your YAML file and walks through every path and method. For each endpoint it builds a "fuzz target" dictionary containing the URL path, HTTP method, any path parameters with their expected types, the request body schema, a list of expected status codes, and whether authentication is required.
- Save fuzzer.py.
- Test the parser by running this command in your terminal:
python -c "from fuzzer import parse_openapi_spec; targets = parse_openapi_spec('openapi.yaml'); [print(f\"{t['method']} {t['path']} params={t['path_params']} auth={t['requires_auth']}\") for t in targets]"You should see all four endpoints printed with their methods, path parameters, and auth requirements.
Seeing a ModuleNotFoundError?
- Make sure your virtual environment is active. You should see (venv) in your terminal prompt.
- Confirm you are running the command from inside the api-fuzzer/ folder where fuzzer.py and openapi.yaml live.
Help me debug my spec parser import error.
Implement mutation strategies
A fuzzer is only as good as its mutations. Professional fuzzers use multiple categories of malicious input to catch different classes of bugs.
What is mutation-based fuzzing?
Mutation-based fuzzing takes valid inputs and systematically transforms them into potentially dangerous ones. Instead of random noise, each mutation targets a specific class of vulnerability.
Boundary values catch off-by-one errors and buffer overflows. Type confusion catches missing type checks. Injection payloads catch unsanitized database and shell commands. Missing fields catch assumptions about required data.
- In fuzzer.py, add the MutationEngine class with its payload constants below the parse_openapi_spec() function:
# --- Mutation Strategies ---
class MutationEngine:
"""Generates mutated payloads for fuzzing."""
BOUNDARY_STRINGS = [
"",
" ",
"a" * 10000,
"null",
"undefined",
"0",
"-1",
"99999999999999999999",
"\x00",
"\n\r\n",
]
BOUNDARY_INTEGERS = [
0, -1, -999999999, 999999999, 2147483647, -2147483648,
]
TYPE_CONFUSION = [
None,
True,
False,
[],
{},
[None],
{"__proto__": "polluted"},
12345,
3.14159,
]
INJECTION_PAYLOADS = [
"' OR 1=1--",
"'; DROP TABLE users;--",
"1; SELECT * FROM users",
"UNION SELECT NULL,NULL,NULL--",
"; ls -la",
"| cat /etc/passwd",
"$(whoami)",
"`id`",
"<script>alert(1)</script>",
"<img src=x onerror=alert(1)>",
"{{7*7}}",
"${7*7}",
]What do these constants represent?
- BOUNDARY_STRINGS tests edge cases: empty input, a 10,000-character string, null bytes, and values that might confuse type coercion.
- BOUNDARY_INTEGERS includes zero, negatives, and MAX_INT values (2,147,483,647) that trigger integer overflow bugs.
- TYPE_CONFUSION sends arrays where objects are expected, booleans where strings are expected, and prototype pollution payloads.
- INJECTION_PAYLOADS covers SQL injection, command injection, XSS, and template injection patterns.
- Save fuzzer.py. You will not see a change yet. The next chunk adds the methods that use these constants.
- Below the INJECTION_PAYLOADS list, add the mutate_path_param method inside the MutationEngine class:
def mutate_path_param(self, param_type):
"""Generate mutations for a path parameter."""
mutations = []
if param_type == "integer":
# For integer params: boundary ints as strings, plus type-confused values
mutations.extend([str(v) for v in self.BOUNDARY_INTEGERS])
mutations.extend(["abc", "", "null", "1.5", "1e10", "-0", "0x1A"])
else:
# For string params: boundary strings plus injection payloads
mutations.extend(self.BOUNDARY_STRINGS)
mutations.extend(self.INJECTION_PAYLOADS)
return mutationsWhat does this method do?
Path parameters appear directly in the URL (like /users/abc). This method checks the expected type from the spec. If the spec says "integer", it sends strings and floats that will break type casting. If it says "string", it sends injection payloads and boundary values.
- Save fuzzer.py. You will not see a change yet. The next chunk adds the body mutation methods.
- Below mutate_path_param, add the body mutation method and its two helper methods to complete the MutationEngine class:
def mutate_body(self, schema):
"""Generate mutated request bodies based on a JSON schema."""
mutations = []
properties = schema.get("properties", {})
required_fields = schema.get("required", [])
# Strategy 1: Send each field with boundary/injection values
for field_name, field_schema in properties.items():
field_type = field_schema.get("type", "string")
payloads = self._get_payloads_for_type(field_type)
for payload in payloads:
body = self._build_valid_body(properties)
body[field_name] = payload
mutations.append(body)
# Strategy 2: Omit each required field
for field_name in required_fields:
body = self._build_valid_body(properties)
del body[field_name]
mutations.append(body)
# Strategy 3: Add unexpected fields
body = self._build_valid_body(properties)
body["__admin"] = True
body["role"] = "admin"
body["extra_field"] = "A" * 5000
mutations.append(body)
# Strategy 4: Send completely wrong types as body
mutations.append(None)
mutations.append([])
mutations.append("just a string")
mutations.append(12345)
return mutations
def _get_payloads_for_type(self, field_type):
"""Get mutation payloads appropriate for a field type."""
payloads = []
if field_type == "string":
payloads.extend(self.BOUNDARY_STRINGS)
payloads.extend(self.INJECTION_PAYLOADS)
payloads.extend(self.TYPE_CONFUSION)
elif field_type == "integer" or field_type == "number":
payloads.extend(self.BOUNDARY_INTEGERS)
payloads.extend(self.BOUNDARY_STRINGS[:5])
payloads.extend([None, True, [], {}])
else:
payloads.extend(self.BOUNDARY_STRINGS)
payloads.extend(self.TYPE_CONFUSION)
return payloads
def _build_valid_body(self, properties):
"""Build a minimal valid request body."""
body = {}
for field_name, field_schema in properties.items():
field_type = field_schema.get("type", "string")
if field_type == "string":
body[field_name] = "fuzztest"
elif field_type == "integer":
body[field_name] = 1
elif field_type == "number":
body[field_name] = 1.0
elif field_type == "boolean":
body[field_name] = True
else:
body[field_name] = "fuzztest"
return bodyWhat do these methods do?
- mutate_body() applies four strategies: inject bad values into each field, omit required fields, add unexpected fields with privilege-escalation names, and send completely wrong body types (None, a list, a raw string, an integer).
- _get_payloads_for_type() picks the right set of attack payloads based on what the schema says the field type should be.
- _build_valid_body() creates a baseline valid request body so you can mutate one field at a time while keeping the rest valid. This isolates which field caused a crash.
- Save fuzzer.py.
- Verify the mutations work by running this command:
python -c "from fuzzer import MutationEngine; m = MutationEngine(); print(f'Path param mutations (integer): {len(m.mutate_path_param(\"integer\"))}'); print(f'Path param mutations (string): {len(m.mutate_path_param(\"string\"))}')"You should see counts showing how many mutations each method generates. Integer path params produce around 13 mutations. String path params produce around 22.
Seeing an IndentationError?
- All methods (mutate_path_param, mutate_body, etc.) must be indented one level inside the MutationEngine class. They should start with 4 spaces.
- Make sure the class constants (BOUNDARY_STRINGS, etc.) are also indented inside the class, at the same level as the methods.
Help me fix my MutationEngine indentation.
Wire the parser output into the mutation engine
The parser extracts fuzz targets with parameter type information. The mutation engine generates payloads based on those types. Now you will connect them and see how many test cases the fuzzer generates for your target API.
- Run this command to parse the spec and generate mutations for every endpoint:
python -c "
from fuzzer import parse_openapi_spec, MutationEngine
targets = parse_openapi_spec('openapi.yaml')
engine = MutationEngine()
total = 0
for t in targets:
count = 0
for p in t['path_params']:
count += len(engine.mutate_path_param(p['type']))
if t['body_schema']:
count += len(engine.mutate_body(t['body_schema']))
total += count
print(f\"{t['method']} {t['path']}: {count} test cases\")
print(f\"\nTotal test cases: {total}\")
"You should see each endpoint with its test case count, and a total in the hundreds. The endpoints with request bodies (POST /users and POST /search) generate the most mutations because they have multiple fields to fuzz.
Seeing 0 test cases for an endpoint?
- The GET /admin endpoint has no path params and no request body, so it correctly shows 0 test cases for now. Auth-based mutations will come in a later step.
- If other endpoints show 0, check that your openapi.yaml has the correct structure under paths with parameters and requestBody sections.
Help me debug why my fuzzer generates 0 test cases.
✔️ Awesome, I've got everything!
Great. Double check you have saved fuzzer.py with all the code from this step.
ⓧ I'd like to double check the full code
Here is the complete fuzzer.py so far. Make sure yours matches:
import json
import time
import sys
from datetime import datetime, timezone
import requests
import yaml
# --- OpenAPI Spec Parser ---
def parse_openapi_spec(spec_path):
"""Parse an OpenAPI YAML spec and extract fuzz targets."""
with open(spec_path, "r") as f:
spec = yaml.safe_load(f)
base_url = spec["servers"][0]["url"]
fuzz_targets = []
for path, methods in spec.get("paths", {}).items():
for method, details in methods.items():
if method not in ("get", "post", "put", "delete", "patch"):
continue
target = {
"path": path,
"method": method.upper(),
"base_url": base_url,
"path_params": [],
"body_schema": None,
"expected_statuses": [],
"requires_auth": False,
}
# Extract path parameters
for param in details.get("parameters", []):
if param.get("in") == "path":
target["path_params"].append({
"name": param["name"],
"type": param.get("schema", {}).get("type", "string"),
})
# Extract request body schema
request_body = details.get("requestBody", {})
content = request_body.get("content", {})
json_content = content.get("application/json", {})
if json_content:
target["body_schema"] = json_content.get("schema", {})
# Extract expected status codes
for status_code in details.get("responses", {}).keys():
target["expected_statuses"].append(str(status_code))
# Check if auth is required
if details.get("security"):
target["requires_auth"] = True
fuzz_targets.append(target)
return fuzz_targets
# --- Mutation Strategies ---
class MutationEngine:
"""Generates mutated payloads for fuzzing."""
BOUNDARY_STRINGS = [
"",
" ",
"a" * 10000,
"null",
"undefined",
"0",
"-1",
"99999999999999999999",
"\x00",
"\n\r\n",
]
BOUNDARY_INTEGERS = [
0, -1, -999999999, 999999999, 2147483647, -2147483648,
]
TYPE_CONFUSION = [
None,
True,
False,
[],
{},
[None],
{"__proto__": "polluted"},
12345,
3.14159,
]
INJECTION_PAYLOADS = [
"' OR 1=1--",
"'; DROP TABLE users;--",
"1; SELECT * FROM users",
"UNION SELECT NULL,NULL,NULL--",
"; ls -la",
"| cat /etc/passwd",
"$(whoami)",
"`id`",
"<script>alert(1)</script>",
"<img src=x onerror=alert(1)>",
"{{7*7}}",
"${7*7}",
]
def mutate_path_param(self, param_type):
"""Generate mutations for a path parameter."""
mutations = []
if param_type == "integer":
mutations.extend([str(v) for v in self.BOUNDARY_INTEGERS])
mutations.extend(["abc", "", "null", "1.5", "1e10", "-0", "0x1A"])
else:
mutations.extend(self.BOUNDARY_STRINGS)
mutations.extend(self.INJECTION_PAYLOADS)
return mutations
def mutate_body(self, schema):
"""Generate mutated request bodies based on a JSON schema."""
mutations = []
properties = schema.get("properties", {})
required_fields = schema.get("required", [])
# Strategy 1: Send each field with boundary/injection values
for field_name, field_schema in properties.items():
field_type = field_schema.get("type", "string")
payloads = self._get_payloads_for_type(field_type)
for payload in payloads:
body = self._build_valid_body(properties)
body[field_name] = payload
mutations.append(body)
# Strategy 2: Omit each required field
for field_name in required_fields:
body = self._build_valid_body(properties)
del body[field_name]
mutations.append(body)
# Strategy 3: Add unexpected fields
body = self._build_valid_body(properties)
body["__admin"] = True
body["role"] = "admin"
body["extra_field"] = "A" * 5000
mutations.append(body)
# Strategy 4: Send completely wrong types as body
mutations.append(None)
mutations.append([])
mutations.append("just a string")
mutations.append(12345)
return mutations
def _get_payloads_for_type(self, field_type):
"""Get mutation payloads appropriate for a field type."""
payloads = []
if field_type == "string":
payloads.extend(self.BOUNDARY_STRINGS)
payloads.extend(self.INJECTION_PAYLOADS)
payloads.extend(self.TYPE_CONFUSION)
elif field_type == "integer" or field_type == "number":
payloads.extend(self.BOUNDARY_INTEGERS)
payloads.extend(self.BOUNDARY_STRINGS[:5])
payloads.extend([None, True, [], {}])
else:
payloads.extend(self.BOUNDARY_STRINGS)
payloads.extend(self.TYPE_CONFUSION)
return payloads
def _build_valid_body(self, properties):
"""Build a minimal valid request body."""
body = {}
for field_name, field_schema in properties.items():
field_type = field_schema.get("type", "string")
if field_type == "string":
body[field_name] = "fuzztest"
elif field_type == "integer":
body[field_name] = 1
elif field_type == "number":
body[field_name] = 1.0
elif field_type == "boolean":
body[field_name] = True
else:
body[field_name] = "fuzztest"
return bodyYour fuzzer now has a brain that understands your API's structure and hundreds of attack payloads ready to fire. Next up, you will build the request sender and response classifier that actually launches these mutations and reports what breaks.
Build the Fuzzer - Request Sender and Response Classifier
Your fuzzer can parse an OpenAPI spec and generate hundreds of mutated payloads. Now it needs the logic to actually fire those payloads at a live API and make sense of the responses.
A fuzzer that sends requests without understanding the responses is just a load tester. The real value comes from classifying each response into severity categories so you can tell a crash apart from normal behavior.
In this step, get ready to:
- Write a response classifier that sorts API responses into severity categories.
- Build the request sender that fires mutated payloads and handles network errors.
- Generate a JSON findings report and run the full fuzzer against your target API.
Write the response classifier
When the fuzzer sends a mutated request, the API might crash (500 error), return an unexpected status code, time out, or leak error details in the response body. Your classifier needs to detect each of these scenarios and tag them with a severity level.
- In fuzzer.py, scroll to the bottom of the MutationEngine class.
- Add the following ResponseClassifier class below it:
# --- Response Classifier ---
class ResponseClassifier:
"""Classifies API responses into severity categories."""
SEVERITY_CRASH = "CRASH"
SEVERITY_UNEXPECTED = "UNEXPECTED_STATUS"
SEVERITY_INTERESTING = "INTERESTING"
SEVERITY_TIMEOUT = "TIMEOUT"
ERROR_SIGNATURES = [
"traceback",
"exception",
"error",
"stack trace",
"internal server error",
"syntax error",
"database error",
]
def classify(self, response, expected_statuses, payload_description):
# No response means the request timed out
if response is None:
return {
"severity": self.SEVERITY_TIMEOUT,
"description": "Request timed out",
"payload": payload_description,
"status_code": None,
"response_snippet": None,
}
status = response.status_code
body_text = response.text[:500] if response.text else ""
# 5xx = server crashed
if status >= 500:
return {
"severity": self.SEVERITY_CRASH,
"description": f"Server returned {status}",
"payload": payload_description,
"status_code": status,
"response_snippet": body_text,
}
# Status code not listed in the OpenAPI spec
if str(status) not in expected_statuses:
body_lower = body_text.lower()
has_error_sig = any(
sig in body_lower for sig in self.ERROR_SIGNATURES
)
if has_error_sig:
return {
"severity": self.SEVERITY_INTERESTING,
"description": f"Unexpected status {status} with error signature in body",
"payload": payload_description,
"status_code": status,
"response_snippet": body_text,
}
return {
"severity": self.SEVERITY_UNEXPECTED,
"description": f"Unexpected status code: {status}",
"payload": payload_description,
"status_code": status,
"response_snippet": body_text,
}
# Response looks normal, no finding
return NoneWhat does this code do?
The classifier inspects each HTTP response and decides whether something went wrong:
- CRASH catches any 5xx status code, meaning the server hit an unhandled exception.
- UNEXPECTED_STATUS flags responses with status codes not listed in the OpenAPI spec.
- INTERESTING is a special case of unexpected status where the body contains error keywords like "traceback" or "database error".
- TIMEOUT handles cases where the server never responded at all.
- Save fuzzer.py.
You won't see a change yet. The next chunk will use this classifier to process live responses.
Write the request sender
The request sender is the core orchestrator. It reads the parsed spec targets, generates mutations for each one, fires them at the API, and feeds every response through the classifier you just wrote.
- In fuzzer.py, add the following APIFuzzer class below the ResponseClassifier class:
# --- Fuzzer Core ---
class APIFuzzer:
"""Main fuzzer that orchestrates parsing, mutation, and classification."""
def __init__(self, spec_path, delay=0.05):
self.spec_path = spec_path
self.delay = delay
self.mutation_engine = MutationEngine()
self.classifier = ResponseClassifier()
self.findings = []
self.requests_sent = 0
def run(self):
"""Run the fuzzer against all targets in the spec."""
print(f"[*] Parsing OpenAPI spec: {self.spec_path}")
targets = parse_openapi_spec(self.spec_path)
print(f"[*] Found {len(targets)} endpoints to fuzz\n")
for target in targets:
self._fuzz_target(target)
self._print_summary()
self._save_report()
def _fuzz_target(self, target):
"""Fuzz a single endpoint with all applicable mutations."""
path = target["path"]
method = target["method"]
base_url = target["base_url"]
print(f"[*] Fuzzing {method} {path}...")
# Fuzz path parameters
for param in target["path_params"]:
mutations = self.mutation_engine.mutate_path_param(param["type"])
for mutation in mutations:
url = base_url + path.replace(
"{" + param["name"] + "}", str(mutation)
)
self._send_and_classify(
method, url, None, target["expected_statuses"],
f"path_param[{param['name']}]={mutation!r}"
)
# Fuzz request body
if target["body_schema"]:
mutations = self.mutation_engine.mutate_body(target["body_schema"])
url = base_url + path
for mutation in mutations:
self._send_and_classify(
method, url, mutation, target["expected_statuses"],
f"body={json.dumps(mutation)[:200]}"
)
def _send_and_classify(self, method, url, body, expected_statuses, payload_desc):
"""Send a request and classify the response."""
self.requests_sent += 1
response = None
try:
if method == "GET":
response = requests.get(url, timeout=5)
elif method == "POST":
response = requests.post(url, json=body, timeout=5)
elif method == "PUT":
response = requests.put(url, json=body, timeout=5)
elif method == "DELETE":
response = requests.delete(url, timeout=5)
except requests.exceptions.Timeout:
response = None
except requests.exceptions.ConnectionError:
return # Target is down, skip
finding = self.classifier.classify(response, expected_statuses, payload_desc)
if finding:
finding["endpoint"] = f"{method} {url}"
finding["timestamp"] = datetime.now(timezone.utc).isoformat()
self.findings.append(finding)
time.sleep(self.delay)What does this code do?
- __init__ sets up the fuzzer with a 50ms delay between requests to avoid flooding the target.
- _fuzz_target loops through each endpoint's path parameters and body schema, generating mutations for each and sending them one by one.
- _send_and_classify fires a single HTTP request, catches timeouts and connection errors gracefully, then passes the response to the classifier.
- Any finding returned by the classifier gets enriched with the endpoint URL and a timestamp before being stored.
- Save fuzzer.py.
You won't see a change yet. The next chunk adds the report generation and entry point so you can run the fuzzer.
Why a 50ms delay between requests?
Without a delay, the fuzzer would blast hundreds of requests per second. That can crash a local Flask server before you get useful data. The small pause gives the target time to respond cleanly to each request, producing more accurate severity classifications.
Generate the findings report and run the fuzzer
The fuzzer needs two final pieces: a summary that prints results to the terminal, and a report saver that writes all findings to a JSON file. You also need the entry point so you can run the script.
- In fuzzer.py, add the following methods inside the APIFuzzer class, below _send_and_classify:
def _print_summary(self):
"""Print a summary of findings to the terminal."""
print(f"\n{'='*60}")
print(f"FUZZING COMPLETE")
print(f"{'='*60}")
print(f"Requests sent: {self.requests_sent}")
print(f"Issues found: {len(self.findings)}")
if self.findings:
severity_counts = {}
for f in self.findings:
sev = f["severity"]
severity_counts[sev] = severity_counts.get(sev, 0) + 1
print(f"\nBreakdown by severity:")
for sev, count in sorted(severity_counts.items()):
print(f" {sev}: {count}")
print(f"\nTop findings:")
for finding in self.findings[:5]:
print(f" [{finding['severity']}] {finding['endpoint']}")
print(f" Payload: {finding['payload'][:80]}")
print()
def _save_report(self):
"""Save findings to a JSON report file."""
report = {
"scan_metadata": {
"spec_file": self.spec_path,
"timestamp": datetime.now(timezone.utc).isoformat(),
"requests_sent": self.requests_sent,
"issues_found": len(self.findings),
},
"findings": self.findings,
}
with open("findings.json", "w") as f:
json.dump(report, f, indent=2)
print(f"\n[*] Full report saved to findings.json")
if __name__ == "__main__":
spec_file = "openapi.yaml"
if len(sys.argv) > 1:
spec_file = sys.argv[1]
fuzzer = APIFuzzer(spec_file)
fuzzer.run()What does this code do?
- _print_summary counts findings by severity and displays the top 5 in the terminal so you get immediate feedback.
- _save_report writes a structured JSON file containing scan metadata (timestamp, total requests) and every individual finding with its payload and response snippet.
- The __main__ block defaults to openapi.yaml but accepts an optional CLI argument to point at a different spec file.
- Save fuzzer.py.
Now it is time to run the fuzzer against your target API. You need two terminal windows: one for the API server and one for the fuzzer.
- In your first terminal, start the target API by running:
python target_api.py- Open a second terminal window and make sure you are inside your api-fuzzer/ directory with the virtual environment active.
- Run the fuzzer by executing:
python fuzzer.pyThe fuzzer takes a few seconds to complete. You should see output showing each endpoint being fuzzed, followed by a summary with a breakdown of findings by severity.
What should I see?
The terminal prints lines like [*] Fuzzing GET /users/{user_id}... for each endpoint. At the end, you see a "FUZZING COMPLETE" banner with the total requests sent and a severity breakdown showing CRASH, UNEXPECTED_STATUS, and INTERESTING findings.
You also see [*] Full report saved to findings.json at the bottom.
- Confirm the report was created by running:
ls findings.json- You should see findings.json in the output. Open it to inspect the structured findings:
cat findings.json | python -m json.tool | head -40You should see a JSON object with scan_metadata at the top and a findings array containing objects with severity, endpoint, payload, and response_snippet fields.
Seeing "Connection refused" or no findings?
- Make sure your target API is running in the other terminal. You should see no output in that terminal (Flask is waiting for requests). If it exited, restart it with python target_api.py.
- Check that both terminals are running inside the same api-fuzzer/ directory with the virtual environment active.
- If you see a Python import error, confirm your virtual environment is active (your prompt shows (venv)).
Why is my fuzzer not finding any issues?
✔️ Awesome, I've got everything!
Great. Double check you have saved fuzzer.py and that your findings.json file exists in your project folder.
ⓧ I'd like to double check the full code
Here is the complete fuzzer.py file with all code from this step and the previous step combined:
import json
import time
import sys
from datetime import datetime, timezone
import requests
import yaml
# --- OpenAPI Spec Parser ---
def parse_openapi_spec(spec_path):
"""Parse an OpenAPI YAML spec and extract fuzz targets."""
with open(spec_path, "r") as f:
spec = yaml.safe_load(f)
base_url = spec["servers"][0]["url"]
fuzz_targets = []
for path, methods in spec.get("paths", {}).items():
for method, details in methods.items():
if method not in ("get", "post", "put", "delete", "patch"):
continue
target = {
"path": path,
"method": method.upper(),
"base_url": base_url,
"path_params": [],
"body_schema": None,
"expected_statuses": [],
"requires_auth": False,
}
for param in details.get("parameters", []):
if param.get("in") == "path":
target["path_params"].append({
"name": param["name"],
"type": param.get("schema", {}).get("type", "string"),
})
request_body = details.get("requestBody", {})
content = request_body.get("content", {})
json_content = content.get("application/json", {})
if json_content:
target["body_schema"] = json_content.get("schema", {})
for status_code in details.get("responses", {}).keys():
target["expected_statuses"].append(str(status_code))
if details.get("security"):
target["requires_auth"] = True
fuzz_targets.append(target)
return fuzz_targets
# --- Mutation Strategies ---
class MutationEngine:
"""Generates mutated payloads for fuzzing."""
BOUNDARY_STRINGS = [
"",
" ",
"a" * 10000,
"null",
"undefined",
"0",
"-1",
"99999999999999999999",
"\x00",
"\n\r\n",
]
BOUNDARY_INTEGERS = [
0, -1, -999999999, 999999999, 2147483647, -2147483648,
]
TYPE_CONFUSION = [
None,
True,
False,
[],
{},
[None],
{"__proto__": "polluted"},
12345,
3.14159,
]
INJECTION_PAYLOADS = [
"' OR 1=1--",
"'; DROP TABLE users;--",
"1; SELECT * FROM users",
"UNION SELECT NULL,NULL,NULL--",
"; ls -la",
"| cat /etc/passwd",
"$(whoami)",
"`id`",
"<script>alert(1)</script>",
"<img src=x onerror=alert(1)>",
"{{7*7}}",
"${7*7}",
]
def mutate_path_param(self, param_type):
mutations = []
if param_type == "integer":
mutations.extend([str(v) for v in self.BOUNDARY_INTEGERS])
mutations.extend(["abc", "", "null", "1.5", "1e10", "-0", "0x1A"])
else:
mutations.extend(self.BOUNDARY_STRINGS)
mutations.extend(self.INJECTION_PAYLOADS)
return mutations
def mutate_body(self, schema):
mutations = []
properties = schema.get("properties", {})
required_fields = schema.get("required", [])
for field_name, field_schema in properties.items():
field_type = field_schema.get("type", "string")
payloads = self._get_payloads_for_type(field_type)
for payload in payloads:
body = self._build_valid_body(properties)
body[field_name] = payload
mutations.append(body)
for field_name in required_fields:
body = self._build_valid_body(properties)
del body[field_name]
mutations.append(body)
body = self._build_valid_body(properties)
body["__admin"] = True
body["role"] = "admin"
body["extra_field"] = "A" * 5000
mutations.append(body)
mutations.append(None)
mutations.append([])
mutations.append("just a string")
mutations.append(12345)
return mutations
def _get_payloads_for_type(self, field_type):
payloads = []
if field_type == "string":
payloads.extend(self.BOUNDARY_STRINGS)
payloads.extend(self.INJECTION_PAYLOADS)
payloads.extend(self.TYPE_CONFUSION)
elif field_type == "integer" or field_type == "number":
payloads.extend(self.BOUNDARY_INTEGERS)
payloads.extend(self.BOUNDARY_STRINGS[:5])
payloads.extend([None, True, [], {}])
else:
payloads.extend(self.BOUNDARY_STRINGS)
payloads.extend(self.TYPE_CONFUSION)
return payloads
def _build_valid_body(self, properties):
body = {}
for field_name, field_schema in properties.items():
field_type = field_schema.get("type", "string")
if field_type == "string":
body[field_name] = "fuzztest"
elif field_type == "integer":
body[field_name] = 1
elif field_type == "number":
body[field_name] = 1.0
elif field_type == "boolean":
body[field_name] = True
else:
body[field_name] = "fuzztest"
return body
# --- Response Classifier ---
class ResponseClassifier:
"""Classifies API responses into severity categories."""
SEVERITY_CRASH = "CRASH"
SEVERITY_UNEXPECTED = "UNEXPECTED_STATUS"
SEVERITY_INTERESTING = "INTERESTING"
SEVERITY_TIMEOUT = "TIMEOUT"
ERROR_SIGNATURES = [
"traceback",
"exception",
"error",
"stack trace",
"internal server error",
"syntax error",
"database error",
]
def classify(self, response, expected_statuses, payload_description):
if response is None:
return {
"severity": self.SEVERITY_TIMEOUT,
"description": "Request timed out",
"payload": payload_description,
"status_code": None,
"response_snippet": None,
}
status = response.status_code
body_text = response.text[:500] if response.text else ""
if status >= 500:
return {
"severity": self.SEVERITY_CRASH,
"description": f"Server returned {status}",
"payload": payload_description,
"status_code": status,
"response_snippet": body_text,
}
if str(status) not in expected_statuses:
body_lower = body_text.lower()
has_error_sig = any(
sig in body_lower for sig in self.ERROR_SIGNATURES
)
if has_error_sig:
return {
"severity": self.SEVERITY_INTERESTING,
"description": f"Unexpected status {status} with error signature in body",
"payload": payload_description,
"status_code": status,
"response_snippet": body_text,
}
return {
"severity": self.SEVERITY_UNEXPECTED,
"description": f"Unexpected status code: {status}",
"payload": payload_description,
"status_code": status,
"response_snippet": body_text,
}
return None
# --- Fuzzer Core ---
class APIFuzzer:
"""Main fuzzer that orchestrates parsing, mutation, and classification."""
def __init__(self, spec_path, delay=0.05):
self.spec_path = spec_path
self.delay = delay
self.mutation_engine = MutationEngine()
self.classifier = ResponseClassifier()
self.findings = []
self.requests_sent = 0
def run(self):
print(f"[*] Parsing OpenAPI spec: {self.spec_path}")
targets = parse_openapi_spec(self.spec_path)
print(f"[*] Found {len(targets)} endpoints to fuzz\n")
for target in targets:
self._fuzz_target(target)
self._print_summary()
self._save_report()
def _fuzz_target(self, target):
path = target["path"]
method = target["method"]
base_url = target["base_url"]
print(f"[*] Fuzzing {method} {path}...")
for param in target["path_params"]:
mutations = self.mutation_engine.mutate_path_param(param["type"])
for mutation in mutations:
url = base_url + path.replace(
"{" + param["name"] + "}", str(mutation)
)
self._send_and_classify(
method, url, None, target["expected_statuses"],
f"path_param[{param['name']}]={mutation!r}"
)
if target["body_schema"]:
mutations = self.mutation_engine.mutate_body(target["body_schema"])
url = base_url + path
for mutation in mutations:
self._send_and_classify(
method, url, mutation, target["expected_statuses"],
f"body={json.dumps(mutation)[:200]}"
)
def _send_and_classify(self, method, url, body, expected_statuses, payload_desc):
self.requests_sent += 1
response = None
try:
if method == "GET":
response = requests.get(url, timeout=5)
elif method == "POST":
response = requests.post(url, json=body, timeout=5)
elif method == "PUT":
response = requests.put(url, json=body, timeout=5)
elif method == "DELETE":
response = requests.delete(url, timeout=5)
except requests.exceptions.Timeout:
response = None
except requests.exceptions.ConnectionError:
return
finding = self.classifier.classify(response, expected_statuses, payload_desc)
if finding:
finding["endpoint"] = f"{method} {url}"
finding["timestamp"] = datetime.now(timezone.utc).isoformat()
self.findings.append(finding)
time.sleep(self.delay)
def _print_summary(self):
print(f"\n{'='*60}")
print(f"FUZZING COMPLETE")
print(f"{'='*60}")
print(f"Requests sent: {self.requests_sent}")
print(f"Issues found: {len(self.findings)}")
if self.findings:
severity_counts = {}
for f in self.findings:
sev = f["severity"]
severity_counts[sev] = severity_counts.get(sev, 0) + 1
print(f"\nBreakdown by severity:")
for sev, count in sorted(severity_counts.items()):
print(f" {sev}: {count}")
print(f"\nTop findings:")
for finding in self.findings[:5]:
print(f" [{finding['severity']}] {finding['endpoint']}")
print(f" Payload: {finding['payload'][:80]}")
print()
def _save_report(self):
report = {
"scan_metadata": {
"spec_file": self.spec_path,
"timestamp": datetime.now(timezone.utc).isoformat(),
"requests_sent": self.requests_sent,
"issues_found": len(self.findings),
},
"findings": self.findings,
}
with open("findings.json", "w") as f:
json.dump(report, f, indent=2)
print(f"\n[*] Full report saved to findings.json")
if __name__ == "__main__":
spec_file = "openapi.yaml"
if len(sys.argv) > 1:
spec_file = sys.argv[1]
fuzzer = APIFuzzer(spec_file)
fuzzer.run()Your fuzzer is complete. It reads the spec, generates mutations, fires them, classifies the responses, and saves a structured report. Next up, you will extend it with authentication bypass detection in the Secret Mission.
Secret mission
Add Authentication Bypass Detection
Your fuzzer catches crashes and unexpected responses, but it misses one of the most dangerous vulnerability classes: broken access control. Can you extend your fuzzer to detect endpoints that should require authentication but happily serve data to anyone who asks?
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally. There are no cloud resources, no paid APIs, and no ongoing costs.
Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- The api-fuzzer/ project folder on your Desktop (contains all source files, the OpenAPI spec, and the generated findings report).
- A Python virtual environment (venv/) inside the project folder.
- Pip packages installed inside the virtual environment (Flask, requests, PyYAML).
Keep everything running
No action needed. Choose this if you are still actively building or want to extend the fuzzer with more mutation strategies.
- Your project folder stays on your Desktop as-is.
- You can reactivate the virtual environment any time and run the fuzzer again.
- Consider adding more mutation strategies, supporting additional HTTP methods, or integrating the fuzzer into a CI pipeline.
Pause - I'll come back to this later
Shut down any running processes and deactivate the virtual environment to free up your terminal.
- If the Flask target API is still running in a terminal, press Ctrl+C to stop it.
- Deactivate your virtual environment by running:
deactivate- Your project files remain on your Desktop. When you are ready to return, navigate back to api-fuzzer/ and reactivate the venv.
Delete - I don't want to use this again
Remove all project files and start fresh. This deletes your fuzzer code, the target API, and the findings report permanently.
- If the Flask target API is still running, press Ctrl+C to stop it.
- Deactivate your virtual environment by running:
deactivate- Navigate to your Desktop and delete the entire project folder.
macOS
- Delete the project folder by running:
cd ~/Desktop
rm -rf api-fuzzer/Windows
- Delete the project folder by running:
cd ~/Desktop
rmdir /s /q api-fuzzer- Confirm the folder is gone by running ls on your Desktop. You should no longer see api-fuzzer/ in the output.
Nice Work!
Nice Work!
You just built a custom API security fuzzer from scratch in Python. This is a real offensive security tool that parses specifications, generates mutated inputs, and reports vulnerabilities automatically.
Here is what you accomplished:
- Built a deliberately vulnerable Flask REST API with 4 endpoints containing planted security flaws, paired with an OpenAPI specification that formally describes expected behavior.
- Built a custom Python fuzzer engine with 4 mutation strategies (boundary values, type confusion, injection payloads, and missing/extra fields) that generates hundreds of test cases from an OpenAPI spec.
- Classified API responses into severity categories and generated a structured JSON findings report with crash detection, unexpected status flagging, and error signature matching.
- Secret Mission: added authentication bypass detection that catches broken access control on protected endpoints.
Ready to quiz yourself?