"""
HTTP Incoming Request Handler Tester
"""
import socket
import ssl
import time
from urllib.parse import urlparse
class HTTPRequestTester:
def __init__(self, url):
self.url = url
self.parsed = urlparse(url)
self.host = self.parsed.netloc.split(':')[0]
self.port = 443 if self.parsed.scheme == 'https' else 80
self.findings = []
def send_raw(self, data, timeout=10):
"""Send raw data to server"""
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(timeout)
if self.parsed.scheme == 'https':
context = ssl.create_default_context()
sock = context.wrap_socket(sock, server_hostname=self.host)
sock.connect((self.host, self.port))
sock.send(data.encode() if isinstance(data, str) else data)
response = b""
try:
while True:
chunk = sock.recv(4096)
if not chunk:
break
response += chunk
except socket.timeout:
pass
sock.close()
return response.decode('utf-8', errors='ignore')
except Exception as e:
return f"Error: {str(e)}"
def test_oversized_headers(self):
"""Test handling of oversized headers"""
print("\n[*] Testing oversized header handling...")
sizes = [1000, 4000, 8000, 16000, 32000]
for size in sizes:
header_value = "A" * size
request = f"""GET / HTTP/1.1\r
Host: {self.host}\r
X-Oversized: {header_value}\r
\r
"""
response = self.send_raw(request)
if 'HTTP/1.1 413' in response:
print(f" [OK] {size} bytes: 413 Request Entity Too Large")
elif 'HTTP/1.1 431' in response:
print(f" [OK] {size} bytes: 431 Request Header Fields Too Large")
elif 'HTTP/1.1 400' in response:
print(f" [OK] {size} bytes: 400 Bad Request")
elif 'HTTP/1.' in response:
print(f" [INFO] {size} bytes: Accepted")
else:
print(f" [WARN] {size} bytes: Unexpected response")
def test_long_uri(self):
"""Test handling of long URIs"""
print("\n[*] Testing long URI handling...")
sizes = [1000, 2000, 4000, 8000, 16000]
for size in sizes:
long_path = "/" + "A" * size
request = f"""GET {long_path} HTTP/1.1\r
Host: {self.host}\r
\r
"""
response = self.send_raw(request)
if 'HTTP/1.1 414' in response:
print(f" [OK] {size} char URI: 414 URI Too Long")
elif 'HTTP/1.1 400' in response:
print(f" [OK] {size} char URI: 400 Bad Request")
elif 'HTTP/1.' in response:
print(f" [INFO] {size} char URI: Accepted")
def test_malformed_requests(self):
"""Test handling of malformed requests"""
print("\n[*] Testing malformed request handling...")
malformed_requests = [
("GET / HTTP/9.9\r\nHost: {host}\r\n\r\n", "Invalid HTTP version"),
("INVALID / HTTP/1.1\r\nHost: {host}\r\n\r\n", "Invalid method"),
("GET /\r\nHost: {host}\r\n\r\n", "Missing HTTP version"),
("GET / HTTP/1.1\r\n\r\n", "Missing Host header"),
("GET / HTTP/1.1\r\nHost:\r\n\r\n", "Empty Host header"),
("GET HTTP/1.1\r\nHost: {host}\r\n\r\n", "Missing path"),
("\x00GET / HTTP/1.1\r\nHost: {host}\r\n\r\n", "Null byte prefix"),
]
for request_template, description in malformed_requests:
request = request_template.format(host=self.host)
response = self.send_raw(request)
if 'HTTP/1.1 400' in response or 'HTTP/1.1 505' in response:
print(f" [OK] {description}: Properly rejected")
elif 'HTTP/1.1 200' in response:
print(f" [WARN] {description}: Accepted (possible issue)")
self.findings.append({
'type': 'Malformed Request Accepted',
'description': description,
'severity': 'Low'
})
else:
print(f" [INFO] {description}: {response[:50]}")
def test_duplicate_headers(self):
"""Test handling of duplicate headers"""
print("\n[*] Testing duplicate header handling...")
request = f"""GET / HTTP/1.1\r
Host: {self.host}\r
Host: evil.com\r
\r
"""
response = self.send_raw(request)
print(f" Multiple Host headers: {'Rejected' if '400' in response else 'Accepted'}")
request = f"""POST / HTTP/1.1\r
Host: {self.host}\r
Content-Length: 10\r
Content-Length: 20\r
\r
1234567890"""
response = self.send_raw(request)
if '400' in response:
print(f" [OK] Multiple Content-Length: Rejected")
else:
print(f" [WARN] Multiple Content-Length: Accepted")
self.findings.append({
'type': 'Duplicate Headers Accepted',
'header': 'Content-Length',
'severity': 'Medium'
})
def test_slow_request(self):
"""Test slow HTTP request handling (Slowloris-style)"""
print("\n[*] Testing slow request handling...")
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(60)
if self.parsed.scheme == 'https':
context = ssl.create_default_context()
sock = context.wrap_socket(sock, server_hostname=self.host)
sock.connect((self.host, self.port))
sock.send(f"GET / HTTP/1.1\r\nHost: {self.host}\r\n".encode())
start_time = time.time()
for i in range(10):
time.sleep(2)
sock.send(b"X-Header: value\r\n")
elapsed = time.time() - start_time
if elapsed > 15:
print(f" [INFO] Connection maintained for {elapsed:.0f}s")
break
sock.close()
if elapsed > 10:
print(f" [WARN] Server allows slow requests (potential DoS)")
self.findings.append({
'type': 'Slow Request Allowed',
'duration': elapsed,
'severity': 'Medium'
})
except socket.timeout:
print(f" [OK] Connection timed out properly")
except Exception as e:
print(f" [INFO] {str(e)}")
def test_large_body(self):
"""Test large request body handling"""
print("\n[*] Testing large body handling...")
sizes = [(1024, "1KB"), (1024*1024, "1MB"), (10*1024*1024, "10MB")]
for size, label in sizes:
body = "A" * size
request = f"""POST / HTTP/1.1\r
Host: {self.host}\r
Content-Type: application/x-www-form-urlencoded\r
Content-Length: {len(body)}\r
\r
{body}"""
response = self.send_raw(request, timeout=30)
if '413' in response:
print(f" [OK] {label}: 413 Payload Too Large")
break
elif '400' in response or '500' in response:
print(f" [INFO] {label}: Rejected with error")
else:
print(f" [INFO] {label}: Accepted")
def generate_report(self):
"""Generate findings report"""
print("\n" + "="*60)
print("HTTP REQUEST HANDLING REPORT")
print("="*60)
if not self.findings:
print("\nNo significant issues found.")
else:
for f in self.findings:
print(f"\n[{f['severity']}] {f['type']}")
if 'description' in f:
print(f" Description: {f['description']}")
def run_tests(self):
"""Run all tests"""
self.test_oversized_headers()
self.test_long_uri()
self.test_malformed_requests()
self.test_duplicate_headers()
self.test_slow_request()
self.test_large_body()
self.generate_report()
tester = HTTPRequestTester("https://target.com")
tester.run_tests()