""" CTF题目文件管理服务 提供统一的文件操作接口，包括： - 文件完整性校验 - 文件存储和读取 - 目录管理 - 存储统计 """ import os import shutil import hashlib import json from pathlib import Path from typing import Dict, List, Optional, Tuple, Any from datetime import datetime import traceback class ChallengeFileManager: """题目文件管理器""" def __init__(self, base_path: str = "ge10"): """初始化文件管理器 Args: base_path: 基础存储路径（默认为 ge10，不再使用 ge10/output） """ self.base_path = Path(base_path) # 不再自动创建目录，因为现在使用 ge10/{category_id}/output 结构 # 只在需要时创建特定目录 def calculate_file_hash(self, filepath: str) -> str: """计算单个文件的MD5哈希值 Args: filepath: 文件路径 Returns: str: MD5哈希值 """ hash_md5 = hashlib.md5() try: with open(filepath, 'rb') as f: for chunk in iter(lambda: f.read(4096), b""): hash_md5.update(chunk) return hash_md5.hexdigest() except Exception as e: print(f"计算文件哈希失败 {filepath}: {e}") return "" def calculate_directory_hash(self, directory: str) -> str: """计算目录的整体哈希值 Args: directory: 目录路径 Returns: str: MD5哈希值 """ hash_md5 = hashlib.md5() dir_path = Path(directory) if not dir_path.exists(): return "" try: # 遍历所有文件，按路径排序以确保一致性 for filepath in sorted(dir_path.rglob('*')): if filepath.is_file(): # 添加相对路径到哈希 rel_path = filepath.relative_to(dir_path) hash_md5.update(str(rel_path).encode('utf-8')) # 添加文件内容到哈希 with open(filepath, 'rb') as f: for chunk in iter(lambda: f.read(4096), b""): hash_md5.update(chunk) return hash_md5.hexdigest() except Exception as e: print(f"计算目录哈希失败 {directory}: {e}") traceback.print_exc() return "" def get_directory_info(self, directory: str) -> Dict[str, Any]: """获取目录的详细信息 Args: directory: 目录路径 Returns: dict: 包含文件数量、总大小、哈希值等信息 """ dir_path = Path(directory) if not dir_path.exists(): return { 'exists': False, 'file_count': 0, 'total_size': 0, 'hash': '', 'files': [] } file_count = 0 total_size = 0 files = [] try: for filepath in dir_path.rglob('*'): if filepath.is_file(): file_count += 1 file_size = filepath.stat().st_size total_size += file_size rel_path = filepath.relative_to(dir_path) files.append({ 'path': str(rel_path), 'size': file_size, 'modified': datetime.fromtimestamp(filepath.stat().st_mtime).isoformat() }) return { 'exists': True, 'file_count': file_count, 'total_size': total_size, 'hash': self.calculate_directory_hash(directory), 'files': files } except Exception as e: print(f"获取目录信息失败 {directory}: {e}") traceback.print_exc() return { 'exists': True, 'file_count': 0, 'total_size': 0, 'hash': '', 'files': [], 'error': str(e) } def get_file_content(self, directory: str, relative_path: str) -> Optional[str]: """读取文件内容 Args: directory: 目录路径 relative_path: 相对路径 Returns: str: 文件内容，失败返回None """ try: filepath = Path(directory) / relative_path # 安全检查：确保文件在目录内 filepath = filepath.resolve() dir_path = Path(directory).resolve() if not str(filepath).startswith(str(dir_path)): print(f"安全检查失败：文件不在目录内 {filepath}") return None if not filepath.exists() or not filepath.is_file(): return None with open(filepath, 'r', encoding='utf-8', errors='ignore') as f: return f.read() except Exception as e: print(f"读取文件失败 {directory}/{relative_path}: {e}") return None def delete_directory(self, directory: str) -> bool: """删除目录及其所有内容 Args: directory: 目录路径 Returns: bool: 是否成功删除 """ try: dir_path = Path(directory) if dir_path.exists(): shutil.rmtree(dir_path) print(f"已删除目录: {directory}") return True return False except Exception as e: print(f"删除目录失败 {directory}: {e}") traceback.print_exc() return False def get_storage_stats(self) -> Dict[str, Any]: """获取存储统计信息 Returns: dict: 包含总大小、文件数量、目录数量等信息 """ total_size = 0 total_files = 0 total_dirs = 0 challenges = [] try: if not self.base_path.exists(): return { 'total_size': 0, 'total_files': 0, 'total_dirs': 0, 'challenges': [] } # 遍历所有题目目录 for challenge_dir in self.base_path.iterdir(): if challenge_dir.is_dir(): total_dirs += 1 dir_info = self.get_directory_info(str(challenge_dir)) total_size += dir_info['total_size'] total_files += dir_info['file_count'] challenges.append({ 'name': challenge_dir.name, 'path': str(challenge_dir), 'size': dir_info['total_size'], 'file_count': dir_info['file_count'], 'created': datetime.fromtimestamp(challenge_dir.stat().st_ctime).isoformat(), 'modified': datetime.fromtimestamp(challenge_dir.stat().st_mtime).isoformat() }) # 按修改时间排序 challenges.sort(key=lambda x: x['modified'], reverse=True) return { 'total_size': total_size, 'total_files': total_files, 'total_dirs': total_dirs, 'challenges': challenges, 'base_path': str(self.base_path) } except Exception as e: print(f"获取存储统计失败: {e}") traceback.print_exc() return { 'total_size': 0, 'total_files': 0, 'total_dirs': 0, 'challenges': [], 'error': str(e) } def verify_integrity(self, directory: str, expected_hash: str) -> bool: """验证目录完整性 Args: directory: 目录路径 expected_hash: 期望的哈希值 Returns: bool: 是否完整 """ actual_hash = self.calculate_directory_hash(directory) return actual_hash == expected_hash def list_challenges(self) -> List[Dict[str, Any]]: """列出所有题目目录 Returns: list: 题目目录列表 """ challenges = [] try: if not self.base_path.exists(): return [] for challenge_dir in self.base_path.iterdir(): if challenge_dir.is_dir(): dir_info = self.get_directory_info(str(challenge_dir)) challenges.append({ 'name': challenge_dir.name, 'path': str(challenge_dir), 'size': dir_info['total_size'], 'file_count': dir_info['file_count'], 'created': datetime.fromtimestamp(challenge_dir.stat().st_ctime).isoformat(), 'modified': datetime.fromtimestamp(challenge_dir.stat().st_mtime).isoformat() }) # 按修改时间排序 challenges.sort(key=lambda x: x['modified'], reverse=True) return challenges except Exception as e: print(f"列出题目目录失败: {e}") traceback.print_exc() return [] def format_size(self, size_bytes: int) -> str: """格式化文件大小 Args: size_bytes: 字节数 Returns: str: 格式化后的大小（如：1.5 MB） """ for unit in ['B', 'KB', 'MB', 'GB', 'TB']: if size_bytes < 1024.0: return f"{size_bytes:.2f} {unit}" size_bytes /= 1024.0 return f"{size_bytes:.2f} PB" # 创建全局实例（使用新的目录结构，不再自动创建 ge10/output） # 注意：全局实例主要用于兼容性，实际使用时应根据 category_id 创建特定实例 challenge_file_manager = ChallengeFileManager(base_path="ge10")