Elasticsearch Query Injection: Detection and Deep Assessment

┌─────────────────────────────────────────────────────────────────┐
│ ELASTICSEARCH INJECTION ATTACK │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 1. Attacker identifies Elasticsearch-backed application │
│ ↓ │
│ 2. Attacker injects Query DSL operators in search parameters │
│ ↓ │
│ 3. Application concatenates input into query without validation │
│ ↓ │
│ 4. Malicious query sent to Elasticsearch │
│ ↓ │
│ 5. Elasticsearch executes unauthorized query │
│ ↓ │
│ 6. Attacker receives sensitive data or confirmation │
│ ↓ │
│ 7. If scripts enabled: RCE via Painless injection │
│ │
└─────────────────────────────────────────────────────────────────┘

# VULNERABLE: String concatenation in query building
def search_users(username):
 query = {
 "query": {
 "match": {
 "username": username # Attacker controls this!
 }
 }
 }
 return es.search(index="users", body=query)

# Attacker input: {"$ne": null}
# Result: Returns ALL users (authentication bypass)

# What Nuclei CAN do:
id: elasticsearch-detect
requests:
 - method: GET
 path:
 - "{{BaseURL}}:9200/"
 matchers:
 - type: word
 words:
 - '"tagline" : "You Know, for Search"'

┌──────────────────────────────────────────────────────────────────┐
│ CERT-X-GEN DETECTION FLOW │
├──────────────────────────────────────────────────────────────────┤
│ │
│ Scanner ──────► Target: GET http://host:9200/ (fingerprint) │
│ │ │
│ ▼ │
│ Scanner: Version detected - Elasticsearch 8.x │
│ │ │
│ ▼ │
│ Scanner ──────► API Test: GET /_cluster/health │
│ │ │
│ ▼ │
│ Scanner: Cluster API accessible (exposed APIs found) │
│ │ │
│ ▼ │
│ Scanner ──────► Injection Test: POST /_all/_search │
│ │ Body: {"query": {"match_all": {}}} │
│ ▼ │
│ Response: Data returned? ───► MEDIUM: Query injection! │
│ │ │
│ ▼ │
│ Scanner ──────► Script Test: POST /_all/_search │
│ │ Body: {"script_fields": {"test": {...}}} │
│ ▼ │
│ Script executed? ───────────► CRITICAL: RCE possible! │
│ Script blocked? ─────────────► Scripts disabled │
│ │
└──────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│ LEGITIMATE QUERY (SAFE) │
├─────────────────────────────────────────────────────────────────┤
│ POST /users/_search │
│ { │
│ "query": { │
│ "match": { │
│ "username": "john" │
│ } │
│ } │
│ } │
│ Result: Returns John's profile only │
└─────────────────────────────────────────────────────────────────┘
 │
 │ Query DSL Injection
 ▼
┌─────────────────────────────────────────────────────────────────┐
│ INJECTED QUERY (MALICIOUS) │
├─────────────────────────────────────────────────────────────────┤
│ POST /users/_search │
│ { │
│ "query": { │
│ "bool": { ◀── Injected operator! │
│ "must": [{"match_all": {}}] ◀── Matches ALL documents! │
│ } │
│ } │
│ } │
│ Result: Returns ALL user profiles AUTH BYPASSED │
└─────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│ PAINLESS SCRIPT INJECTION (RCE) │
├─────────────────────────────────────────────────────────────────┤
│ POST /_all/_search │
│ { │
│ "query": {"match_all": {}}, │
│ "script_fields": { │
│ "exploit": { │
│ "script": { │
│ "source": "1+1", ◀── Benign test │
│ "lang": "painless" │
│ } │
│ } │
│ } │
│ } │
│ │
│ If script executes → Painless is enabled → RCE POSSIBLE! │
│ Advanced: "Runtime.getRuntime().exec('calc')" │
└─────────────────────────────────────────────────────────────────┘

