#!/usr/bin/env python3 """ Proxy WebSocket Server Servidor WebSocket para eventos em tempo real do IP Whitelist Proxy Porta: 8889 Uso: python3 proxy-websocket-server.py [--port 8889] [--proxy-url http://localhost:8888] Eventos enviados: - connected: Conexão estabelecida - stats: Estatísticas atualizadas (a cada 5s) - request: Nova requisição processada - blocked: Requisição bloqueada - whitelist_change: Whitelist modificada """ import argparse import asyncio import json import logging import signal import sys from datetime import datetime from typing import Set, Dict, Any import urllib.request import urllib.error # Tenta importar websockets, se não tiver, mostra instrução try: import websockets from websockets.server import serve except ImportError: print("Erro: Módulo 'websockets' não encontrado.") print("Instale com: pip3 install websockets") sys.exit(1) # Configuração de logging logging.basicConfig( level=logging.INFO, format='%(asctime)s | %(levelname)s | %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) logger = logging.getLogger('ws-server') # Configurações DEFAULT_PORT = 8889 DEFAULT_PROXY_URL = "http://localhost:8888" # Clientes conectados connected_clients: Set[websockets.WebSocketServerProtocol] = set() # Cache do último estado last_stats: Dict[str, Any] = {} last_history_count: int = 0 async def broadcast(event_type: str, data: Dict[str, Any]) -> None: """Envia evento para todos os clientes conectados""" if not connected_clients: return message = json.dumps({ 'type': event_type, 'timestamp': datetime.now().isoformat(), 'data': data }) # Cria lista de tarefas para envio paralelo tasks = [] disconnected = set() for client in connected_clients: try: tasks.append(asyncio.create_task(client.send(message))) except websockets.exceptions.ConnectionClosed: disconnected.add(client) # Remove clientes desconectados connected_clients.difference_update(disconnected) # Aguarda todos os envios if tasks: await asyncio.gather(*tasks, return_exceptions=True) async def fetch_proxy_stats(proxy_url: str) -> Dict[str, Any]: """Busca estatísticas do proxy via REST""" try: url = f"{proxy_url}/_proxy/stats" loop = asyncio.get_event_loop() def do_request(): with urllib.request.urlopen(url, timeout=5) as response: return json.loads(response.read().decode()) return await loop.run_in_executor(None, do_request) except Exception as e: logger.warning(f"Erro ao buscar stats: {e}") return {} async def fetch_proxy_history(proxy_url: str, limit: int = 10) -> list: """Busca últimas requisições do proxy""" try: url = f"{proxy_url}/_proxy/history?limit={limit}" loop = asyncio.get_event_loop() def do_request(): with urllib.request.urlopen(url, timeout=5) as response: data = json.loads(response.read().decode()) return data.get('data', []) return await loop.run_in_executor(None, do_request) except Exception as e: logger.warning(f"Erro ao buscar history: {e}") return [] async def stats_broadcaster(proxy_url: str, interval: float = 5.0) -> None: """Task que envia stats periodicamente""" global last_stats, last_history_count while True: try: await asyncio.sleep(interval) if not connected_clients: continue # Busca stats atuais stats = await fetch_proxy_stats(proxy_url) if not stats: continue # Verifica se houve mudança significativa current_total = stats.get('proxy', {}).get('requests_total', 0) last_total = last_stats.get('proxy', {}).get('requests_total', 0) # Envia stats await broadcast('stats', stats) last_stats = stats # Verifica novas requisições if current_total > last_total: # Busca últimas requisições new_count = current_total - last_total history = await fetch_proxy_history(proxy_url, limit=min(new_count, 10)) for req in history: if req.get('was_blocked'): await broadcast('blocked', req) else: await broadcast('request', req) last_history_count = stats.get('total_logged', 0) except asyncio.CancelledError: break except Exception as e: logger.error(f"Erro no broadcaster: {e}") async def handle_client(websocket: websockets.WebSocketServerProtocol, proxy_url: str) -> None: """Handler para cada cliente WebSocket""" client_ip = websocket.remote_address[0] if websocket.remote_address else 'unknown' logger.info(f"Cliente conectado: {client_ip}") connected_clients.add(websocket) try: # Envia evento de conexão com stats iniciais stats = await fetch_proxy_stats(proxy_url) await websocket.send(json.dumps({ 'type': 'connected', 'timestamp': datetime.now().isoformat(), 'data': { 'message': 'Conectado ao Proxy WebSocket Server', 'stats': stats } })) # Mantém conexão aberta e processa mensagens async for message in websocket: try: data = json.loads(message) cmd = data.get('command') if cmd == 'ping': await websocket.send(json.dumps({ 'type': 'pong', 'timestamp': datetime.now().isoformat() })) elif cmd == 'get_stats': stats = await fetch_proxy_stats(proxy_url) await websocket.send(json.dumps({ 'type': 'stats', 'timestamp': datetime.now().isoformat(), 'data': stats })) elif cmd == 'get_history': limit = data.get('limit', 50) history = await fetch_proxy_history(proxy_url, limit=limit) await websocket.send(json.dumps({ 'type': 'history', 'timestamp': datetime.now().isoformat(), 'data': history })) except json.JSONDecodeError: await websocket.send(json.dumps({ 'type': 'error', 'message': 'JSON inválido' })) except websockets.exceptions.ConnectionClosed: pass finally: connected_clients.discard(websocket) logger.info(f"Cliente desconectado: {client_ip}") async def main(port: int, proxy_url: str) -> None: """Função principal""" logger.info(f"Iniciando WebSocket Server na porta {port}") logger.info(f"Proxy URL: {proxy_url}") # Inicia broadcaster de stats broadcaster_task = asyncio.create_task(stats_broadcaster(proxy_url)) # Handler com proxy_url via closure async def handler(websocket): await handle_client(websocket, proxy_url) # Configura shutdown graceful stop = asyncio.Future() def signal_handler(): stop.set_result(None) loop = asyncio.get_event_loop() for sig in (signal.SIGINT, signal.SIGTERM): try: loop.add_signal_handler(sig, signal_handler) except NotImplementedError: # Windows não suporta add_signal_handler pass # Inicia servidor async with serve(handler, "0.0.0.0", port): print(f""" ╔══════════════════════════════════════════════════════════════╗ ║ PROXY WEBSOCKET SERVER ║ ╠══════════════════════════════════════════════════════════════╣ ║ WebSocket: ws://0.0.0.0:{port:<42}║ ║ Proxy: {proxy_url:<49}║ ╠══════════════════════════════════════════════════════════════╣ ║ Eventos: ║ ║ - connected: Conexão estabelecida ║ ║ - stats: Estatísticas (a cada 5s) ║ ║ - request: Nova requisição ║ ║ - blocked: Requisição bloqueada ║ ╠══════════════════════════════════════════════════════════════╣ ║ Comandos (enviar JSON): ║ ║ {{"command": "ping"}} ║ ║ {{"command": "get_stats"}} ║ ║ {{"command": "get_history", "limit": 50}} ║ ╚══════════════════════════════════════════════════════════════╝ Ctrl+C para parar """) try: await stop except asyncio.CancelledError: pass # Cancela broadcaster broadcaster_task.cancel() try: await broadcaster_task except asyncio.CancelledError: pass logger.info("Servidor encerrado") if __name__ == '__main__': parser = argparse.ArgumentParser(description='WebSocket Server para Proxy Monitor') parser.add_argument('-p', '--port', type=int, default=DEFAULT_PORT, help=f'Porta do WebSocket (default: {DEFAULT_PORT})') parser.add_argument('--proxy-url', type=str, default=DEFAULT_PROXY_URL, help=f'URL do proxy HTTP (default: {DEFAULT_PROXY_URL})') args = parser.parse_args() try: asyncio.run(main(args.port, args.proxy_url)) except KeyboardInterrupt: print("\nEncerrando...")