"""
HTTP protocol analyzer implementation.
"""
from typing import Dict, List, Any, Optional
from collections import defaultdict
import re
from urllib.parse import urlparse
from .base import BaseProtocolAnalyzer
class HTTPProtocolAnalyzer(BaseProtocolAnalyzer):
"""
HTTP protocol analyzer for extracting and analyzing HTTP traffic.
"""
protocol_name = "HTTP"
# Common HTTP status code categories
STATUS_CATEGORIES = {
"1xx": "Informational",
"2xx": "Success",
"3xx": "Redirection",
"4xx": "Client Error",
"5xx": "Server Error"
}
# Security-sensitive HTTP headers
SECURITY_HEADERS = [
"content-security-policy",
"x-xss-protection",
"x-content-type-options",
"strict-transport-security",
"x-frame-options",
"referrer-policy"
]
def extract_features(self,
packets: List[Dict[str, Any]],
include_headers: bool = True,
include_body: bool = False,
**kwargs) -> Dict[str, Any]:
"""
Extract HTTP-specific features from packet data.
Args:
packets: List of packet dictionaries
include_headers: Whether to include HTTP headers
include_body: Whether to include HTTP bodies
**kwargs: Additional extraction parameters
Returns:
Dictionary of extracted HTTP features
"""
# Filter to only HTTP packets
http_packets = self._filter_packets(packets)
# Group packets into HTTP conversations
conversations = self._extract_conversations(http_packets)
# Extract HTTP requests and responses
http_data = {
"conversations": {},
"statistics": {
"total_requests": 0,
"total_responses": 0,
"status_codes": defaultdict(int),
"methods": defaultdict(int),
"hosts": defaultdict(int),
"content_types": defaultdict(int),
"total_bytes": 0
},
"security_findings": []
}
# Process each conversation
for conv_id, conv_packets in conversations.items():
requests = []
responses = []
for packet in conv_packets:
http = packet.get('http', {})
# Determine if this is a request or response
if 'request' in http:
requests.append(self._extract_http_request(packet, include_headers, include_body))
http_data["statistics"]["total_requests"] += 1
# Update method statistics
method = http.get('request_method', '')
if method:
http_data["statistics"]["methods"][method] += 1
# Update host statistics
host = None
for header in http.get('request_header', []):
if header.get('name', '').lower() == 'host':
host = header.get('value', '')
break
if host:
http_data["statistics"]["hosts"][host] += 1
elif 'response' in http:
response = self._extract_http_response(packet, include_headers, include_body)
responses.append(response)
http_data["statistics"]["total_responses"] += 1
# Update status code statistics
status_code = http.get('response_code', '')
if status_code:
http_data["statistics"]["status_codes"][status_code] += 1
# Update content type statistics
content_type = None
for header in http.get('response_header', []):
if header.get('name', '').lower() == 'content-type':
content_type = header.get('value', '').split(';')[0].strip()
break
if content_type:
http_data["statistics"]["content_types"][content_type] += 1
# Check for missing security headers
self._check_security_headers(response, http_data["security_findings"])
# Update byte count
http_data["statistics"]["total_bytes"] += int(packet.get('length', 0))
# Match requests with responses where possible
matched_exchanges = self._match_requests_responses(requests, responses)
http_data["conversations"][conv_id] = {
"exchanges": matched_exchanges,
"unmatched_requests": [req for req in requests if not req.get('_matched')],
"unmatched_responses": [resp for resp in responses if not resp.get('_matched')]
}
# Remove temporary matching flags
for req in requests:
req.pop('_matched', None)
for resp in responses:
resp.pop('_matched', None)
return http_data
def generate_context(self,
features: Dict[str, Any],
detail_level: int = 2,
max_conversations: int = 10,
**kwargs) -> Dict[str, Any]:
"""
Generate AI-friendly context from HTTP features.
Args:
features: Dictionary of extracted HTTP features
detail_level: Level of detail (1-3, where 3 is most detailed)
max_conversations: Maximum number of conversations to include
**kwargs: Additional context parameters
Returns:
Dictionary with formatted HTTP context
"""
conversations = features.get("conversations", {})
statistics = features.get("statistics", {})
security_findings = features.get("security_findings", [])
# Prepare context
context = {
"protocol": "HTTP",
"summary": {
"total_requests": statistics.get("total_requests", 0),
"total_responses": statistics.get("total_responses", 0),
"total_conversations": len(conversations),
"total_bytes_transferred": statistics.get("total_bytes", 0)
},
"top_statistics": {
"status_codes": self._get_top_items(statistics.get("status_codes", {}), 10),
"methods": self._get_top_items(statistics.get("methods", {}), 5),
"hosts": self._get_top_items(statistics.get("hosts", {}), 5),
"content_types": self._get_top_items(statistics.get("content_types", {}), 5)
},
"security": {
"findings": security_findings[:10] # Limit to top 10 findings
},
"conversations": {}
}
# Add HTTP status code descriptions
status_descriptions = {}
for status, count in statistics.get("status_codes", {}).items():
category = self._get_status_category(status)
status_descriptions[status] = f"{status} - {category}"
context["status_code_descriptions"] = status_descriptions
# Add conversation details based on detail level
sorted_conversations = sorted(
conversations.items(),
key=lambda x: len(x[1]["exchanges"]),
reverse=True
)[:max_conversations]
for conv_id, conv_data in sorted_conversations:
exchanges = conv_data["exchanges"]
context["conversations"][conv_id] = {
"exchanges_count": len(exchanges),
"exchanges": []
}
# Add exchange details based on detail level
for exchange in exchanges:
exchange_summary = {
"request": {
"method": exchange["request"].get("method", ""),
"uri": exchange["request"].get("uri", ""),
"version": exchange["request"].get("version", "")
},
"response": {
"status_code": exchange["response"].get("status_code", ""),
"status_phrase": exchange["response"].get("status_phrase", ""),
"content_length": exchange["response"].get("content_length", 0),
"content_type": exchange["response"].get("content_type", "")
}
}
# Add headers for higher detail levels
if detail_level >= 2:
exchange_summary["request"]["headers"] = exchange["request"].get("headers", {})
exchange_summary["response"]["headers"] = exchange["response"].get("headers", {})
# Add body preview for highest detail level
if detail_level >= 3:
# Include a preview of the body (truncated)
req_body = exchange["request"].get("body", "")
resp_body = exchange["response"].get("body", "")
exchange_summary["request"]["body_preview"] = req_body[:200] + "..." if len(req_body) > 200 else req_body
exchange_summary["response"]["body_preview"] = resp_body[:200] + "..." if len(resp_body) > 200 else resp_body
context["conversations"][conv_id]["exchanges"].append(exchange_summary)
return context
def extract_insights(self,
packets: List[Dict[str, Any]],
extract_queries: bool = True,
analyze_response_codes: bool = True,
detect_tunneling: bool = False,
**kwargs) -> Dict[str, Any]:
"""
Extract deeper HTTP-specific insights.
Args:
packets: List of packet dictionaries
extract_queries: Whether to extract URL query patterns
analyze_response_codes: Whether to analyze response code patterns
detect_tunneling: Whether to look for HTTP tunneling
**kwargs: Additional parameters
Returns:
Dictionary of HTTP insights
"""
features = self.extract_features(packets)
insights = {
"protocol": "HTTP",
"findings": [],
"patterns": {}
}
# Extract URL and query patterns
if extract_queries:
query_patterns = self._analyze_query_patterns(features)
insights["patterns"]["queries"] = query_patterns
# Analyze response code patterns
if analyze_response_codes:
code_patterns = self._analyze_response_codes(features)
insights["patterns"]["response_codes"] = code_patterns
# Detect possible HTTP tunneling
if detect_tunneling:
tunneling = self._detect_tunneling(features)
if tunneling:
insights["tunneling"] = tunneling
return insights
def _extract_http_request(self,
packet: Dict[str, Any],
include_headers: bool = True,
include_body: bool = False) -> Dict[str, Any]:
"""Extract HTTP request details from a packet."""
http = packet.get('http', {})
request = {
"method": http.get('request_method', ''),
"uri": http.get('request_uri', ''),
"version": http.get('request_version', ''),
"timestamp": packet.get('timestamp', 0),
"frame_number": packet.get('frame_number', '')
}
# Parse URI components if available
uri = request["uri"]
if uri:
parsed_uri = urlparse(uri)
request["uri_path"] = parsed_uri.path
request["uri_query"] = parsed_uri.query
# Extract headers if requested
if include_headers:
headers = {}
for header in http.get('request_header', []):
name = header.get('name', '').lower()
value = header.get('value', '')
headers[name] = value
request["headers"] = headers
# Extract body if requested
if include_body and 'request_body' in http:
request["body"] = http.get('request_body', '')
request["body_length"] = len(request["body"])
return request
def _extract_http_response(self,
packet: Dict[str, Any],
include_headers: bool = True,
include_body: bool = False) -> Dict[str, Any]:
"""Extract HTTP response details from a packet."""
http = packet.get('http', {})
response = {
"status_code": http.get('response_code', ''),
"status_phrase": http.get('response_phrase', ''),
"version": http.get('response_version', ''),
"timestamp": packet.get('timestamp', 0),
"frame_number": packet.get('frame_number', '')
}
# Extract headers if requested
if include_headers:
headers = {}
for header in http.get('response_header', []):
name = header.get('name', '').lower()
value = header.get('value', '')
headers[name] = value
response["headers"] = headers
# Extract common useful headers
response["content_type"] = headers.get('content-type', '').split(';')[0].strip()
try:
response["content_length"] = int(headers.get('content-length', 0))
except (ValueError, TypeError):
response["content_length"] = 0
# Extract body if requested
if include_body and 'response_body' in http:
response["body"] = http.get('response_body', '')
response["body_length"] = len(response["body"])
return response
def _match_requests_responses(self,
requests: List[Dict[str, Any]],
responses: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Match HTTP requests with their corresponding responses."""
exchanges = []
# Sort by timestamp
sorted_requests = sorted(requests, key=lambda x: float(x.get('timestamp', 0)))
sorted_responses = sorted(responses, key=lambda x: float(x.get('timestamp', 0)))
# Match requests with responses
for req in sorted_requests:
req_time = float(req.get('timestamp', 0))
# Find the first response that comes after this request
for resp in sorted_responses:
if resp.get('_matched'):
continue
resp_time = float(resp.get('timestamp', 0))
if resp_time > req_time:
# Match found
exchanges.append({
"request": req,
"response": resp,
"time_delta": resp_time - req_time
})
# Mark as matched
req['_matched'] = True
resp['_matched'] = True
break
return exchanges
def _check_security_headers(self,
response: Dict[str, Any],
findings: List[str]) -> None:
"""Check HTTP response for missing security headers."""
if 'headers' not in response:
return
headers = response.get('headers', {})
for header in self.SECURITY_HEADERS:
if header not in headers:
findings.append(f"Missing security header: {header}")
def _get_status_category(self, status_code: str) -> str:
"""Get the category for an HTTP status code."""
try:
category_key = status_code[0] + "xx"
return self.STATUS_CATEGORIES.get(category_key, "Unknown")
except (IndexError, TypeError):
return "Unknown"
def _get_top_items(self,
items: Dict[str, int],
limit: int = 10) -> Dict[str, int]:
"""Get the top N items from a dictionary by count."""
return dict(sorted(items.items(), key=lambda x: x[1], reverse=True)[:limit])
def _analyze_query_patterns(self, features: Dict[str, Any]) -> Dict[str, Any]:
"""Analyze HTTP query patterns."""
query_patterns = {
"common_parameters": defaultdict(int),
"parameter_types": defaultdict(set),
"suspicious_patterns": []
}
# Extract all request URIs with queries
for conv_id, conv_data in features.get("conversations", {}).items():
for exchange in conv_data.get("exchanges", []):
request = exchange.get("request", {})
uri = request.get("uri", "")
if "?" in uri:
parsed = urlparse(uri)
query = parsed.query
if query:
# Extract parameters
params = query.split("&")
for param in params:
if "=" in param:
name, value = param.split("=", 1)
query_patterns["common_parameters"][name] += 1
# Detect parameter types
param_type = self._detect_parameter_type(value)
query_patterns["parameter_types"][name].add(param_type)
# Check for suspicious patterns
if self._is_suspicious_parameter(name, value):
query_patterns["suspicious_patterns"].append({
"parameter": name,
"value_sample": value,
"uri": uri
})
# Convert parameter types from sets to lists
query_patterns["parameter_types"] = {
k: list(v) for k, v in query_patterns["parameter_types"].items()
}
# Get top parameters
query_patterns["top_parameters"] = dict(
sorted(query_patterns["common_parameters"].items(),
key=lambda x: x[1],
reverse=True)[:10]
)
return query_patterns
def _detect_parameter_type(self, value: str) -> str:
"""Detect the data type of a parameter value."""
if not value:
return "empty"
# Check for numeric types
if value.isdigit():
return "integer"
if re.match(r'^-?\d+(\.\d+)?$', value):
return "number"
# Check for dates
if re.match(r'^\d{4}-\d{2}-\d{2}', value):
return "date"
# Check for UUIDs
if re.match(r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$', value, re.I):
return "uuid"
# Check for email addresses
if re.match(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', value):
return "email"
# Check for URLs
if re.match(r'^https?://', value):
return "url"
# Otherwise, assume it's a string
return "string"
def _is_suspicious_parameter(self, name: str, value: str) -> bool:
"""Check if a parameter name or value looks suspicious."""
# Check for common injection patterns
suspicious_patterns = [
r"['\"].*--", # SQL injection
r"<script.*>", # XSS
r"/etc/passwd", # Path traversal
r"\.\.(/|\\)", # Directory traversal
r";.*\s*\w+\s*=", # Command injection
r"(exec|eval|system)\(", # Code injection
]
for pattern in suspicious_patterns:
if re.search(pattern, value, re.I):
return True
# Check for suspicious parameter names
suspicious_names = [
"passwd", "password", "pwd",
"token", "key", "secret",
"command", "cmd", "exec",
"query", "sql", "debug"
]
name_lower = name.lower()
for sus_name in suspicious_names:
if sus_name in name_lower:
return True
return False
def _analyze_response_codes(self, features: Dict[str, Any]) -> Dict[str, Any]:
"""Analyze HTTP response code patterns."""
status_codes = features.get("statistics", {}).get("status_codes", {})
code_patterns = {
"success_rate": 0,
"error_rate": 0,
"redirection_rate": 0,
"top_codes": {},
"unusual_codes": []
}
total_responses = features.get("statistics", {}).get("total_responses", 0)
if total_responses > 0:
success_count = sum(count for code, count in status_codes.items()
if code.startswith('2'))
error_count = sum(count for code, count in status_codes.items()
if code.startswith('4') or code.startswith('5'))
redirect_count = sum(count for code, count in status_codes.items()
if code.startswith('3'))
code_patterns["success_rate"] = success_count / total_responses
code_patterns["error_rate"] = error_count / total_responses
code_patterns["redirection_rate"] = redirect_count / total_responses
# Get top status codes
code_patterns["top_codes"] = dict(
sorted(status_codes.items(), key=lambda x: x[1], reverse=True)[:5]
)
# Identify unusual status codes
common_codes = {'200', '301', '302', '304', '400', '401', '403', '404', '500'}
for code in status_codes:
if code not in common_codes:
code_patterns["unusual_codes"].append(code)
return code_patterns
def _detect_tunneling(self, features: Dict[str, Any]) -> Dict[str, Any]:
"""Detect potential HTTP tunneling."""
tunneling_indicators = []
# Look for indicators of HTTP tunneling
for conv_id, conv_data in features.get("conversations", {}).items():
for exchange in conv_data.get("exchanges", []):
request = exchange.get("request", {})
response = exchange.get("response", {})
# Check for CONNECT method
if request.get("method") == "CONNECT":
tunneling_indicators.append({
"type": "CONNECT method",
"uri": request.get("uri", ""),
"frame": request.get("frame_number")
})
# Check for unusually large request or response bodies
if request.get("body_length", 0) > 10000:
tunneling_indicators.append({
"type": "Large request body",
"size": request.get("body_length", 0),
"uri": request.get("uri", ""),
"frame": request.get("frame_number")
})
# Check for unusual content types in large responses
if response.get("body_length", 0) > 10000:
content_type = response.get("content_type", "")
if content_type and content_type not in ["text/html", "application/json", "text/javascript"]:
tunneling_indicators.append({
"type": "Unusual content type for large response",
"content_type": content_type,
"size": response.get("body_length", 0),
"frame": response.get("frame_number")
})
if tunneling_indicators:
return {
"detected": True,
"indicators": tunneling_indicators
}
return {"detected": False}
