Adding Payload Collections
Adding Payload Collections
The following files were used as context for generating this wiki page:
Purpose: This document provides a guide for developers to add new payload collections to WSHawk's payload database. It covers two approaches: adding static payload files to the core payload system, and creating extensible payload plugins.
For information about using existing payloads during scans, see Vulnerability Detection Overview. For information about creating detector plugins that work with payloads, see Creating Detector Plugins.
Overview
WSHawk supports two methods for adding payload collections:
| Method | Use Case | Location | Requires Code Changes |
|--------|----------|----------|----------------------|
| Static Payload Files | Core payload types that ship with WSHawk | wshawk/payloads/*.txt | Yes - add getter method |
| Payload Plugins | Custom payload packs for specific frameworks or testing scenarios | plugins/*.py | No - auto-discovered |
The static file approach integrates payloads directly into the WSPayloads class with lazy loading and caching. The plugin approach provides extensibility without modifying core code.
Sources: wshawk/main.py:61-143, wshawk/plugin_system.py:52-63
Static Payload File Architecture
graph TB
subgraph "Payload Storage"
PayloadFiles["wshawk/payloads/<br/>sql_injection.txt<br/>xss.txt<br/>command_injection.txt<br/>nosql_injection.txt<br/>path_traversal.txt<br/>ldap_injection.txt<br/>xxe.txt<br/>ssti.txt<br/>open_redirect.txt<br/>csv_injection.txt"]
end
subgraph "WSPayloads Class (wshawk/__main__.py:61-143)"
LoadMethod["load_payloads(filename)<br/>Line 70-102"]
Cache["_payloads_cache<br/>Dict[str, List[str]]<br/>Line 67"]
Getters["Getter Methods:<br/>get_sql_injection() → Line 105<br/>get_xss() → Line 109<br/>get_command_injection() → Line 113<br/>get_nosql_injection() → Line 117<br/>get_path_traversal() → Line 121<br/>get_ldap_injection() → Line 125<br/>get_xxe() → Line 129<br/>get_ssti() → Line 133<br/>get_open_redirect() → Line 137<br/>get_csv_injection() → Line 141"]
end
subgraph "Import Mechanisms"
ImportLibResources["importlib.resources<br/>Lines 78-87<br/>For pip-installed packages"]
FilesystemFallback["Filesystem Fallback<br/>Lines 89-92<br/>For development/source installs"]
end
subgraph "Scanner Usage"
TestMethods["test_sql_injection()<br/>test_xss()<br/>test_command_injection()<br/>etc."]
end
PayloadFiles --> LoadMethod
LoadMethod --> ImportLibResources
ImportLibResources --> |"Success"| Cache
ImportLibResources --> |"Fallback"| FilesystemFallback
FilesystemFallback --> Cache
Getters --> |"Calls"| LoadMethod
Cache --> Getters
Getters --> TestMethods
The WSPayloads class provides a centralized payload management system with lazy loading and caching. Each payload file is loaded once and cached in the _payloads_cache dictionary.
Sources: wshawk/main.py:61-143
Adding a New Static Payload File
Step 1: Create the Payload File
Create a new text file in the wshawk/payloads/ directory:
wshawk/payloads/jwt_injection.txt
File Format Requirements:
- Plain text, one payload per line
- UTF-8 encoding
- Empty lines are automatically filtered
- Leading/trailing whitespace is stripped
- Comments not supported (all lines treated as payloads)
Example payload file content:
{"alg":"none"}
eyJhbGciOiJub25lIn0
{"typ":"JWT","alg":"HS256"}
{"kid":"../../etc/passwd"}
{"jku":"http://attacker.com/jwks.json"}
Sources: wshawk/main.py:70-102
Step 2: Add Getter Method to WSPayloads
Edit wshawk/__main__.py and add a new static method to the WSPayloads class:
@staticmethod
def get_jwt_injection():
return WSPayloads.load_payloads('jwt_injection.txt')
This method should be added after the existing getter methods (after line 142).
Naming Convention:
- Method name:
get_<vulnerability_type>() - File name:
<vulnerability_type>.txt - Use underscores for multi-word types
Sources: wshawk/main.py:105-142
Step 3: Register with Package Manifest
Ensure the payload file is included in the package distribution by verifying it's covered by the MANIFEST.in patterns:
recursive-include wshawk/payloads *.txt
This pattern should already exist and will automatically include your new payload file.
Payload Loading Mechanism Deep Dive
sequenceDiagram
participant Scanner as "Scanner Method"
participant Getter as "WSPayloads.get_xss()"
participant Loader as "WSPayloads.load_payloads()"
participant Cache as "_payloads_cache"
participant ImportLib as "importlib.resources"
participant Filesystem as "File System"
Scanner->>Getter: Request payloads
Getter->>Loader: load_payloads('xss.txt')
Loader->>Cache: Check if cached
alt Payload Cached
Cache-->>Loader: Return cached payloads
Loader-->>Getter: Return payloads
else Not Cached
Loader->>ImportLib: Try importlib.resources
alt Pip Install (importlib succeeds)
ImportLib-->>Loader: Read from package resources
else Development Install (importlib fails)
ImportLib--xFilesystem: FileNotFoundError
Loader->>Filesystem: Read from wshawk/payloads/
Filesystem-->>Loader: File contents
end
Loader->>Loader: Strip whitespace & filter empty
Loader->>Cache: Store in _payloads_cache
Loader-->>Getter: Return payloads
end
Getter-->>Scanner: Return payloads
The dual-loading strategy ensures payloads work both when WSHawk is installed via pip (using importlib.resources) and when running from source (using filesystem paths).
Key Implementation Details:
| Aspect | Implementation | Location |
|--------|---------------|----------|
| Cache Check | if filename in WSPayloads._payloads_cache | Line 72 |
| Pip Install Loading | importlib.resources.files('wshawk').joinpath('payloads', filename) | Lines 80-82 |
| Python 3.8 Compatibility | importlib.resources.read_text() | Lines 85-87 |
| Filesystem Fallback | os.path.join(os.path.dirname(__file__), 'payloads', filename) | Line 90 |
| Whitespace Stripping | [p.strip() for p in payloads if p.strip()] | Line 94 |
| Cache Storage | WSPayloads._payloads_cache[filename] = payloads | Line 95 |
Sources: wshawk/main.py:70-102
Using New Payloads in Scanner Methods
After adding a payload collection, integrate it into a scanner test method:
Example: Using JWT Injection Payloads
async def test_jwt_injection(self) -> List[Dict]:
"""Test for JWT manipulation vulnerabilities"""
all_payloads = WSPayloads.get_jwt_injection()
payload_count = len(all_payloads) if self.max_payloads is None else min(self.max_payloads, len(all_payloads))
Logger.info(f"Testing JWT injection ({payload_count}/{len(all_payloads)} payloads)...")
results = []
ws = await self.connect()
if not ws:
return results
try:
payloads = all_payloads[:payload_count]
for payload in payloads:
message = json.dumps({"token": payload, "action": "authenticate"})
response = await self.send_and_receive(ws, message)
if response:
jwt_indicators = ['invalid signature', 'algorithm none', 'jwt malformed']
if any(ind in response.lower() for ind in jwt_indicators):
Logger.vuln(f"JWT injection detected: {payload[:50]}")
self.vulnerabilities.append({
'type': 'JWT Injection',
'severity': 'HIGH',
'description': 'JWT manipulation vulnerability detected',
'payload': payload,
'recommendation': 'Validate JWT signatures and algorithms'
})
results.append({'payload': payload, 'vulnerable': True})
await asyncio.sleep(0.1)
await ws.close()
except Exception as e:
Logger.error(f"JWT injection test error: {e}")
return results
This pattern matches the existing vulnerability test methods in wshawk/main.py:347-744.
Sources: wshawk/main.py:347-397, wshawk/main.py:399-438
Alternative: Creating Payload Plugins
For custom payload collections that don't belong in the core distribution, use the plugin system:
Plugin Architecture
graph TB
subgraph "Plugin Manager (wshawk/plugin_system.py)"
PluginMgr["PluginManager<br/>Lines 98-438"]
Registry["_payload_plugins<br/>Dict[str, PayloadPlugin]<br/>Line 114"]
LoadLazy["_load_plugin_lazy()<br/>Lines 158-201"]
GetPayloads["get_payloads(vuln_type, plugin_name)<br/>Lines 335-373"]
end
subgraph "Plugin Base Classes"
PayloadPlugin["PayloadPlugin<br/>Lines 52-62"]
PluginBase["PluginBase<br/>Lines 35-50"]
PluginMetadata["PluginMetadata<br/>Lines 17-33"]
end
subgraph "Custom Plugin (plugins/*.py)"
CustomPlugin["class CustomXSSPayloads(PayloadPlugin)<br/>Lines 443-465"]
GetMeta["get_metadata() → PluginMetadata<br/>Lines 446-453"]
GetPayloadsImpl["get_payloads(vuln_type) → List[str]<br/>Lines 455-465"]
end
subgraph "Scanner Integration"
Scanner["Scanner Methods"]
end
PayloadPlugin -.->|"Inherits"| PluginBase
CustomPlugin -.->|"Implements"| PayloadPlugin
CustomPlugin --> GetMeta
CustomPlugin --> GetPayloadsImpl
PluginMgr --> LoadLazy
LoadLazy --> |"Validates & Registers"| Registry
Registry --> CustomPlugin
Scanner --> GetPayloads
GetPayloads --> |"Lazy loads if needed"| LoadLazy
GetPayloads --> |"Retrieves from"| Registry
Sources: wshawk/plugin_system.py:52-495
Example Payload Plugin Implementation
Create plugins/graphql_injection.py:
from wshawk.plugin_system import PayloadPlugin, PluginMetadata
from typing import List
import os
class GraphQLInjectionPayloads(PayloadPlugin):
"""Custom GraphQL injection payload pack"""
def get_metadata(self) -> PluginMetadata:
return PluginMetadata(
name="graphql_injection",
version="1.0.0",
description="GraphQL injection and introspection payloads",
author="Your Name",
min_wshawk_version="2.0.0"
)
def get_payloads(self, vuln_type: str) -> List[str]:
if vuln_type == "graphql":
# Option 1: Return hardcoded payloads
return [
"{__schema{types{name}}}",
"{__type(name:\"User\"){fields{name}}}",
"query{__typename}",
"' OR '1'='1",
"{user(id:\"1' OR '1'='1\"){name}}"
]
elif vuln_type == "graphql_dos":
# Option 2: Load from external file
plugin_dir = os.path.dirname(os.path.abspath(__file__))
payload_file = os.path.join(plugin_dir, "graphql_dos_payloads.txt")
if os.path.exists(payload_file):
with open(payload_file, 'r') as f:
return [line.strip() for line in f if line.strip()]
return []
Plugin Metadata Fields:
| Field | Required | Purpose |
|-------|----------|---------|
| name | Yes | Unique plugin identifier |
| version | Yes | Semantic version (major.minor.patch) |
| description | Yes | Human-readable description |
| author | No | Plugin author name |
| requires | No | List of required Python packages |
| min_wshawk_version | No | Minimum WSHawk version (default: "2.0.0") |
Sources: wshawk/plugin_system.py:17-50, wshawk/plugin_system.py:443-465
Using Plugin Payloads in Scans
The plugin manager automatically discovers and lazy-loads plugins:
from wshawk.plugin_system import PluginManager
# Create plugin manager (auto-discovers plugins/ directory)
plugin_manager = PluginManager(plugin_dir="plugins")
# Get payloads from specific plugin
graphql_payloads = plugin_manager.get_payloads("graphql", plugin_name="graphql_injection")
# Get payloads from all plugins
all_graphql_payloads = plugin_manager.get_payloads("graphql")
Lazy Loading Behavior:
- Plugins are scanned but not loaded until first use (wshawk/plugin_system.py:141-156)
- First call to
get_payloads()triggers loading (wshawk/plugin_system.py:350-352) - Payloads are cached using
@lru_cache(maxsize=128)(wshawk/plugin_system.py:335)
Sources: wshawk/plugin_system.py:335-373
Payload File Naming and Organization
Directory Structure
wshawk/
├── payloads/ # Core static payloads
│ ├── sql_injection.txt
│ ├── xss.txt
│ ├── command_injection.txt
│ ├── nosql_injection.txt
│ ├── path_traversal.txt
│ ├── ldap_injection.txt
│ ├── xxe.txt
│ ├── ssti.txt
│ ├── open_redirect.txt
│ └── csv_injection.txt
│
plugins/ # Custom payload plugins
├── README.md
├── graphql_injection.py
└── custom_framework.py
Naming Conventions
| Convention | Rationale | Example |
|------------|-----------|---------|
| Lowercase with underscores | Python module naming | jwt_injection.txt |
| Descriptive vulnerability type | Clear purpose | xml_external_entity.txt not xxe.txt |
| .txt extension for static files | Plain text format | ssti.txt |
| .py extension for plugins | Python modules | graphql_injection.py |
Sources: wshawk/main.py:105-142, plugins/README.md:1-37
Testing New Payload Collections
Validation Checklist
| Aspect | Verification Method | Location |
|--------|-------------------|----------|
| File Loading | Import WSPayloads and call getter method | Python REPL |
| Encoding | Verify UTF-8 encoding, no BOM | Text editor |
| Format | One payload per line, no JSON/XML wrapping | Text editor |
| Caching | Call getter twice, verify second call is instant | Python REPL |
| Package Distribution | Install with pip, verify payloads accessible | pip install -e . |
| Plugin Loading | Check PluginManager.list_plugins() output | Python REPL |
| Plugin Validation | Verify metadata passes validation | Check console output |
Testing Static Payloads
# Test in Python REPL
from wshawk.__main__ import WSPayloads
# Test loading
payloads = WSPayloads.get_jwt_injection()
print(f"Loaded {len(payloads)} JWT injection payloads")
print(f"First payload: {payloads[0]}")
# Test caching
import time
start = time.time()
payloads1 = WSPayloads.get_jwt_injection()
elapsed1 = time.time() - start
start = time.time()
payloads2 = WSPayloads.get_jwt_injection() # Should be cached
elapsed2 = time.time() - start
print(f"First load: {elapsed1:.6f}s, Cached load: {elapsed2:.6f}s")
assert payloads1 is payloads2 # Same object reference
Testing Payload Plugins
from wshawk.plugin_system import PluginManager
# Initialize manager
manager = PluginManager(plugin_dir="plugins")
# List available plugins
plugins = manager.list_plugins()
print(f"Available: {plugins['available']}")
print(f"Loaded: {plugins['loaded']}") # Should be empty (lazy loading)
# Load payloads (triggers lazy loading)
graphql = manager.get_payloads("graphql", plugin_name="graphql_injection")
print(f"Loaded {len(graphql)} GraphQL payloads")
# Verify plugin is now loaded
plugins_after = manager.list_plugins()
print(f"Loaded: {plugins_after['loaded']}") # Should include graphql_injection
# Check metadata
info = manager.get_plugin_info("graphql_injection")
print(f"Plugin version: {info['version']}")
print(f"Min WSHawk: {info['min_wshawk_version']}")
Sources: wshawk/main.py:70-102, wshawk/plugin_system.py:409-438
Best Practices
Payload Quality
| Practice | Rationale | Example |
|----------|-----------|---------|
| Test payload effectiveness | Verify payloads actually trigger vulnerabilities | Use against known-vulnerable test servers |
| Include payload variants | Different contexts require different syntax | Both ' OR '1'='1 and " OR "1"="1 |
| Add WAF evasion techniques | Payloads should work against protected targets | Encoding, case variations, null bytes |
| Document payload sources | Track where payloads originated | Add header comment with attribution |
| Remove duplicates | Reduces scan time without losing coverage | Use sort -u or Python set deduplication |
Performance Considerations
| Consideration | Impact | Mitigation |
|--------------|--------|------------|
| File size | Large files slow initial load | Split into multiple files by subcategory |
| Payload count | More payloads = longer scans | Use max_payloads parameter for quick scans |
| Caching | Prevents repeated file I/O | Automatic via _payloads_cache |
| Lazy plugin loading | Reduces startup time | Automatic via PluginManager |
Security Considerations
Do Not:
- Include payloads that cause actual harm (DoS, data destruction)
- Use payloads with real credentials or sensitive data
- Include payloads targeting specific production systems
Do:
- Use safe, detection-focused payloads
- Test against controlled environments only
- Follow responsible disclosure practices
- Add appropriate severity classifications
Sources: wshawk/main.py:371-387, wshawk/plugin_system.py:203-235
Integration with Scanner Methods
Payload Usage Pattern
All scanner test methods follow this pattern:
graph LR
GetPayloads["1. Get Payloads<br/>WSPayloads.get_xss()"]
LimitPayloads["2. Apply Limit<br/>payloads[:max_payloads]"]
ConnectWS["3. Connect WebSocket<br/>await self.connect()"]
IteratePayloads["4. Iterate Payloads<br/>for payload in payloads"]
SendPayload["5. Send Payload<br/>await send_and_receive()"]
AnalyzeResponse["6. Analyze Response<br/>Check for indicators"]
RecordVuln["7. Record Vulnerability<br/>self.vulnerabilities.append()"]
RateLimit["8. Rate Limit<br/>await asyncio.sleep(0.1)"]
GetPayloads --> LimitPayloads
LimitPayloads --> ConnectWS
ConnectWS --> IteratePayloads
IteratePayloads --> SendPayload
SendPayload --> AnalyzeResponse
AnalyzeResponse --> |"Vulnerable"| RecordVuln
AnalyzeResponse --> |"Continue"| RateLimit
RecordVuln --> RateLimit
RateLimit --> |"Next payload"| IteratePayloads
Key Components:
- Payload Retrieval: wshawk/main.py:351-353 - Gets payloads and respects
max_payloadslimit - Connection Handling: wshawk/main.py:356-358 - Establishes WebSocket connection
- Payload Iteration: wshawk/main.py:364 - Loops through payload list
- Message Sending: wshawk/main.py:366-367 - Injects payload into JSON message
- Vulnerability Detection: wshawk/main.py:371-377 - Analyzes response for indicators
- Recording: wshawk/main.py:379-387 - Adds vulnerability to results
- Rate Limiting: wshawk/main.py:390 - 0.1s delay between payloads
Sources: wshawk/main.py:347-397
Summary
| Method | When to Use | Pros | Cons | |--------|------------|------|------| | Static Files + WSPayloads | Core vulnerability types shipping with WSHawk | - Simple implementation- Fast loading (cached)- Works with pip install | - Requires code changes- Limited to core distribution | | Payload Plugins | Custom frameworks, specialized testing, community contributions | - No core code changes- Auto-discovered- Versioned & validated- Can load from external sources | - Slightly more complex- Requires Python class |
Both approaches support lazy loading, caching, and integration with the scanner's vulnerability testing framework.
Sources: wshawk/main.py:61-143, wshawk/plugin_system.py:52-495