import threading import time import logging import ssl import os import uuid from http.server import HTTPServer, BaseHTTPRequestHandler from typing import Dict, Any from plugins import DriverPlugin from core.log_utils import sanitize_for_log # 配置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class HTTPHoneypotHandler(BaseHTTPRequestHandler): '''HTTP 蜜罐请求处理器''' def do_GET(self): self._handle_request('GET') def do_POST(self): self._handle_request('POST') def do_PUT(self): self._handle_request('PUT') def do_DELETE(self): self._handle_request('DELETE') def _handle_request(self, method): '''处理请求并记录日志''' # 记录请求信息 safe_path = sanitize_for_log(self.path) safe_ip = sanitize_for_log(self.client_address[0]) logger.info(f"Received {method} request for {safe_path} from {safe_ip}") # 发送一个简单的响应 self.send_response(200) self.send_header('Content-type', 'text/html; charset=utf-8') self.end_headers() response_content = f"

欢迎访问蜜罐!

您请求的路径是: {self.path}

" self.wfile.write(response_content.encode('utf-8')) def log_message(self, format, *args): # Override default logging to use our logger safe_addr = sanitize_for_log(self.address_string()) safe_fmt = sanitize_for_log(format % args) logger.info("%s - - [%s] %s\n" % (safe_addr, self.log_date_time_string(), safe_fmt)) class HTTPHoneypotDriver(DriverPlugin): ''' HTTP 蜜罐驱动插件实现 ''' def __init__(self): self.environments = {} # 存储环境信息 self.servers = {} # 存储运行中的服务器实例 def create_environment(self, env_config: Dict[str, Any]) -> str: '''创建HTTP蜜罐环境''' # 使用 uuid4 生成唯一ID，避免时间戳可能的冲突 env_id = f"http_env_{uuid.uuid4().hex}" self.environments[env_id] = { 'id': env_id, 'config': env_config, 'status': 'created' } logger.info(f"Created HTTP honeypot environment {env_id} with config {env_config}") return env_id def start_environment(self, env_id: str) -> bool: '''启动HTTP蜜罐环境''' env = self.environments.get(env_id) if not env: logger.error(f"Environment {env_id} not found") return False if env['status'] == 'running': logger.warning(f"Environment {env_id} is already running") return False try: config = env['config'] host = config.get('host', '0.0.0.0') port = config.get('port', 8080) ssl_config = config.get('ssl', None) # 创建并启动HTTP服务器 server = HTTPServer((host, port), HTTPHoneypotHandler) # 配置SSL（如果提供） if ssl_config: certfile = ssl_config.get('certfile') keyfile = ssl_config.get('keyfile') if certfile and keyfile and os.path.exists(certfile) and os.path.exists(keyfile): server.socket = ssl.wrap_socket(server.socket, certfile=certfile, keyfile=keyfile, server_side=True) logger.info(f"SSL enabled for environment {env_id}") else: logger.warning(f"Invalid SSL configuration for environment {env_id}. Starting without SSL.") server_thread = threading.Thread(target=server.serve_forever, daemon=True) server_thread.start() # 保存服务器实例 self.servers[env_id] = { 'server': server, 'thread': server_thread } env['status'] = 'running' logger.info(f"Started HTTP honeypot environment {env_id} on {host}:{port}{' (HTTPS)' if ssl_config else ' (HTTP)'}") return True except Exception as e: logger.error(f"Failed to start HTTP honeypot environment {env_id}: {e}") return False def stop_environment(self, env_id: str) -> bool: '''停止HTTP蜜罐环境''' env = self.environments.get(env_id) if not env: logger.error(f"Environment {env_id} not found") return False if env['status'] != 'running': logger.warning(f"Environment {env_id} is not running") return False try: server_info = self.servers.get(env_id) if server_info: server = server_info['server'] server.shutdown() server.server_close() del self.servers[env_id] env['status'] = 'stopped' logger.info(f"Stopped HTTP honeypot environment {env_id}") return True except Exception as e: logger.error(f"Failed to stop HTTP honeypot environment {env_id}: {e}") return False def delete_environment(self, env_id: str) -> bool: '''删除HTTP蜜罐环境''' env = self.environments.get(env_id) if not env: logger.error(f"Environment {env_id} not found") return False # 如果环境正在运行，先停止它 if env['status'] == 'running': self.stop_environment(env_id) del self.environments[env_id] logger.info(f"Deleted HTTP honeypot environment {env_id}") return True def get_environment_status(self, env_id: str) -> str: '''获取环境状态''' env = self.environments.get(env_id) if not env: return 'not_found' return env['status'] def list_environments(self) -> Dict[str, Any]: '''列出所有环境''' return self.environments