1#!/usr/bin/env python3
2# @id: redis-unauthenticated-python
3# @name: Redis Unauthenticated Access Detection (Python)
4# @author: CERT-X-GEN Security Team
5# @severity: critical
6# @description: Detects Redis instances exposed without authentication using Python
7# @tags: redis, unauthenticated, database, nosql, cwe-306
8# @cwe: CWE-306
9# @cvss: 9.8
10# @references: https://redis.io/docs/management/security/, https://cwe.mitre.org/data/definitions/306.html
11# @confidence: 95
12# @version: 1.0.0
13"""
14Tests for Redis instances exposed without authentication.
15"""
16
17import socket
18import sys
19import json
20from datetime import datetime
21
22# Template metadata
23METADATA = {
40}
41
42def test_redis(host, port=6379, timeout=10):
43 """Test Redis for unauthenticated access"""
44 findings = []
45
46 try:
47 # Create socket connection
48 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
49 sock.settimeout(timeout)
50 sock.connect((host, port))
51
52 # Test commands
53 commands = [
54 b"INFO\r\n",
55 b"PING\r\n",
56 b"*1\r\n$4\r\nINFO\r\n",
57 b"*1\r\n$4\r\nPING\r\n"
58 ]
59
60 responses = []
61 for cmd in commands:
62 try:
63 sock.sendall(cmd)
64 response = sock.recv(8192)
65 if response:
66 responses.append(response.decode('utf-8', errors='ignore'))
67 except Exception as e:
68 continue
69
70 sock.close()
71
72 # Check for Redis indicators
73 full_response = ''.join(responses)
74
75 indicators = [
76 'redis_version',
77 'redis_mode',
78 'used_memory',
79 'connected_clients',
80 'role:master',
81 'role:slave',
82 '+PONG'
83 ]
84
85 matched_patterns = [ind for ind in indicators if ind in full_response]
86
87 if matched_patterns:
88 finding = {
89 "target": f"{host}:{port}",
90 "template_id": METADATA["id"],
91 "severity": METADATA["severity"],
92 "confidence": METADATA["confidence"],
93 "title": METADATA["name"],
94 "description": METADATA["description"],
95 "evidence": {
96 "request": "\\r\\n".join([cmd.decode('utf-8', errors='ignore').strip() for cmd in commands]),
97 "response": full_response[:1000], # First 1000 chars
98 "matched_patterns": matched_patterns,
99 "data": {
100 "protocol": "tcp",
101 "port": port,
102 "response_length": len(full_response)
103 }
104 },
105 "cwe_ids": METADATA["cwe"],
106 "tags": METADATA["tags"],
107 "timestamp": datetime.utcnow().isoformat() + "Z"
108 }
109 findings.append(finding)
110
111 except socket.timeout:
112 pass
113 except ConnectionRefusedError:
114 pass
115 except Exception as e:
116 pass
117
118 return findings
119
120def main():
148
149if __name__ == "__main__":
150 main()
#!/usr/bin/env python3
# @id: redis-unauthenticated-python
# @name: Redis Unauthenticated Access Detection (Python)
# @author: CERT-X-GEN Security Team
# @severity: critical
# @description: Detects Redis instances exposed without authentication using Python
# @tags: redis, unauthenticated, database, nosql, cwe-306
# @cwe: CWE-306
# @cvss: 9.8
# @references: https://redis.io/docs/management/security/, https://cwe.mitre.org/data/definitions/306.html
# @confidence: 95
# @version: 1.0.0
"""
Tests for Redis instances exposed without authentication.
"""
import socket
import sys
import json
from datetime import datetime
# Template metadata
METADATA = {
}
def test_redis(host, port=6379, timeout=10):
"""Test Redis for unauthenticated access"""
findings = []
try:
# Create socket connection
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(timeout)
sock.connect((host, port))
# Test commands
commands = [
b"INFO\r\n",
b"PING\r\n",
b"*1\r\n$4\r\nINFO\r\n",
b"*1\r\n$4\r\nPING\r\n"
]
responses = []
for cmd in commands:
try:
sock.sendall(cmd)
response = sock.recv(8192)
if response:
responses.append(response.decode('utf-8', errors='ignore'))
except Exception as e:
continue
sock.close()
# Check for Redis indicators
full_response = ''.join(responses)
indicators = [
'redis_version',
'redis_mode',
'used_memory',
'connected_clients',
'role:master',
'role:slave',
'+PONG'
]
matched_patterns = [ind for ind in indicators if ind in full_response]
if matched_patterns:
finding = {
"target": f"{host}:{port}",
"template_id": METADATA["id"],
"severity": METADATA["severity"],
"confidence": METADATA["confidence"],
"title": METADATA["name"],
"description": METADATA["description"],
"evidence": {
"request": "\\r\\n".join([cmd.decode('utf-8', errors='ignore').strip() for cmd in commands]),
"response": full_response[:1000], # First 1000 chars
"matched_patterns": matched_patterns,
"data": {
"protocol": "tcp",
"port": port,
"response_length": len(full_response)
}
},
"cwe_ids": METADATA["cwe"],
"tags": METADATA["tags"],
"timestamp": datetime.utcnow().isoformat() + "Z"
}
findings.append(finding)
except socket.timeout:
pass
except ConnectionRefusedError:
pass
except Exception as e:
pass
return findings
def main():
if __name__ == "__main__":
main()