class SSRFProtectionArchitectOptimizer:
def __init__(self):
self.context7_client = Context7Client()
self.threat_analyzer = ThreatAnalyzer()
self.url_validator = URLValidator()
async def design_optimal_ssrf_protection(self,
requirements: SecurityRequirements) -> SSRFProtectionArchitecture:
"""Design optimal SSRF protection architecture using AI analysis."""
ssrf_docs = await self.context7_client.get_library_docs(
context7_library_id='/ssrf-protection/docs',
topic="ssrf security url validation web security 2025",
tokens=3000
)
security_docs = await self.context7_client.get_library_docs(
context7_library_id='/web-security/docs',
topic="request filtering threat prevention best practices 2025",
tokens=2000
)
url_validation = self.url_validator.optimize_validation(
requirements.url_requirements,
requirements.security_level,
ssrf_docs
)
threat_analysis = self.threat_analyzer.analyze_ssrf_threats(
requirements.application_context,
requirements.attack_surface,
security_docs
)
return SSRFProtectionArchitecture(
url_validation_system=url_validation,
request_filtering=self._design_request_filtering(requirements),
network_segmentation=self._design_network_segmentation(requirements),
threat_detection=threat_analysis,
monitoring_system=self._configure_monitoring(),
compliance_framework=self._ensure_compliance(requirements)
)
import { URL } from 'url';
import { createHash } from 'crypto';
import { SecurityLogger } from './security-logger';
interface SSRFProtectionConfig {
allowedDomains: string[];
allowedIPRanges: string[];
blockedDomains: string[];
blockedIPRanges: string[];
allowedProtocols: string[];
maxRedirects: number;
maxResponseSize: number;
timeoutMs: number;
}
export class SSRFProtection {
private config: SSRFProtectionConfig;
private logger: SecurityLogger;
private requestCache: Map<string, RequestResult> = new Map();
constructor(config: SSRFProtectionConfig) {
this.config = config;
this.logger = new SecurityLogger();
}
async validateURL(url: string): Promise<URLValidationResult> {
try {
const parsedURL = new URL(url);
if (!this.config.allowedProtocols.includes(parsedURL.protocol)) {
return this.createInvalidResult(
'PROTOCOL_NOT_ALLOWED',
`Protocol ${parsedURL.protocol} is not allowed`
);
}
const domainValidation = await this.validateDomain(parsedURL.hostname);
if (!domainValidation.isValid) {
return domainValidation;
}
const ipValidation = await this.validateIPAddress(parsedURL.hostname);
if (!ipValidation.isValid) {
return ipValidation;
}
const securityValidation = await this.performSecurityChecks(parsedURL);
if (!securityValidation.isValid) {
return securityValidation;
}
return {
isValid: true,
normalizedURL: this.normalizeURL(parsedURL),
riskScore: this.calculateRiskScore(parsedURL),
};
} catch (error) {
return this.createInvalidResult('INVALID_URL', `Invalid URL: ${error.message}`);
}
}
private async validateDomain(hostname: string): Promise<URLValidationResult> {
for (const blockedDomain of this.config.blockedDomains) {
if (hostname.includes(blockedDomain)) {
return this.createInvalidResult(
'DOMAIN_BLOCKED',
`Domain ${hostname} is blocked`
);
}
}
if (this.config.allowedDomains.length > 0) {
const isAllowed = this.config.allowedDomains.some(allowedDomain =>
hostname === allowedDomain || hostname.endsWith(`.${allowedDomain}`)
);
if (!isAllowed) {
return this.createInvalidResult(
'DOMAIN_NOT_ALLOWED',
`Domain ${hostname} is not in allowlist`
);
}
}
const suspiciousPatterns = [
/localhost/i,
/^127\./,
/^0x[0-9a-f]+/i,
/^0[0-7]{3,}/,
/internal/i,
/private/i,
/metadata/i,
];
for (const pattern of suspiciousPatterns) {
if (pattern.test(hostname)) {
return this.createInvalidResult(
'SUSPICIOUS_DOMAIN',
`Domain ${hostname} matches suspicious pattern`
);
}
}
return { isValid: true };
}
private async validateIPAddress(hostname: string): Promise<URLValidationResult> {
const IP = require('ip');
try {
const addresses = await this.resolveHostname(hostname);
for (const address of addresses) {
if (IP.isV4Format(address) || IP.isV6Format(address)) {
for (const blockedRange of this.config.blockedIPRanges) {
if (IP.cidrSubnet(address, blockedRange)) {
return this.createInvalidResult(
'IP_BLOCKED',
`IP address ${address} is in blocked range ${blockedRange}`
);
}
}
if (this.config.allowedIPRanges.length > 0) {
const isAllowed = this.config.allowedIPRanges.some(allowedRange =>
IP.cidrSubnet(address, allowedRange)
);
if (!isAllowed) {
return this.createInvalidResult(
'IP_NOT_ALLOWED',
`IP address ${address} is not in allowed ranges`
);
}
}
if (this.isPrivateIP(address)) {
return this.createInvalidResult(
'PRIVATE_IP_BLOCKED',
`Private IP address ${address} is not allowed`
);
}
}
}
return { isValid: true };
} catch (error) {
return this.createInvalidResult(
'IP_VALIDATION_ERROR',
`IP validation failed: ${error.message}`
);
}
}
private async resolveHostname(hostname: string): Promise<string[]> {
const dns = require('dns').promises;
try {
const { addresses } = await dns.resolve4(hostname);
return addresses;
} catch (error) {
try {
const { addresses } = await dns.resolve6(hostname);
return addresses;
} catch (ipv6Error) {
throw new Error(`Unable to resolve hostname: ${hostname}`);
}
}
}
private isPrivateIP(ip: string): boolean {
const IP = require('ip');
const privateRanges = [
'10.0.0.0/8',
'172.16.0.0/12',
'192.168.0.0/16',
'127.0.0.0/8',
'169.254.0.0/16',
'224.0.0.0/4',
];
return privateRanges.some(range => IP.cidrSubnet(ip, range));
}
private async performSecurityChecks(url: URL): Promise<URLValidationResult> {
if (url.protocol === 'file:') {
return this.createInvalidResult(
'FILE_PROTOCOL_BLOCKED',
'File protocol is not allowed'
);
}
const suspiciousParams = [
'redirect',
'url',
'callback',
'return',
'dest',
'destination',
];
for (const param of suspiciousParams) {
if (url.searchParams.has(param)) {
const paramValue = url.searchParams.get(param);
if (paramValue && this.isSuspiciousURL(paramValue)) {
return this.createInvalidResult(
'SUSPICIOUS_PARAMETER',
`Suspicious parameter ${param} detected`
);
}
}
}
if (url.toString().length > 2048) {
return this.createInvalidResult(
'URL_TOO_LONG',
'URL exceeds maximum allowed length'
);
}
return { isValid: true };
}
private isSuspiciousURL(url: string): boolean {
const suspiciousPatterns = [
/localhost/i,
/127\.0\.0\.1/,
/0x[0-9a-f]+/i,
/internal/i,
/private/i,
/metadata/i,
/169\.254\./,
/file:\/\//,
];
return suspiciousPatterns.some(pattern => pattern.test(url));
}
async makeSecureRequest(url: string, options: RequestOptions = {}): Promise<SecureRequestResult> {
const validationResult = await this.validateURL(url);
if (!validationResult.isValid) {
throw new SSRFError('URL_VALIDATION_FAILED', validationResult.reason || 'Invalid URL');
}
const cacheKey = this.generateCacheKey(url, options);
const cached = this.requestCache.get(cacheKey);
if (cached && !this.isCacheExpired(cached)) {
return cached.result;
}
try {
const startTime = Date.now();
const response = await this.makeRequest(url, {
timeout: this.config.timeoutMs,
maxRedirects: this.config.maxRedirects,
...options,
});
const responseTime = Date.now() - startTime;
const responseValidation = await this.validateResponse(response);
if (!responseValidation.isValid) {
throw new SSRFError('RESPONSE_VALIDATION_FAILED', responseValidation.reason || 'Invalid response');
}
const result: SecureRequestResult = {
success: true,
data: response.data,
status: response.status,
headers: response.headers,
responseTime,
url: validationResult.normalizedURL,
};
this.requestCache.set(cacheKey, {
result,
timestamp: Date.now(),
});
this.logger.log('SECURE_REQUEST_SUCCESS', {
url: validationResult.normalizedURL,
responseTime,
status: response.status,
riskScore: validationResult.riskScore,
});
return result;
} catch (error) {
this.logger.log('SECURE_REQUEST_FAILED', {
url: url,
error: error.message,
type: error.name,
});
throw error;
}
}
private async makeRequest(url: string, options: RequestOptions): Promise<RequestResponse> {
const axios = require('axios');
const response = await axios({
url,
method: options.method || 'GET',
timeout: options.timeout,
maxRedirects: options.maxRedirects,
headers: {
'User-Agent': 'SecureRequest/1.0',
...options.headers,
},
responseType: 'arraybuffer',
maxContentLength: this.config.maxResponseSize,
});
return {
data: response.data,
status: response.status,
headers: response.headers,
};
}
private async validateResponse(response: RequestResponse): Promise<ValidationResult> {
if (Buffer.byteLength(response.data) > this.config.maxResponseSize) {
return {
isValid: false,
reason: `Response size exceeds maximum allowed size`,
};
}
const contentType = response.headers['content-type'];
if (contentType && this.isSuspiciousContentType(contentType)) {
return {
isValid: false,
reason: `Suspicious content type: ${contentType}`,
};
}
return { isValid: true };
}
private isSuspiciousContentType(contentType: string): boolean {
const suspiciousTypes = [
'application/octet-stream',
'application/x-executable',
'application/x-msdownload',
'application/x-msdos-program',
];
return suspiciousTypes.some(type => contentType.includes(type));
}
private calculateRiskScore(url: URL): number {
let score = 0;
if (url.toString().length > 1000) score += 10;
if (url.toString().length > 2000) score += 20;
if (url.protocol !== 'https:') score += 5;
if (url.hostname.split('.').length > 4) score += 5;
if (url.searchParams.size > 10) score += 5;
if (url.searchParams.size > 20) score += 15;
return Math.min(score, 100);
}
private createInvalidResult(reason: string, details?: string): URLValidationResult {
return {
isValid: false,
reason,
details,
riskScore: 100,
};
}
}
export class SSRFError extends Error {
constructor(
public code: string,
message: string,
public details?: any
) {
super(message);
this.name = 'SSRFError';
}
}
interface URLValidationResult {
isValid: boolean;
normalizedURL?: string;
reason?: string;
details?: string;
riskScore?: number;
}
interface SecureRequestResult {
success: boolean;
data: Buffer;
status: number;
headers: Record<string, string>;
responseTime: number;
url: string;
}
interface RequestResponse {
data: Buffer;
status: number;
headers: Record<string, string>;
}
interface RequestOptions {
method?: string;
timeout?: number;
maxRedirects?: number;
headers?: Record<string, string>;
}
interface ValidationResult {
isValid: boolean;
reason?: string;
}
interface RequestCache {
result: SecureRequestResult;
timestamp: number;
}
import ipaddress
import socket
from typing import List, Set, Tuple
class NetworkSegmentation:
def __init__(self):
self.allowed_networks = self._initialize_allowed_networks()
self.blocked_networks = self._initialize_blocked_networks()
self.gateway_rules = self._initialize_gateway_rules()
def _initialize_allowed_networks(self) -> List[ipaddress.IPNetwork]:
"""Initialize allowed network ranges for external requests."""
return [
ipaddress.IPNetwork('0.0.0.0/0'),
]
def _initialize_blocked_networks(self) -> List[ipaddress.IPNetwork]:
"""Initialize blocked network ranges to prevent internal access."""
return [
ipaddress.IPNetwork('10.0.0.0/8'),
ipaddress.IPNetwork('172.16.0.0/12'),
ipaddress.IPNetwork('192.168.0.0/16'),
ipaddress.IPNetwork('127.0.0.0/8'),
ipaddress.IPNetwork('169.254.0.0/16'),
ipaddress.IPNetwork('224.0.0.0/4'),
ipaddress.IPNetwork('240.0.0.0/4'),
ipaddress.IPNetwork('::1/128'),
ipaddress.IPNetwork('fc00::/7'),
ipaddress.IPNetwork('fe80::/10'),
]
def is_ip_allowed(self, ip_address: str) -> Tuple[bool, str]:
"""Check if an IP address is allowed for external requests."""
try:
ip = ipaddress.ip_address(ip_address)
for blocked_network in self.blocked_networks:
if ip in blocked_network:
return False, f"IP {ip_address} is in blocked network {blocked_network}"
for allowed_network in self.allowed_networks:
if ip in allowed_network:
return True, f"IP {ip_address} is in allowed network {allowed_network}"
return False, f"IP {ip_address} is not in any allowed network"
except ValueError:
return False, f"Invalid IP address: {ip_address}"
def validate_network_access(self, host: str, port: int) -> NetworkValidationResult:
"""Validate network access to a specific host and port."""
try:
addresses = socket.getaddrinfo(host, port, socket.AF_UNSPEC, socket.SOCK_STREAM)
for addr_info in addresses:
family, socktype, proto, canonname, sockaddr = addr_info
ip_address = sockaddr[0]
is_allowed, reason = self.is_ip_allowed(ip_address)
if not is_allowed:
return NetworkValidationResult(
is_valid=False,
reason=reason,
blocked_ip=ip_address,
)
return NetworkValidationResult(
is_valid=True,
resolved_addresses=[addr[4][0] for addr in addresses],
)
except socket.gaierror as e:
return NetworkValidationResult(
is_valid=False,
reason=f"DNS resolution failed: {str(e)}",
)
def create_network_filter(self) -> 'NetworkFilter':
"""Create a network filter for use in firewalls or proxies."""
return NetworkFilter(
allowed_networks=self.allowed_networks,
blocked_networks=self.blocked_networks,
gateway_rules=self.gateway_rules,
)
class NetworkValidationResult:
def __init__(self, is_valid: bool, reason: str = None,
blocked_ip: str = None, resolved_addresses: List[str] = None):
self.is_valid = is_valid
self.reason = reason
self.blocked_ip = blocked_ip
self.resolved_addresses = resolved_addresses or []
class NetworkFilter:
def __init__(self, allowed_networks: List[ipaddress.IPNetwork],
blocked_networks: List[ipaddress.IPNetwork],
gateway_rules: List[dict]):
self.allowed_networks = allowed_networks
self.blocked_networks = blocked_networks
self.gateway_rules = gateway_rules
def to_firewall_rules(self) -> List[str]:
"""Convert to firewall rules format."""
rules = []
for network in self.allowed_networks:
rules.append(f"ALLOW {network}")
for network in self.blocked_networks:
rules.append(f"BLOCK {network}")
return rules
