""" 简化版 Docker 执行服务 删除了所有 AI 相关功能，只保留核心的 Docker 部署功能 """ import os import subprocess import shutil import time import uuid import threading import re from typing import Dict, List, Optional from datetime import datetime, timedelta # 尝试导入yaml，如果失败则只使用正则表达式 try: import yaml HAS_YAML = True except ImportError: HAS_YAML = False class DockerExecutionService: """Docker执行服务，提供docker-compose命令的执行和管理功能""" def __init__(self, config=None): """初始化Docker执行服务""" self.config = config or self._get_default_config() self.deployment_root = self.config.get('deployment_root', '/tmp/ctf_deployments') self.compose_command = self.config.get('compose_command', 'docker-compose') self.command_timeout = self.config.get('command_timeout', 300) # 确保部署根目录存在 os.makedirs(self.deployment_root, exist_ok=True) # 部署记录缓存 self._deployment_cache = {} # 端口池管理 self.port_range_start = self.config.get('port_range_start', 42500) self.port_range_end = self.config.get('port_range_end', 42600) self._used_ports = set() # 初始化锁，用于并发控制 self._lock = threading.Lock() print(f"Docker执行服务初始化完成，部署根目录: {self.deployment_root}") def _get_default_config(self) -> Dict: """获取默认配置""" return { 'deployment_root': '/tmp/ctf_deployments', 'compose_command': 'docker-compose', 'command_timeout': 300, 'port_range_start': 42500, 'port_range_end': 42600, 'deployment_timeout': 86400, # 24小时 'auto_cleanup': True, 'cleanup_interval': 3600 } def _yield_line(self, message): """辅助函数：确保输出带换行符""" if not message.endswith('\n'): return message + '\n' return message def execute_deployment_stream(self, challenge_id: int, challenge_dir: str, options: Dict = None): """执行部署操作（流式输出版本） Args: challenge_id: 题目ID challenge_dir: 题目目录 options: 部署选项 Yields: str: 部署过程的实时输出（每行带换行符） """ options = options or {} if not os.path.exists(challenge_dir): yield self._yield_line(f"错误: 题目目录不存在: {challenge_dir}") return # 查找 Docker 配置文件 yield self._yield_line("查找Docker配置文件...") docker_files = self._find_docker_files(challenge_dir) if not docker_files: yield "错误: 题目目录中没有发现Dockerfile或docker-compose.yml文件" return docker_compose_path = docker_files.get('docker_compose') dockerfile_path = docker_files.get('dockerfile') if docker_compose_path: yield f"找到docker-compose.yml: {docker_compose_path}" elif dockerfile_path: yield f"找到Dockerfile: {dockerfile_path}" else: yield "错误: 未找到有效的Docker配置文件" return # 生成唯一部署ID deployment_id = str(uuid.uuid4()) # 生成安全的容器名称（使用UUID的一部分，避免冲突） challenge_name = f"challenge_{challenge_id}" # 使用UUID的前8位，确保唯一性 uuid_short = deployment_id.replace('-', '')[:8] container_name = f"{challenge_name}_{uuid_short}" container_name = ''.join(c if c.isalnum() else '_' for c in container_name) yield f"生成部署ID: {deployment_id}" yield f"容器名称: {container_name}" # 在部署开始时就创建部署记录（状态为creating） yield "创建部署记录..." try: from app.models.database.operations import create_deployment_record, add_deployment_log from datetime import datetime, timedelta # 计算过期时间 created_at = datetime.now() expires_at = created_at + timedelta(days=options.get('timeout_days', 1)) # 创建初始部署记录 initial_deployment_record = { 'deployment_uuid': deployment_id, 'challenge_id': challenge_id, 'user_id': options.get('user_id'), 'container_name': container_name, 'status': 'creating', # 初始状态为creating 'working_directory': challenge_dir, 'created_at': created_at.isoformat(), 'expires_at': expires_at.isoformat() } create_deployment_record(initial_deployment_record) add_deployment_log(deployment_id, 'creating', '部署开始，正在准备环境...') yield self._yield_line("部署记录已创建") except Exception as e: yield self._yield_line(f"警告: 创建部署记录失败: {str(e)}") import traceback traceback.print_exc() # 分配端口 port = options.get('port') external_port = None try: if port: # 检查指定端口是否可用 if self._check_port_available(port): external_port = port else: yield f"警告: 指定端口 {port} 不可用，将自动分配端口" external_port = self._allocate_port() else: external_port = self._allocate_port() if not external_port: yield "错误: 无法分配可用端口" # 更新部署状态为错误 try: from app.models.database.operations import update_deployment_status, add_deployment_log update_deployment_status(deployment_id, 'error', message='无法分配可用端口') add_deployment_log(deployment_id, 'error', '无法分配可用端口') except: pass return yield f"分配端口: {external_port}" # 使用 docker-compose 部署 if docker_compose_path: yield "使用docker-compose.yml部署" # 获取docker-compose.yml所在目录 docker_dir = os.path.dirname(docker_compose_path) # 修改端口映射 self._update_compose_port(docker_compose_path, external_port) yield f"已更新外部端口映射为: {external_port}" # 更新容器名称，避免冲突 self._update_compose_container_name(docker_compose_path, container_name) yield f"已更新容器名称为: {container_name}" # 执行 docker-compose up -d yield "开始执行docker-compose up命令..." # 使用相对路径，因为 cwd 已经设置为 docker_dir compose_file = os.path.basename(docker_compose_path) cmd = [ 'docker-compose', '-p', deployment_id, '-f', compose_file, 'up', '-d' ] yield f"执行命令: {' '.join(cmd)}" yield f"工作目录: {docker_dir}" # 执行命令并捕获输出 process = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, cwd=docker_dir ) # 实时输出日志 for line in process.stdout: yield line.rstrip() # 等待进程完成 return_code = process.wait() if return_code != 0: yield f"错误: docker-compose up 命令执行失败，返回码: {return_code}" self._release_port(external_port) # 更新部署状态为错误 try: from app.models.database.operations import update_deployment_status, add_deployment_log update_deployment_status(deployment_id, 'error', message=f'docker-compose up 命令执行失败，返回码: {return_code}') add_deployment_log(deployment_id, 'error', f'docker-compose up 命令执行失败，返回码: {return_code}') except: pass return yield "docker-compose up 命令执行成功" # 获取容器ID yield "获取容器ID..." container_id = self._get_container_id(deployment_id, docker_dir) if not container_id: yield "错误: 无法获取容器ID" self._release_port(external_port) # 更新部署状态为错误 try: from app.models.database.operations import update_deployment_status, add_deployment_log update_deployment_status(deployment_id, 'error', message='无法获取容器ID') add_deployment_log(deployment_id, 'error', '无法获取容器ID') except: pass return yield f"容器ID: {container_id}" # 生成访问地址 - 优先使用环境变量或请求中的host_address host_address = options.get('host_address') or os.environ.get('HOST_ADDRESS', 'localhost') access_url = f"http://{host_address}:{external_port}" yield f"访问地址: {access_url}" # 更新部署记录到数据库（记录已存在，只需更新状态和详细信息） yield "更新部署记录..." try: from app.models.database.operations import update_deployment_status, add_deployment_log # 更新部署记录状态和详细信息 update_deployment_status( deployment_id, 'running', container_id=container_id, external_port=external_port, access_url=access_url, working_directory=docker_dir ) add_deployment_log(deployment_id, 'running', f'部署成功完成，容器ID: {container_id}, 端口: {external_port}') yield self._yield_line("部署记录已更新") except Exception as e: yield self._yield_line(f"警告: 更新部署记录失败: {str(e)}") import traceback traceback.print_exc() # 返回成功信号（重要：DEPLOYMENT_SUCCESS 必须单独一行） yield self._yield_line("部署成功完成！") yield self._yield_line(f"DEPLOYMENT_SUCCESS:{deployment_id}:{container_id}:{external_port}:{access_url}") else: # 单独的Dockerfile yield "使用Dockerfile部署" yield "注意: 此功能暂未实现，请使用docker-compose.yml配置" self._release_port(external_port) except Exception as e: if external_port: self._release_port(external_port) yield f"部署过程中出错: {str(e)}" import traceback yield f"错误详情:\n{traceback.format_exc()}" # 更新部署状态为错误 try: from app.models.database.operations import update_deployment_status, add_deployment_log update_deployment_status(deployment_id, 'error', message=f'部署过程中出错: {str(e)}') add_deployment_log(deployment_id, 'error', f'部署过程中出错: {str(e)}') except: pass def _find_docker_files(self, directory: str) -> Dict[str, str]: """递归查找Docker配置文件""" result = {} for root, _, files in os.walk(directory): for file in files: if file == 'docker-compose.yml' or file == 'docker-compose.yaml': result['docker_compose'] = os.path.join(root, file) elif file == 'Dockerfile': if 'dockerfile' not in result: result['dockerfile'] = os.path.join(root, file) return result def _allocate_port(self) -> Optional[int]: """从端口池中分配一个可用端口""" with self._lock: for port in range(self.port_range_start, self.port_range_end): if port not in self._used_ports: if self._check_port_available(port): self._used_ports.add(port) return port return None def _release_port(self, port: int): """释放端口""" with self._lock: self._used_ports.discard(port) def _check_port_available(self, port: int) -> bool: """检查端口是否可用""" import socket try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind(('', port)) return True except OSError: return False def _update_compose_port(self, compose_path: str, external_port: int): """更新docker-compose.yml中的端口映射 支持多种端口映射格式： - "10000:80" - "10000:8080" - "10000:3000" - 0.0.0.0:10000:80 """ try: with open(compose_path, 'r') as f: content = f.read() import re # 匹配 ports 部分的端口映射，支持任意内部端口 # 格式: - "外部端口:内部端口" 或 - 外部端口:内部端口 或 - "IP:外部端口:内部端口" # 只替换第一个端口映射（外部端口） pattern = r'(\s+-\s+["\']?)(?:\d+\.\d+\.\d+\.\d+:)?(\d+)(:\d+["\']?)' replacement = rf'\g<1>{external_port}\g<3>' new_content = re.sub(pattern, replacement, content, count=1) with open(compose_path, 'w') as f: f.write(new_content) except Exception as e: print(f"更新端口映射失败: {str(e)}") def _update_compose_container_name(self, compose_path: str, container_name: str): """更新docker-compose.yml中的容器名称 避免容器名称冲突 """ try: with open(compose_path, 'r', encoding='utf-8') as f: content = f.read() # 尝试使用yaml解析 if HAS_YAML: try: data = yaml.safe_load(content) if data and 'services' in data: # 更新所有服务的容器名称 for service_name in data['services']: if 'container_name' in data['services'][service_name]: data['services'][service_name]['container_name'] = container_name # 写回文件 with open(compose_path, 'w', encoding='utf-8') as f: yaml.dump(data, f, default_flow_style=False, allow_unicode=True) return except Exception as e: print(f"YAML解析失败，使用正则表达式: {str(e)}") # 如果yaml解析失败，使用正则表达式 # 匹配 container_name: "name" 或 container_name: name pattern = r'(container_name\s*:\s*["\']?)([^"\'\n]+)(["\']?)' replacement = rf'\g<1>{container_name}\g<3>' new_content = re.sub(pattern, replacement, content) with open(compose_path, 'w', encoding='utf-8') as f: f.write(new_content) except Exception as e: print(f"更新容器名称失败: {str(e)}") def _get_container_id(self, deployment_id: str, docker_dir: str) -> Optional[str]: """获取容器ID""" try: result = subprocess.run( ['docker-compose', '-p', deployment_id, 'ps', '-q'], cwd=docker_dir, capture_output=True, text=True, timeout=30 ) if result.returncode == 0 and result.stdout.strip(): return result.stdout.strip().split('\n')[0] except Exception as e: print(f"获取容器ID失败: {str(e)}") return None def stop_deployment(self, deployment_uuid: str, working_directory: str) -> Dict: """停止部署""" try: # 转换为绝对路径 if not os.path.isabs(working_directory): working_directory = os.path.abspath(working_directory) # 检查工作目录是否存在 if not os.path.exists(working_directory): return {'status': 'error', 'message': f'工作目录不存在: {working_directory}'} result = subprocess.run( ['docker-compose', '-p', deployment_uuid, 'stop'], cwd=working_directory, capture_output=True, text=True, timeout=60 ) if result.returncode == 0: return {'status': 'success', 'message': '部署已停止'} else: return {'status': 'error', 'message': f'停止失败: {result.stderr}'} except Exception as e: return {'status': 'error', 'message': f'停止失败: {str(e)}'} def start_deployment(self, deployment_uuid: str, working_directory: str) -> Dict: """启动部署""" try: # 转换为绝对路径 if not os.path.isabs(working_directory): working_directory = os.path.abspath(working_directory) # 检查工作目录是否存在 if not os.path.exists(working_directory): return {'status': 'error', 'message': f'工作目录不存在: {working_directory}'} result = subprocess.run( ['docker-compose', '-p', deployment_uuid, 'start'], cwd=working_directory, capture_output=True, text=True, timeout=60 ) if result.returncode == 0: return {'status': 'success', 'message': '部署已启动'} else: return {'status': 'error', 'message': f'启动失败: {result.stderr}'} except Exception as e: return {'status': 'error', 'message': f'启动失败: {str(e)}'} def delete_deployment(self, deployment_uuid: str, working_directory: str, external_port: int = None) -> Dict: """删除部署""" try: # 转换为绝对路径 if not os.path.isabs(working_directory): working_directory = os.path.abspath(working_directory) # 检查工作目录是否存在 if not os.path.exists(working_directory): print(f"警告: 工作目录不存在: {working_directory}") # 即使目录不存在，也尝试停止容器 try: subprocess.run( ['docker-compose', '-p', deployment_uuid, 'down', '-v'], capture_output=True, text=True, timeout=60 ) except Exception as e: print(f"尝试停止容器失败: {str(e)}") # 释放端口 if external_port: self._release_port(external_port) return {'status': 'success', 'message': '部署已删除（工作目录不存在，已清理容器）'} # 停止并删除容器 result = subprocess.run( ['docker-compose', '-p', deployment_uuid, 'down', '-v'], cwd=working_directory, capture_output=True, text=True, timeout=60 ) # 释放端口 if external_port: self._release_port(external_port) # ⚠️ 重要：不删除工作目录！ # 工作目录包含题目的源文件，应该保留以便用户可以多次部署同一题目 # 只有在删除题目记录时才应该删除题目文件 # 这里只删除容器和释放端口 print(f"已停止并删除容器，保留工作目录: {working_directory}") if result.returncode == 0: return {'status': 'success', 'message': '部署已删除（容器已停止，题目文件已保留）'} else: # 即使docker-compose命令失败，如果是因为容器不存在，也认为删除成功 stderr = result.stderr.lower() if 'no such service' in stderr or 'not found' in stderr or 'no resource found' in stderr: return {'status': 'success', 'message': '部署已删除（容器不存在，题目文件已保留）'} return {'status': 'error', 'message': f'删除失败: {result.stderr}'} except Exception as e: return {'status': 'error', 'message': f'删除失败: {str(e)}'} def get_container_logs(self, deployment_uuid: str, working_directory: str, tail: int = 100) -> str: """获取容器日志""" try: # 确保工作目录是绝对路径 if not os.path.isabs(working_directory): # 尝试从项目根目录解析 base_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))) working_directory = os.path.join(base_dir, working_directory) if not os.path.exists(working_directory): return f"获取日志失败: 工作目录不存在 {working_directory}" result = subprocess.run( ['docker-compose', '-p', deployment_uuid, 'logs', '--tail', str(tail)], cwd=working_directory, capture_output=True, text=True, timeout=30 ) return result.stdout if result.returncode == 0 else result.stderr except Exception as e: return f"获取日志失败: {str(e)}" def get_container_status(self, deployment_uuid: str, working_directory: str) -> Dict: """获取容器状态""" try: result = subprocess.run( ['docker-compose', '-p', deployment_uuid, 'ps'], cwd=working_directory, capture_output=True, text=True, timeout=30 ) if result.returncode == 0: # 简单解析输出判断状态 output = result.stdout if 'Up' in output: return {'status': 'running', 'details': output} elif 'Exit' in output: return {'status': 'stopped', 'details': output} else: return {'status': 'unknown', 'details': output} else: return {'status': 'error', 'details': result.stderr} except Exception as e: return {'status': 'error', 'details': str(e)}