def test_query_dsl_injection(host, port, protocol, timeout=5):
 """
 Test for Query DSL injection vulnerabilities.
 Tests boolean-based injection in search queries.
 """
 vulnerabilities = []
 base_url = f"{protocol}://{host}:{port}"
 
 # Injection payloads for Query DSL
 injection_payloads = [
 # Boolean match_all injection - returns everything
 {
 "query": {
 "bool": {
 "must": [
 {"match_all": {}}
 ]
 }
 }
 },
 # Script query injection (if scripts enabled)
 {
 "query": {
 "script": {
 "script": {
 "source": "true",
 "lang": "painless"
 }
 }
 }
 },
 # Range query injection
 {
 "query": {
 "range": {
 "@timestamp": {
 "gte": "now-1d/d",
 "lt": "now/d"
 }
 }
 }
 }
 ]
 
 # Test against common indices
 test_indices = ['_all', '*', 'logs-*', 'test']
 
 for index in test_indices:
 search_url = f"{base_url}/{index}/_search"
 
 for payload in injection_payloads:
 try:
 json_payload = json.dumps(payload)
 req = Request(search_url, data=json_payload.encode('utf-8'), method='POST')
 req.add_header('Content-Type', 'application/json')
 
 with urlopen(req, timeout=timeout) as response:
 status_code = response.getcode()
 data = response.read().decode('utf-8')
 
 if status_code == 200:
 result = json.loads(data)
 hits = result.get('hits', {}).get('hits', [])
 
 # Data returned = injection successful!
 if hits or result.get('hits', {}).get('total', {}).get('value', 0) > 0:
 vuln = {
 'type': 'Query DSL Injection',
 'index': index,
 'payload': payload,
 'hits_returned': len(hits),
 'total_hits': result.get('hits', {}).get('total', {}).get('value', 0)
 }
 vulnerabilities.append(vuln)
 break # Found vulnerability
 
 except (HTTPError, URLError, socket.timeout):
 continue
 
 return vulnerabilities

def test_script_field_injection(host, port, protocol, timeout=5):
 """
 Test for script field injection - CRITICAL vulnerability.
 If Painless scripts execute, RCE is possible.
 """
 vulnerabilities = []
 base_url = f"{protocol}://{host}:{port}"
 
 # Script field injection payload
 script_payload = {
 "query": {"match_all": {}},
 "script_fields": {
 "test_field": {
 "script": {
 "source": "1+1", # Benign arithmetic test
 "lang": "painless"
 }
 }
 },
 "size": 1
 }
 
 for index in ['_all', '*']:
 try:
 search_url = f"{base_url}/{index}/_search"
 json_payload = json.dumps(script_payload)
 req = Request(search_url, data=json_payload.encode('utf-8'), method='POST')
 req.add_header('Content-Type', 'application/json')
 
 with urlopen(req, timeout=timeout) as response:
 status_code = response.getcode()
 data = response.read().decode('utf-8')
 
 if status_code == 200:
 result = json.loads(data)
 hits = result.get('hits', {}).get('hits', [])
 
 # Check if script executed
 if hits:
 for hit in hits:
 if 'fields' in hit and 'test_field' in hit['fields']:
 # CRITICAL: Script execution confirmed!
 vuln = {
 'type': 'Script Field Injection',
 'index': index,
 'payload': script_payload,
 'script_result': hit['fields']['test_field']
 }
 vulnerabilities.append(vuln)
 return vulnerabilities # Critical - stop testing
 
 except (HTTPError, URLError, socket.timeout):
 continue
 
 return vulnerabilities

# Scan a target for Elasticsearch injection
cxg scan --scope elasticsearch.example.com --template elasticsearch-query-injection.py

# Scan with explicit port
cxg scan --scope elasticsearch.example.com --ports 9200 --template elasticsearch-query-injection.py

# JSON output for automation
cxg scan --scope elasticsearch.example.com --template elasticsearch-query-injection.py --output-format json

# Verbose mode for debugging
cxg scan --scope elasticsearch.example.com --template elasticsearch-query-injection.py -vv

