Advanced HTTP load test with realistic patterns
(self, target_ip, target_port, duration)
| 295 | return session |
| 296 | |
| 297 | def realistic_http_flood(self, target_ip, target_port, duration): |
| 298 | """Advanced HTTP load test with realistic patterns""" |
| 299 | end_time = time.time() + duration |
| 300 | session = self.create_http_session() |
| 301 | scheme = "https" if target_port == 443 else "http" |
| 302 | base_url = f"{scheme}://{target_ip}:{target_port}" |
| 303 | |
| 304 | # Realistic endpoints for a website |
| 305 | endpoints = [ |
| 306 | '/', '/home', '/index.html', '/about', '/contact', |
| 307 | '/products', '/services', '/blog', '/api/health', |
| 308 | '/static/css/main.css', '/static/js/app.js', |
| 309 | '/images/logo.png', '/api/v1/users', '/api/v1/products' |
| 310 | ] |
| 311 | |
| 312 | # Realistic request patterns |
| 313 | request_patterns = [ |
| 314 | {'method': 'GET', 'endpoint': random.choice(endpoints)}, |
| 315 | {'method': 'POST', 'endpoint': '/api/v1/login', 'data': {'username': 'test', 'password': 'test'}}, |
| 316 | {'method': 'GET', 'endpoint': '/api/v1/products'}, |
| 317 | {'method': 'HEAD', 'endpoint': random.choice(endpoints)}, |
| 318 | ] |
| 319 | |
| 320 | while time.time() < end_time and self.running: |
| 321 | try: |
| 322 | pattern = random.choice(request_patterns) |
| 323 | url = f"{base_url}{pattern['endpoint']}" |
| 324 | |
| 325 | start_time = time.time() |
| 326 | |
| 327 | if pattern['method'] == 'GET': |
| 328 | response = session.get(url, timeout=self.timeout) |
| 329 | elif pattern['method'] == 'POST': |
| 330 | response = session.post(url, json=pattern.get('data', {}), timeout=self.timeout) |
| 331 | elif pattern['method'] == 'HEAD': |
| 332 | response = session.head(url, timeout=self.timeout) |
| 333 | |
| 334 | response_time = (time.time() - start_time) * 1000 # Convert to milliseconds |
| 335 | |
| 336 | self.stats['requests_sent'] += 1 |
| 337 | self.stats['total_bytes'] += len(response.content) if hasattr(response, 'content') else 0 |
| 338 | self.stats['response_times'].append(response_time) |
| 339 | |
| 340 | if 200 <= response.status_code < 400: |
| 341 | self.stats['requests_successful'] += 1 |
| 342 | else: |
| 343 | self.stats['requests_failed'] += 1 |
| 344 | |
| 345 | # Adaptive rate limiting based on response |
| 346 | if response.status_code == 429: # Too Many Requests |
| 347 | time.sleep(1) |
| 348 | elif response.status_code >= 500: |
| 349 | time.sleep(0.1) |
| 350 | |
| 351 | except requests.exceptions.RequestException as e: |
| 352 | self.stats['requests_failed'] += 1 |
| 353 | self.stats['requests_sent'] += 1 |
| 354 | except Exception as e: |
nothing calls this directly
no test coverage detected