# Extended timeout for slow networks
cxg scan --scope elasticsearch.example.com --template elasticsearch-query-injection.py --timeout 90s

# From file
cxg scan --scope @elasticsearch-targets.txt --template elasticsearch-query-injection.py

# CIDR range
cxg scan --scope 10.0.0.0/24 --ports 9200 --template elasticsearch-query-injection.py

# Run Python template directly
export CERT_X_GEN_TARGET_HOST=elasticsearch.example.com
export CERT_X_GEN_TARGET_PORT=9200
python3 elasticsearch-query-injection.py

{
 "findings": [{
 "template_id": "elasticsearch-query-injection",
 "severity": "critical",
 "name": "Elasticsearch Query DSL Injection Vulnerability",
 "description": "CRITICAL: Script field injection vulnerability detected - Painless script execution is enabled. Found 2 Query DSL injection vulnerabilities...",
 "cvss_score": 9.8,
 "evidence": {
 "elasticsearch_detected": true,
 "version": "8.13.0",
 "script_injections": [{
 "type": "Script Field Injection",
 "script_result": [2]
 }],
 "query_injections": [...]
 }
 }]
}

{
 "findings": [{
 "template_id": "elasticsearch-query-injection",
 "severity": "medium",
 "description": "Elasticsearch service detected on host:9200 (version 8.3.2). Found 4 exposed Elasticsearch API endpoints.",
 "cvss_score": 5.0,
 "evidence": {
 "elasticsearch_detected": true,
 "version": "8.3.2",
 "api_exposures": [...]
 }
 }]
}

Total Targets Scanned: 5
Accessible Targets: 3 (60%)
Vulnerable Targets: 2 (40%)
Script Injection Vulns: 2 (CRITICAL)
Query Injection Vulns: 2 (HIGH)
API Exposure Only: 1 (MEDIUM)
False Positives: 0 (0%)

# SECURE: Disable inline scripting completely
PUT /_cluster/settings
{
 "persistent": {
 "script.inline": false,
 "script.stored": false,
 "script.max_compilations_rate": "75/5m"
 }
}

# Enable X-Pack Security (now free!)
xpack.security.enabled: true
xpack.security.transport.ssl.enabled: true
xpack.security.http.ssl.enabled: true

# VULNERABLE: String concatenation
def search_users(username):
 query = f'{{"query": {{"match": {{"username": "{username}"}}}}}}'
 return es.search(index="users", body=query)

# SECURE: Use Elasticsearch Python client with typed queries
from elasticsearch import Elasticsearch

def search_users_secure(username: str):
 """Secure user search with input validation."""
 # 1. Validate input type
 if not isinstance(username, str):
 raise ValueError("Username must be string")
 
 # 2. Sanitize input (reject Query DSL operators)
 forbidden_chars = ['{', '}', '[', ']', '"', '$']
 if any(char in username for char in forbidden_chars):
 raise ValueError("Invalid characters in username")
 
 # 3. Use client library's query builder
 query = {
 "query": {
 "term": { # Use 'term' not 'match' for exact matching
 "username.keyword": username
 }
 }
 }
 
 # 4. Execute with proper error handling
 try:
 result = es.search(index="users", body=query, request_timeout=5)
 return result['hits']['hits']
 except Exception as e:
 logger.error(f"Search error: {e}")
 return []

# elasticsearch.yml - Secure Configuration

# 1. Enable Security Features
xpack.security.enabled: true
xpack.security.enrollment.enabled: true

# 2. Disable Dynamic Scripting
script.allowed_types: none
script.allowed_contexts: []

# 3. Network Security
network.host: 127.0.0.1 # Bind to localhost only
http.port: 9200
http.cors.enabled: false

# 4. Authentication
xpack.security.authc:
 realms:
 native:
 native1:
 order: 0

# 5. Audit Logging
xpack.security.audit.enabled: true
xpack.security.audit.logfile.events.include:
 - authentication_failed
 - access_denied
 - tampered_request

# 6. Index-Level Security
xpack.security.dls_fls.enabled: true

# 7. Field-Level Security
xpack.security.fls.enabled: true

# 8. Circuit Breakers (DoS Protection)
indices.breaker.total.limit: 70%
indices.breaker.request.limit: 40%
indices.breaker.fielddata.limit: 40%

# Input Validation Layer
class QueryValidator:
 """Validate and sanitize Elasticsearch queries."""
 
 FORBIDDEN_OPERATORS = [
 'script', '$where', 'exec', 'eval', 
 'function', 'return', 'import', 'require'
 ]
 
 MAX_QUERY_DEPTH = 5
 MAX_QUERY_SIZE = 1024
 
 @staticmethod
 def validate_query(query: dict) -> bool:
 """
 Validate query structure and content.
 Returns True if safe, raises ValueError if malicious.
 """
 # Check query size
 query_json = json.dumps(query)
 if len(query_json) > QueryValidator.MAX_QUERY_SIZE:
 raise ValueError("Query exceeds size limit")
 
 # Check for forbidden operators
 if any(op in query_json.lower() for op in QueryValidator.FORBIDDEN_OPERATORS):
 raise ValueError("Query contains forbidden operators")
 
 # Validate query depth (prevent nested injection)
 if QueryValidator._get_depth(query) > QueryValidator.MAX_QUERY_DEPTH:
 raise ValueError("Query nesting too deep")
 
 return True
 
 @staticmethod
 def _get_depth(obj, current_depth=0):
 """Calculate nesting depth of dict/list."""
 if isinstance(obj, dict):
 if not obj:
 return current_depth
 return max(QueryValidator._get_depth(v, current_depth + 1) for v in obj.values())
 elif isinstance(obj, list):
 if not obj:
 return current_depth
 return max(QueryValidator._get_depth(item, current_depth + 1) for item in obj)
 return current_depth

# Usage
def safe_search(user_query: dict):
 try:
 QueryValidator.validate_query(user_query)
 return es.search(body=user_query)
 except ValueError as e:
 logger.warning(f"Malicious query blocked: {e}")
 return {"error": "Invalid query"}

# Add domain-specific query patterns
custom_payloads = [
 # GeoPoint injection
 {
 "query": {
 "geo_distance": {
 "distance": "12km",
 "location": {"lat": 40, "lon": -70}
 }
 }
 },
 # Nested query injection
 {
 "query": {
 "nested": {
 "path": "user",
 "query": {"match_all": {}}
 }
 }
 }
]

# Version-specific payloads
VERSION_PAYLOADS = {
 "8.x": [
 # Modern security features
 {"query": {"match_all": {}}},
 ],
 "7.x": [
 # Legacy Groovy scripts
 {"query": {"script": {"script": "1+1", "lang": "groovy"}}},
 ],
 "6.x": [
 # Older DSL syntax
 {"query": {"filtered": {"query": {"match_all": {}}}}},
 ]
}

# GitHub Actions - Elasticsearch Security Scan
name: Elasticsearch Security Scan

on:
 schedule:
 - cron: '0 2 * * *' # Daily at 2 AM
 workflow_dispatch:

jobs:
 scan:
 runs-on: ubuntu-latest
 steps:
 - name: Checkout
 uses: actions/checkout@v3
 
 - name: Run CERT-X-GEN Scan
 run: |
 cxg scan \
 --scope ${{ secrets.ELASTICSEARCH_HOST }} \
 --ports 9200 \
 --template elasticsearch-query-injection.py \
 --output-format json \
 --output results.json
 
 - name: Check for Vulnerabilities
 run: |
 CRITICAL=$(jq '[.findings[] | select(.severity=="critical")] | length' results.json)
 if [ "$CRITICAL" -gt 0 ]; then
 echo " CRITICAL vulnerabilities found!"
 exit 1
 fi
 
 - name: Upload Results
 uses: actions/upload-artifact@v3
 with:
 name: scan-results
 path: results.json