diff --git a/api/v2.py b/api/v2.py index 8d868f6..97247d9 100644 --- a/api/v2.py +++ b/api/v2.py @@ -1020,7 +1020,7 @@ class APIv2(APIv1): "path": rel_path, "type": mime_type, "size": size, - "size_formatted": self.format_file_size(size), + "size_formatted": format_file_size(size), "modified": datetime.fromtimestamp(mtime).isoformat(), "match_score": self._calculate_match_score(file_name, search_term, search_mode) } @@ -1222,7 +1222,7 @@ class APIv2(APIv1): "total_files": total_files, "total_dirs": total_dirs, "total_size": total_size, - "total_size_formatted": self.format_file_size(total_size) + "total_size_formatted": format_file_size(total_size) }, "file_types": dict(sorted(file_types.items(), key=lambda x: x[1], reverse=True)), "size_distribution": size_distribution, @@ -1238,7 +1238,7 @@ class APIv2(APIv1): "largest_file": { "name": largest_file[0] if largest_file else None, "size": largest_file[1] if largest_file else None, - "size_formatted": self.format_file_size(largest_file[1]) if largest_file else None + "size_formatted": format_file_size(largest_file[1]) if largest_file else None } }, "updated": datetime.now().isoformat() @@ -1263,7 +1263,7 @@ class APIv2(APIv1): "path": filepath, "name": os.path.basename(filepath), "size": info.st_size, - "size_formatted": self.format_file_size(info.st_size), + "size_formatted": format_file_size(info.st_size), "modified": datetime.fromtimestamp(info.st_mtime).isoformat(), "downloads": count }) @@ -2249,8 +2249,10 @@ class APIv2(APIv1): except Exception as e: handler.send_json_response({"error": str(e)}, 500) - def api_start_sync(self, handler, source_name): - """启动同步""" + def api_start_sync(self, handler, source_name=None): + """启动同步（兼容 v1 调用：无 source_name 时回退到 v1 行为）""" + if source_name is None: + return super().api_start_sync(handler) if hasattr(handler, 'sync_manager') and handler.sync_manager: task_id = handler.sync_manager.start_sync(source_name) if task_id: @@ -2268,8 +2270,10 @@ class APIv2(APIv1): else: handler.send_json_response({"error": "Sync manager not available"}, 500) - def api_stop_sync(self, handler, source_name): - """停止同步""" + def api_stop_sync(self, handler, source_name=None): + """停止同步（兼容 v1 调用：无 source_name 时回退到 v1 行为）""" + if source_name is None: + return super().api_stop_sync(handler) if hasattr(handler, 'sync_manager') and handler.sync_manager: handler.sync_manager.stop_all_tasks_for_source(source_name) handler.send_json_response({ @@ -2279,8 +2283,10 @@ class APIv2(APIv1): else: handler.send_json_response({"error": "Sync manager not available"}, 500) - def api_get_sync_status(self, handler, source_name): - """获取同步状态（真实数据）""" + def api_get_sync_status(self, handler, source_name=None): + """获取同步状态（真实数据；兼容 v1 调用）""" + if source_name is None: + return super().api_get_sync_status(handler) # source_name 格式: source_name/status，需要提取 if source_name.endswith('/status'): source_name = source_name[:-7] diff --git a/core/config.py b/core/config.py index cd58efe..0da1d13 100644 --- a/core/config.py +++ b/core/config.py @@ -198,9 +198,10 @@ class ConfigManager: if 'auth_pass' not in config: config['auth_pass'] = 'admin123' elif auth_type == 'token': - # 每次运行都重新生成标准的 token - import secrets - config['auth_token'] = secrets.token_hex(32) + # 只在缺失时生成 token，避免每次配置验证/更新都使已签发 token 失效 + if not config.get('auth_token'): + import secrets + config['auth_token'] = secrets.token_hex(32) # 验证上传大小配置 if 'max_upload_size' in config: diff --git a/core/database.py b/core/database.py index 0a9181c..1ac1447 100644 --- a/core/database.py +++ b/core/database.py @@ -1220,8 +1220,8 @@ class DatabaseManager: return True return False - def verify_user(self, username: str, password_hash: str) -> dict: - """验证用户登录""" + def verify_user(self, username: str, password: str) -> dict: + """验证用户登录（password 为明文密码，用 bcrypt 校验）""" with self.session() as session: user = session.query(UserRecord).filter_by(username=username).first() @@ -1236,8 +1236,8 @@ class DatabaseManager: if not user.enabled: return {'valid': False, 'reason': '账号已被禁用'} - # 验证密码 - if user.password_hash == password_hash: + # 验证密码（bcrypt.checkpw，恒定时间比较） + if self.verify_password(password, user.password_hash): # 登录成功 user.last_login = time.time() user.login_count = (user.login_count or 0) + 1 @@ -1291,16 +1291,22 @@ class DatabaseManager: print(f"[数据库] 已创建默认用户: {username}") def hash_password(self, password: str) -> str: - """使用 bcrypt 加密密码""" + """使用 bcrypt 加密密码（回退: 自描述格式 PBKDF2-HMAC-SHA256）""" try: import bcrypt salt = bcrypt.gensalt(rounds=12) return bcrypt.hashpw(password.encode('utf-8'), salt).decode('utf-8') except ImportError: - # 如果 bcrypt 不可用，回退到 SHA256 + # 如果 bcrypt 不可用，回退到 PBKDF2-HMAC-SHA256 + # 格式: $pbkdf2-sha256$$$ import hashlib - salt = hashlib.sha256(str(time.time()).encode()).hexdigest()[:16] - return hashlib.sha256((password + salt).encode()).hexdigest() + import os as _os + import base64 + iterations = 100_000 + salt = _os.urandom(16) + dk = hashlib.pbkdf2_hmac('sha256', password.encode('utf-8'), salt, iterations) + return (f"$pbkdf2-sha256${iterations}${base64.b64encode(salt).decode()}" + f"${base64.b64encode(dk).decode()}") def verify_password(self, password: str, password_hash: str) -> bool: """验证密码是否正确""" @@ -1308,19 +1314,21 @@ class DatabaseManager: import bcrypt return bcrypt.checkpw(password.encode('utf-8'), password_hash.encode('utf-8')) except ImportError: - # 如果 bcrypt 不可用，回退到 SHA256 验证 + # 回退: 只接受自描述 PBKDF2 格式，绝不接受明文 import hashlib - # 尝试直接比较（可能是旧格式） - if password_hash == password: - return True - # 尝试带 salt 验证 - for salt_len in range(1, 17): - salt = password_hash[:salt_len] if len(password_hash) > salt_len else "" - if len(salt) >= 4: - test_hash = hashlib.sha256((password + salt).encode()).hexdigest() - if test_hash == password_hash: - return True - return False + import base64 + import hmac + if not password_hash.startswith('$pbkdf2-sha256$'): + return False + try: + _, algo, iters_s, salt_b64, hash_b64 = password_hash.split('$') + iterations = int(iters_s) + salt = base64.b64decode(salt_b64) + expected = base64.b64decode(hash_b64) + dk = hashlib.pbkdf2_hmac('sha256', password.encode('utf-8'), salt, iterations) + return hmac.compare_digest(dk, expected) + except Exception: + return False def get_user_by_token(self, token: str) -> dict: """通过 token 获取用户信息""" diff --git a/core/mirror_sync.py b/core/mirror_sync.py index d719576..aed78d3 100644 --- a/core/mirror_sync.py +++ b/core/mirror_sync.py @@ -1278,6 +1278,15 @@ class MirrorSyncManager: """获取同步状态""" with self.sync_lock: return self.sync_status.copy() + + def get_source_status(self, name): + """获取指定同步源状态""" + with self.sync_lock: + return dict(self.sync_status.get(name, {})) + + def stop_all_tasks_for_source(self, name): + """停止指定同步源的所有任务（v2 API 兼容）""" + return self.stop_sync(name) def start(self): """启动同步管理器""" diff --git a/core/scheduler.py b/core/scheduler.py index b091731..69afd24 100644 --- a/core/scheduler.py +++ b/core/scheduler.py @@ -98,7 +98,10 @@ class ScheduledTask: # 处理列表: "1,2,3" if ',' in part: - return self._parse_cron_part(part.replace(',', ' '), min_val, max_val) + result = [] + for sub in part.split(','): + result.extend(self._parse_cron_part(sub.strip(), min_val, max_val)) + return result # 处理范围: "1-5" if '-' in part: @@ -144,12 +147,15 @@ class ScheduledTask: if not self.cron_parts: return False + # cron 约定: 0=周日...6=周六; datetime.weekday(): 0=周一...6=周日 + cron_weekday = (dt.weekday() + 1) % 7 + return ( dt.minute in self.cron_parts['minute'] and dt.hour in self.cron_parts['hour'] and dt.day in self.cron_parts['day'] and dt.month in self.cron_parts['month'] and - dt.weekday() in self.cron_parts['weekday'] + cron_weekday in self.cron_parts['weekday'] ) def get_next_run_time(self) -> Optional[datetime]: diff --git a/core/server.py b/core/server.py index ad34800..9fa0025 100644 --- a/core/server.py +++ b/core/server.py @@ -92,6 +92,8 @@ class MirrorServer: if self.config.get('enable_sync', True): self.sync_manager = MirrorSyncManager(self.config) self.sync_manager.start() + # 注入到配置，供 SyncScheduler 定时回调使用 + self.config['_sync_manager'] = self.sync_manager # 创建系统监控器（仅当启用时） self.monitor = None diff --git a/core/sync_scheduler.py b/core/sync_scheduler.py index 367cd06..9f4217e 100644 --- a/core/sync_scheduler.py +++ b/core/sync_scheduler.py @@ -52,6 +52,9 @@ class SyncScheduler: self.on_file_updated: Optional[Callable] = None self.on_sync_complete: Optional[Callable] = None + # 定时同步执行器（注入 MirrorSyncManager.start_sync 或等效 callable） + self.sync_starter: Optional[Callable] = None + # 待同步队列 self._pending_add = [] # 待添加的文件 self._pending_update = [] # 待更新的文件 @@ -97,10 +100,25 @@ class SyncScheduler: """创建同步回调函数""" def sync_callback(task_name: str, config: dict): logger.info(f"定时同步任务触发: {source_name}") - self.start_sync(source_name) - return True + # 委托给注入的同步执行器（MirrorSyncManager.start_sync） + starter = self.sync_starter or self._default_sync_starter() + if starter: + try: + return bool(starter(source_name)) + except Exception as e: + logger.error(f"触发同步 {source_name} 失败: {e}") + return False + logger.error(f"无法触发同步 {source_name}: 未配置同步执行器(sync_starter)") + return False return sync_callback + def _default_sync_starter(self): + """尝试从配置获取 MirrorSyncManager 实例（server.py 启动时注入）""" + mgr = self.config.get('_sync_manager') + if mgr and hasattr(mgr, 'start_sync'): + return mgr.start_sync + return None + def start(self): """启动同步调度器""" if self._running: diff --git a/mirrors/apt.py b/mirrors/apt.py index d930fd6..ad0e228 100644 --- a/mirrors/apt.py +++ b/mirrors/apt.py @@ -1,432 +1,436 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- - -""" -APT镜像代理处理器 -支持Debian/Ubuntu软件源 -""" - -import os -import json -import time -import gzip -import re -import urllib.request -from typing import Dict, List, Optional, Tuple -from datetime import datetime - - -class APTMirror: - """APT镜像代理""" - - def __init__(self, config: dict): - self.config = config - - # 配置 - 使用 storage_dir（基于 base_dir） - self.mirrors = config.get('mirrors', [ - 'http://archive.ubuntu.com/ubuntu', - 'http://security.ubuntu.com/ubuntu' - ]) - self.storage_dir = config.get('storage_dir', './downloads/apt') - self.base_dir = config.get('base_dir', './downloads') - self.default_suite = config.get('suite', 'jammy') - self.default_components = config.get('components', ['main', 'restricted', 'universe', 'multiverse']) - self.default_arch = config.get('arch', 'amd64') - - # 确保存储目录存在 - os.makedirs(self.storage_dir, exist_ok=True) - - def handle_request(self, handler, path: str) -> bool: - """ - 处理APT请求 - 路径格式: /ubuntu/dists/jammy/main/binary-amd64/Packages.gz - """ - try: - # 解析路径 - parts = path.strip('/').split('/') - - if len(parts) < 5: - # 返回镜像列表或帮助信息 - return self._handle_index(handler) - - # 提取组件 - distro = parts[0] # ubuntu, debian 等 - dist_type = parts[1] # dists - suite = parts[2] # jammy, focal 等 - component = parts[3] # main, updates 等 - rest = '/'.join(parts[4:]) - - # 确定请求类型 - if rest.endswith('Packages.gz'): - return self._handle_packages(handler, distro, suite, component, rest) - elif rest.endswith('Packages'): - return self._handle_packages_uncompressed(handler, distro, suite, component, rest) - elif rest.endswith('Release'): - return self._handle_release(handler, distro, suite, component, rest) - elif rest.endswith('Release.gpg'): - return self._handle_release_gpg(handler, distro, suite, component, rest) - elif rest.endswith('InRelease'): - return self._handle_inrelease(handler, distro, suite, component, rest) - else: - # 其他文件（源码包等） - return self._handle_file(handler, distro, suite, rest) - - except Exception as e: - handler.send_error(500, str(e)) - return False - - def _handle_index(self, handler) -> bool: - """处理索引请求""" - handler.send_json_response({ - 'mirrors': self.mirrors, - 'default_suite': self.default_suite, - 'default_components': self.default_components, - 'cache_stats': self.get_cache_stats() - }) - return True - - def _handle_packages(self, handler, distro: str, suite: str, component: str, path: str) -> bool: - """处理Packages.gz请求""" - cache_key = f"packages:{distro}:{suite}:{component}:{self.default_arch}" - - # 检查架构 - if 'binary-' in path: - arch = path.split('binary-')[1].split('/')[0] - else: - arch = self.default_arch - - cache_key = f"packages:{distro}:{suite}:{component}:{arch}" - - cached = self._get_cache(cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'application/x-gzip') - handler.send_header('Content-Length', str(len(cached))) - handler.end_headers() - handler.wfile.write(cached) - return True - - # 从上游获取 - for mirror in self.mirrors: - url = f"{mirror}/{path}" - - try: - data = self._fetch(url) - if data: - # 缓存 - if self.cache_enabled: - self._set_cache(cache_key, data) - - handler.send_response(200) - handler.send_header('Content-Type', 'application/x-gzip') - handler.send_header('Content-Length', str(len(data))) - handler.end_headers() - handler.wfile.write(data) - return True - - except Exception as e: - continue - - handler.send_error(502, "Failed to fetch from all mirrors") - return False - - def _handle_packages_uncompressed(self, handler, distro: str, suite: str, component: str, path: str) -> bool: - """处理未压缩的Packages文件""" - # 先获取gz版本 - gz_path = path + '.gz' - - for mirror in self.mirrors: - url = f"{mirror}/{gz_path}" - try: - data = self._fetch(url) - if data: - # 解压 - packages_data = gzip.decompress(data) - - handler.send_response(200) - handler.send_header('Content-Type', 'text/plain') - handler.send_header('Content-Length', str(len(packages_data))) - handler.end_headers() - handler.wfile.write(packages_data) - return True - - except Exception: - continue - - handler.send_error(502, "Failed to fetch packages") - return False - - def _handle_release(self, handler, distro: str, suite: str, component: str, path: str) -> bool: - """处理Release文件""" - cache_key = f"release:{distro}:{suite}" - - cached = self._get_cache(cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'text/plain') - handler.send_header('Content-Length', str(len(cached))) - handler.end_headers() - handler.wfile.write(cached) - return True - - # 获取Release文件 - release_path = f"/{distro}/dists/{suite}/Release" - - for mirror in self.mirrors: - url = mirror + release_path - try: - data = self._fetch(url) - if data: - if self.cache_enabled: - self._set_cache(cache_key, data) - - handler.send_response(200) - handler.send_header('Content-Type', 'text/plain') - handler.send_header('Content-Length', str(len(data))) - handler.end_headers() - handler.wfile.write(data) - return True - - except Exception: - continue - - handler.send_error(502, "Failed to fetch Release") - return False - - def _handle_release_gpg(self, handler, distro: str, suite: str, component: str, path: str) -> bool: - """处理Release.gpg文件""" - cache_key = f"release_gpg:{distro}:{suite}" - - cached = self._get_cache(cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'application/pgp-signature') - handler.end_headers() - handler.wfile.write(cached) - return True - - # 尝试获取 - gpg_path = f"/{distro}/dists/{suite}/Release.gpg" - - for mirror in self.mirrors: - url = mirror + gpg_path - try: - data = self._fetch(url) - if data: - if self.cache_enabled: - self._set_cache(cache_key, data) - - handler.send_response(200) - handler.send_header('Content-Type', 'application/pgp-signature') - handler.end_headers() - handler.wfile.write(data) - return True - - except Exception: - continue - - handler.send_error(404, "Release.gpg not found") - return False - - def _handle_inrelease(self, handler, distro: str, suite: str, component: str, path: str) -> bool: - """处理InRelease文件 - 获取或生成签名后的Release信息""" - cache_key = f"inrelease:{distro}:{suite}" - - cached = self._get_cache(cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'text/plain') - handler.send_header('Content-Length', str(len(cached))) - handler.end_headers() - handler.wfile.write(cached) - return True - - # 尝试从上游获取 InRelease - inrelease_path = f"/{distro}/dists/{suite}/InRelease" - - for mirror in self.mirrors: - url = mirror + inrelease_path - try: - data = self._fetch(url) - if data: - if self.cache_enabled: - self._set_cache(cache_key, data) - - handler.send_response(200) - handler.send_header('Content-Type', 'text/plain') - handler.send_header('Content-Length', str(len(data))) - handler.end_headers() - handler.wfile.write(data) - return True - except Exception: - continue - - # 如果没有 InRelease，尝试生成一个（基于 Release + 方括号注释） - # 注意：这不是有效的签名，但可以用于不验证签名的客户端 - release_cache_key = f"release:{distro}:{suite}" - release_data = self._get_cache(release_cache_key) - - if not release_data: - # 尝试获取 Release - release_path = f"/{distro}/dists/{suite}/Release" - for mirror in self.mirrors: - url = mirror + release_path - try: - release_data = self._fetch(url) - if release_data: - break - except Exception: - continue - - if release_data: - # 添加注释说明这是未签名的 Release - comment = f"# Note: This is a synthesized InRelease (original InRelease not available)\n" - inrelease_data = comment + release_data.decode('utf-8', errors='replace') - - if self.cache_enabled: - self._set_cache(cache_key, inrelease_data.encode()) - - handler.send_response(200) - handler.send_header('Content-Type', 'text/plain') - handler.send_header('Content-Length', str(len(inrelease_data))) - handler.end_headers() - handler.wfile.write(inrelease_data.encode()) - return True - - handler.send_error(502, "Failed to fetch InRelease") - return False - - def _handle_file(self, handler, distro: str, suite: str, path: str) -> bool: - """处理普通文件请求（如源码包）""" - cache_key = f"file:{distro}:{path.replace('/', ':')}" - - cached = self._get_cache(cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'application/octet-stream') - handler.send_header('Content-Length', str(len(cached))) - handler.end_headers() - handler.wfile.write(cached) - return True - - # 从上游获取 - for mirror in self.mirrors: - url = f"{mirror}/{path}" - try: - data = self._fetch(url) - if data: - if self.cache_enabled: - self._set_cache(cache_key, data) - - handler.send_response(200) - handler.send_header('Content-Type', 'application/octet-stream') - handler.send_header('Content-Length', str(len(data))) - handler.end_headers() - handler.wfile.write(data) - return True - - except Exception: - continue - - handler.send_error(404, "File not found") - return False - - def _fetch(self, url: str) -> Optional[bytes]: - """从URL获取数据""" - try: - req = urllib.request.Request(url) - req.add_header('User-Agent', 'APT-Mirror/1.0') - - with urllib.request.urlopen(req, timeout=30) as response: - return response.read() - - except Exception: - return None - - def _get_cache(self, cache_key: str) -> Optional[bytes]: - """获取缓存""" - if not self.cache_enabled: - return None - - cache_path = self._get_cache_path(cache_key) - meta_path = cache_path + '.meta' - - if not os.path.exists(cache_path): - return None - - # 检查过期 - if os.path.exists(meta_path): - try: - with open(meta_path, 'r') as f: - meta = json.load(f) - if time.time() > meta.get('expires', 0): - return None - except Exception: - pass - - try: - with open(cache_path, 'rb') as f: - return f.read() - except Exception: - return None - - def _set_cache(self, cache_key: str, data: bytes): - """设置缓存""" - cache_path = self._get_cache_path(cache_key) - meta_path = cache_path + '.meta' - - os.makedirs(os.path.dirname(cache_path), exist_ok=True) - - try: - with open(cache_path, 'wb') as f: - f.write(data) - - meta = { - 'cached_at': time.time(), - 'expires': time.time() + self.cache_ttl, - 'size': len(data) - } - - with open(meta_path, 'w') as f: - json.dump(meta, f) - - except Exception as e: - print(f"APT缓存写入失败: {e}") - - def _get_cache_path(self, cache_key: str) -> str: - """获取缓存路径""" - subdir = cache_key[:2] - return os.path.join(self.storage_dir, subdir, cache_key) - - def get_cache_stats(self) -> dict: - """获取缓存统计""" - if not os.path.exists(self.storage_dir): - return {'files': 0, 'size': 0} - - total_size = 0 - file_count = 0 - - for root, dirs, files in os.walk(self.storage_dir): - for f in files: - if not f.endswith('.meta'): - file_count += 1 - total_size += os.path.getsize(os.path.join(root, f)) - - return { - 'files': file_count, - 'size': total_size, - 'size_formatted': self._format_size(total_size) - } - - def _format_size(self, size_bytes: int) -> str: - """格式化文件大小""" - if size_bytes == 0: - return "0 B" - - units = ["B", "KB", "MB", "GB"] - i = 0 - while size_bytes >= 1024 and i < len(units) - 1: - size_bytes /= 1024.0 - i += 1 - - return f"{size_bytes:.2f} {units[i]}" +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +""" +APT镜像代理处理器 +支持Debian/Ubuntu软件源 +""" + +import os +import json +import time +import gzip +import re +import urllib.request +from typing import Dict, List, Optional, Tuple +from datetime import datetime + + +class APTMirror: + """APT镜像代理""" + + def __init__(self, config: dict): + self.config = config + + # 配置 - 使用 storage_dir（基于 base_dir） + self.mirrors = config.get('mirrors', [ + 'http://archive.ubuntu.com/ubuntu', + 'http://security.ubuntu.com/ubuntu' + ]) + self.storage_dir = config.get('storage_dir', './downloads/apt') + self.base_dir = config.get('base_dir', './downloads') + self.default_suite = config.get('suite', 'jammy') + self.default_components = config.get('components', ['main', 'restricted', 'universe', 'multiverse']) + self.default_arch = config.get('arch', 'amd64') + + # 缓存配置（缺失时默认启用，避免 AttributeError） + self.cache_enabled = config.get('cache_enabled', True) + self.cache_ttl = config.get('cache_ttl', 3600) + + # 确保存储目录存在 + os.makedirs(self.storage_dir, exist_ok=True) + + def handle_request(self, handler, path: str) -> bool: + """ + 处理APT请求 + 路径格式: /ubuntu/dists/jammy/main/binary-amd64/Packages.gz + """ + try: + # 解析路径 + parts = path.strip('/').split('/') + + if len(parts) < 5: + # 返回镜像列表或帮助信息 + return self._handle_index(handler) + + # 提取组件 + distro = parts[0] # ubuntu, debian 等 + dist_type = parts[1] # dists + suite = parts[2] # jammy, focal 等 + component = parts[3] # main, updates 等 + rest = '/'.join(parts[4:]) + + # 确定请求类型 + if rest.endswith('Packages.gz'): + return self._handle_packages(handler, distro, suite, component, rest) + elif rest.endswith('Packages'): + return self._handle_packages_uncompressed(handler, distro, suite, component, rest) + elif rest.endswith('Release'): + return self._handle_release(handler, distro, suite, component, rest) + elif rest.endswith('Release.gpg'): + return self._handle_release_gpg(handler, distro, suite, component, rest) + elif rest.endswith('InRelease'): + return self._handle_inrelease(handler, distro, suite, component, rest) + else: + # 其他文件（源码包等） + return self._handle_file(handler, distro, suite, rest) + + except Exception as e: + handler.send_error(500, str(e)) + return False + + def _handle_index(self, handler) -> bool: + """处理索引请求""" + handler.send_json_response({ + 'mirrors': self.mirrors, + 'default_suite': self.default_suite, + 'default_components': self.default_components, + 'cache_stats': self.get_cache_stats() + }) + return True + + def _handle_packages(self, handler, distro: str, suite: str, component: str, path: str) -> bool: + """处理Packages.gz请求""" + cache_key = f"packages:{distro}:{suite}:{component}:{self.default_arch}" + + # 检查架构 + if 'binary-' in path: + arch = path.split('binary-')[1].split('/')[0] + else: + arch = self.default_arch + + cache_key = f"packages:{distro}:{suite}:{component}:{arch}" + + cached = self._get_cache(cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'application/x-gzip') + handler.send_header('Content-Length', str(len(cached))) + handler.end_headers() + handler.wfile.write(cached) + return True + + # 从上游获取 + for mirror in self.mirrors: + url = f"{mirror}/{path}" + + try: + data = self._fetch(url) + if data: + # 缓存 + if self.cache_enabled: + self._set_cache(cache_key, data) + + handler.send_response(200) + handler.send_header('Content-Type', 'application/x-gzip') + handler.send_header('Content-Length', str(len(data))) + handler.end_headers() + handler.wfile.write(data) + return True + + except Exception as e: + continue + + handler.send_error(502, "Failed to fetch from all mirrors") + return False + + def _handle_packages_uncompressed(self, handler, distro: str, suite: str, component: str, path: str) -> bool: + """处理未压缩的Packages文件""" + # 先获取gz版本 + gz_path = path + '.gz' + + for mirror in self.mirrors: + url = f"{mirror}/{gz_path}" + try: + data = self._fetch(url) + if data: + # 解压 + packages_data = gzip.decompress(data) + + handler.send_response(200) + handler.send_header('Content-Type', 'text/plain') + handler.send_header('Content-Length', str(len(packages_data))) + handler.end_headers() + handler.wfile.write(packages_data) + return True + + except Exception: + continue + + handler.send_error(502, "Failed to fetch packages") + return False + + def _handle_release(self, handler, distro: str, suite: str, component: str, path: str) -> bool: + """处理Release文件""" + cache_key = f"release:{distro}:{suite}" + + cached = self._get_cache(cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'text/plain') + handler.send_header('Content-Length', str(len(cached))) + handler.end_headers() + handler.wfile.write(cached) + return True + + # 获取Release文件 + release_path = f"/{distro}/dists/{suite}/Release" + + for mirror in self.mirrors: + url = mirror + release_path + try: + data = self._fetch(url) + if data: + if self.cache_enabled: + self._set_cache(cache_key, data) + + handler.send_response(200) + handler.send_header('Content-Type', 'text/plain') + handler.send_header('Content-Length', str(len(data))) + handler.end_headers() + handler.wfile.write(data) + return True + + except Exception: + continue + + handler.send_error(502, "Failed to fetch Release") + return False + + def _handle_release_gpg(self, handler, distro: str, suite: str, component: str, path: str) -> bool: + """处理Release.gpg文件""" + cache_key = f"release_gpg:{distro}:{suite}" + + cached = self._get_cache(cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'application/pgp-signature') + handler.end_headers() + handler.wfile.write(cached) + return True + + # 尝试获取 + gpg_path = f"/{distro}/dists/{suite}/Release.gpg" + + for mirror in self.mirrors: + url = mirror + gpg_path + try: + data = self._fetch(url) + if data: + if self.cache_enabled: + self._set_cache(cache_key, data) + + handler.send_response(200) + handler.send_header('Content-Type', 'application/pgp-signature') + handler.end_headers() + handler.wfile.write(data) + return True + + except Exception: + continue + + handler.send_error(404, "Release.gpg not found") + return False + + def _handle_inrelease(self, handler, distro: str, suite: str, component: str, path: str) -> bool: + """处理InRelease文件 - 获取或生成签名后的Release信息""" + cache_key = f"inrelease:{distro}:{suite}" + + cached = self._get_cache(cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'text/plain') + handler.send_header('Content-Length', str(len(cached))) + handler.end_headers() + handler.wfile.write(cached) + return True + + # 尝试从上游获取 InRelease + inrelease_path = f"/{distro}/dists/{suite}/InRelease" + + for mirror in self.mirrors: + url = mirror + inrelease_path + try: + data = self._fetch(url) + if data: + if self.cache_enabled: + self._set_cache(cache_key, data) + + handler.send_response(200) + handler.send_header('Content-Type', 'text/plain') + handler.send_header('Content-Length', str(len(data))) + handler.end_headers() + handler.wfile.write(data) + return True + except Exception: + continue + + # 如果没有 InRelease，尝试生成一个（基于 Release + 方括号注释） + # 注意：这不是有效的签名，但可以用于不验证签名的客户端 + release_cache_key = f"release:{distro}:{suite}" + release_data = self._get_cache(release_cache_key) + + if not release_data: + # 尝试获取 Release + release_path = f"/{distro}/dists/{suite}/Release" + for mirror in self.mirrors: + url = mirror + release_path + try: + release_data = self._fetch(url) + if release_data: + break + except Exception: + continue + + if release_data: + # 添加注释说明这是未签名的 Release + comment = f"# Note: This is a synthesized InRelease (original InRelease not available)\n" + inrelease_data = comment + release_data.decode('utf-8', errors='replace') + + if self.cache_enabled: + self._set_cache(cache_key, inrelease_data.encode()) + + handler.send_response(200) + handler.send_header('Content-Type', 'text/plain') + handler.send_header('Content-Length', str(len(inrelease_data))) + handler.end_headers() + handler.wfile.write(inrelease_data.encode()) + return True + + handler.send_error(502, "Failed to fetch InRelease") + return False + + def _handle_file(self, handler, distro: str, suite: str, path: str) -> bool: + """处理普通文件请求（如源码包）""" + cache_key = f"file:{distro}:{path.replace('/', ':')}" + + cached = self._get_cache(cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'application/octet-stream') + handler.send_header('Content-Length', str(len(cached))) + handler.end_headers() + handler.wfile.write(cached) + return True + + # 从上游获取 + for mirror in self.mirrors: + url = f"{mirror}/{path}" + try: + data = self._fetch(url) + if data: + if self.cache_enabled: + self._set_cache(cache_key, data) + + handler.send_response(200) + handler.send_header('Content-Type', 'application/octet-stream') + handler.send_header('Content-Length', str(len(data))) + handler.end_headers() + handler.wfile.write(data) + return True + + except Exception: + continue + + handler.send_error(404, "File not found") + return False + + def _fetch(self, url: str) -> Optional[bytes]: + """从URL获取数据""" + try: + req = urllib.request.Request(url) + req.add_header('User-Agent', 'APT-Mirror/1.0') + + with urllib.request.urlopen(req, timeout=30) as response: + return response.read() + + except Exception: + return None + + def _get_cache(self, cache_key: str) -> Optional[bytes]: + """获取缓存""" + if not self.cache_enabled: + return None + + cache_path = self._get_cache_path(cache_key) + meta_path = cache_path + '.meta' + + if not os.path.exists(cache_path): + return None + + # 检查过期 + if os.path.exists(meta_path): + try: + with open(meta_path, 'r') as f: + meta = json.load(f) + if time.time() > meta.get('expires', 0): + return None + except Exception: + pass + + try: + with open(cache_path, 'rb') as f: + return f.read() + except Exception: + return None + + def _set_cache(self, cache_key: str, data: bytes): + """设置缓存""" + cache_path = self._get_cache_path(cache_key) + meta_path = cache_path + '.meta' + + os.makedirs(os.path.dirname(cache_path), exist_ok=True) + + try: + with open(cache_path, 'wb') as f: + f.write(data) + + meta = { + 'cached_at': time.time(), + 'expires': time.time() + self.cache_ttl, + 'size': len(data) + } + + with open(meta_path, 'w') as f: + json.dump(meta, f) + + except Exception as e: + print(f"APT缓存写入失败: {e}") + + def _get_cache_path(self, cache_key: str) -> str: + """获取缓存路径""" + subdir = cache_key[:2] + return os.path.join(self.storage_dir, subdir, cache_key) + + def get_cache_stats(self) -> dict: + """获取缓存统计""" + if not os.path.exists(self.storage_dir): + return {'files': 0, 'size': 0} + + total_size = 0 + file_count = 0 + + for root, dirs, files in os.walk(self.storage_dir): + for f in files: + if not f.endswith('.meta'): + file_count += 1 + total_size += os.path.getsize(os.path.join(root, f)) + + return { + 'files': file_count, + 'size': total_size, + 'size_formatted': self._format_size(total_size) + } + + def _format_size(self, size_bytes: int) -> str: + """格式化文件大小""" + if size_bytes == 0: + return "0 B" + + units = ["B", "KB", "MB", "GB"] + i = 0 + while size_bytes >= 1024 and i < len(units) - 1: + size_bytes /= 1024.0 + i += 1 + + return f"{size_bytes:.2f} {units[i]}" diff --git a/mirrors/docker.py b/mirrors/docker.py index 29a4bce..75a955e 100644 --- a/mirrors/docker.py +++ b/mirrors/docker.py @@ -1,359 +1,364 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- - -""" -Docker镜像代理处理器 -支持Docker Registry API v2 -""" - -import os -import json -import time -import uuid -import urllib.request -import base64 -import hashlib -import hmac -from typing import Dict, List, Optional, Tuple -from datetime import datetime - - -class DockerMirror: - """Docker镜像代理""" - - def __init__(self, config: dict): - self.config = config - - # 配置 - 使用 storage_dir（基于 base_dir） - self.registry_url = config.get('registry_url', 'https://registry-1.docker.io') - self.mirror_url = config.get('mirror_url', '') - self.storage_dir = config.get('storage_dir', './downloads/docker') - self.base_dir = config.get('base_dir', './downloads') - - # 认证（可选） - self.username = config.get('username') - self.password = config.get('password') - - # 确保存储目录存在 - os.makedirs(self.storage_dir, exist_ok=True) - - def handle_request(self, handler, path: str) -> bool: - """ - 处理Docker镜像请求 - 路径格式: /v2/library/ubuntu/tags/list 或 /v2/library/ubuntu/manifests/latest - """ - try: - # 解析路径 - parts = path.strip('/').split('/') - - if len(parts) < 2 or parts[0] != 'v2': - handler.send_error(400, "Invalid Docker API path") - return False - - # 提取组件 - if parts[1] == 'library': - # 官方镜像 - image = 'library/' + '/'.join(parts[2:-2]) if len(parts) > 4 else 'library/' + parts[2] - action = parts[-2] # tags 或 manifests - reference = parts[-1] - else: - # 非官方镜像 - image = '/'.join(parts[1:-2]) - action = parts[-2] - reference = parts[-1] - - # 根据操作类型处理 - if action == 'tags' and reference == 'list': - return self._handle_tag_list(handler, image.rstrip('/tags')) - elif action == 'manifests': - return self._handle_manifest(handler, image, reference) - elif action == 'blobs': - return self._handle_blob(handler, image, reference) - elif action == 'token': - return self._handle_token(handler) - else: - handler.send_error(404, "Unknown action") - return False - - except Exception as e: - handler.send_error(500, str(e)) - return False - - def _handle_tag_list(self, handler, image: str) -> bool: - """处理标签列表请求""" - cache_key = f"tags:{image}" - cached = self._get_cache(handler, cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'application/json') - handler.send_header('Content-Length', str(len(cached))) - handler.end_headers() - handler.wfile.write(cached) - return True - - # 从上游获取 - url = f"{self.registry_url}/v2/{image}/tags/list" - - try: - data = self._fetch_from_upstream(url) - - # 缓存 - if self.cache_enabled: - self._set_cache(cache_key, data) - - handler.send_response(200) - handler.send_header('Content-Type', 'application/json') - handler.send_header('Content-Length', str(len(data))) - handler.end_headers() - handler.wfile.write(data) - return True - - except Exception as e: - handler.send_error(500, f"Failed to fetch tags: {str(e)}") - return False - - def _handle_manifest(self, handler, image: str, reference: str) -> bool: - """处理清单请求""" - cache_key = f"manifest:{image}:{reference}" - cached = self._get_cache(handler, cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'application/vnd.docker.distribution.manifest.v2+json') - handler.send_header('Content-Length', str(len(cached))) - handler.send_header('Docker-Content-Digest', f"sha256:{hashlib.sha256(cached).hexdigest()}") - handler.end_headers() - handler.wfile.write(cached) - return True - - # 从上游获取 - url = f"{self.registry_url}/v2/{image}/manifests/{reference}" - - try: - req = urllib.request.Request(url) - req.add_header('Accept', 'application/vnd.docker.distribution.manifest.v2+json') - - if self.username and self.password: - auth = base64.b64encode(f"{self.username}:{self.password}".encode()).decode() - req.add_header('Authorization', f"Basic {auth}") - - with urllib.request.urlopen(req) as response: - data = response.read() - - # 缓存 - if self.cache_enabled: - self._set_cache(cache_key, data) - - digest = f"sha256:{hashlib.sha256(data).hexdigest()}" - - handler.send_response(200) - handler.send_header('Content-Type', 'application/vnd.docker.distribution.manifest.v2+json') - handler.send_header('Content-Length', str(len(data))) - handler.send_header('Docker-Content-Digest', digest) - handler.end_headers() - handler.wfile.write(data) - return True - - except Exception as e: - handler.send_error(500, f"Failed to fetch manifest: {str(e)}") - return False - - def _handle_blob(self, handler, image: str, digest: str) -> bool: - """处理Blob层下载""" - # 移除 sha256: 前缀 - if digest.startswith('sha256:'): - digest = digest[7:] - - cache_key = f"blob:{digest}" - cached = self._get_cache(handler, cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'application/octet-stream') - handler.send_header('Content-Length', str(len(cached))) - handler.send_header('Docker-Content-Digest', f"sha256:{digest}") - handler.end_headers() - handler.wfile.write(cached) - return True - - # 从上游获取 - url = f"{self.registry_url}/v2/{image}/blobs/sha256:{digest}" - - try: - req = urllib.request.Request(url) - - if self.username and self.password: - auth = base64.b64encode(f"{self.username}:{self.password}".encode()).decode() - req.add_header('Authorization', f"Basic {auth}") - - with urllib.request.urlopen(req) as response: - data = response.read() - - # 缓存 - if self.cache_enabled: - self._set_cache(cache_key, data) - - handler.send_response(200) - handler.send_header('Content-Type', 'application/octet-stream') - handler.send_header('Content-Length', str(len(data))) - handler.send_header('Docker-Content-Digest', f"sha256:{digest}") - handler.end_headers() - handler.wfile.write(data) - return True - - except Exception as e: - handler.send_error(500, f"Failed to fetch blob: {str(e)}") - return False - - def _handle_token(self, handler) -> bool: - """处理Token请求 - 生成真实的访问令牌""" - # 解析认证信息 - auth_header = handler.headers.get('Authorization', '') - username = None - password = None - - if auth_header.startswith('Basic '): - try: - decoded = base64.b64decode(auth_header[6:]).decode('utf-8') - username, password = decoded.split(':', 1) - except Exception: - pass - - # 验证凭据（如果有） - if self.username and self.password: - if username != self.username or password != self.password: - handler.send_error(401, "Invalid credentials") - return False - - # 生成唯一的访问令牌 - token_id = str(uuid.uuid4()) - issued_at = int(time.time()) - expires_in = 300 # 5分钟 - expires_at = issued_at + expires_in - - # 创建令牌信息（简化版 JWT 结构） - token_data = { - "iss": "hyc-mirror", - "sub": username or "anonymous", - "aud": self.registry_url, - "iat": issued_at, - "exp": expires_at, - "access": [ - {"type": "repository", "actions": ["pull"]}, - {"type": "registry", "actions": ["catalog"]} - ] - } - - # 使用 HMAC-SHA256 对令牌进行简单签名 - secret_key = f"hyc-mirror-{self.registry_url}".encode() - signature = hmac.new( - secret_key, - f"{token_id}:{issued_at}".encode(), - hashlib.sha256 - ).hexdigest()[:32] - - full_token = f"{token_id}-{signature}" - - handler.send_json_response({ - "token": full_token, - "expires_in": expires_in, - "issued_at": issued_at - }) - return True - - def _fetch_from_upstream(self, url: str) -> bytes: - """从上游获取数据""" - req = urllib.request.Request(url) - - if self.username and self.password: - auth = base64.b64encode(f"{self.username}:{self.password}".encode()).decode() - req.add_header('Authorization', f"Basic {auth}") - - with urllib.request.urlopen(req, timeout=30) as response: - return response.read() - - def _get_cache(self, handler, cache_key: str) -> Optional[bytes]: - """获取缓存""" - if not self.cache_enabled: - return None - - cache_path = self._get_cache_path(cache_key) - meta_path = cache_path + '.meta' - - if not os.path.exists(cache_path): - return None - - # 检查是否过期 - if os.path.exists(meta_path): - try: - with open(meta_path, 'r') as f: - meta = json.load(f) - if time.time() > meta.get('expires', 0): - return None - except Exception: - pass - - try: - with open(cache_path, 'rb') as f: - return f.read() - except Exception: - return None - - def _set_cache(self, cache_key: str, data: bytes): - """设置缓存""" - cache_path = self._get_cache_path(cache_key) - meta_path = cache_path + '.meta' - - os.makedirs(os.path.dirname(cache_path), exist_ok=True) - - try: - with open(cache_path, 'wb') as f: - f.write(data) - - meta = { - 'cached_at': time.time(), - 'expires': time.time() + self.cache_ttl, - 'size': len(data) - } - - with open(meta_path, 'w') as f: - json.dump(meta, f) - - except Exception as e: - print(f"Docker缓存写入失败: {e}") - - def _get_cache_path(self, cache_key: str) -> str: - """获取缓存路径""" - subdir = cache_key[:2] - return os.path.join(self.storage_dir, subdir, cache_key) - - def get_cache_stats(self) -> dict: - """获取缓存统计""" - if not os.path.exists(self.storage_dir): - return {'files': 0, 'size': 0} - - total_size = 0 - file_count = 0 - - for root, dirs, files in os.walk(self.storage_dir): - for f in files: - if not f.endswith('.meta'): - file_count += 1 - total_size += os.path.getsize(os.path.join(root, f)) - - return { - 'files': file_count, - 'size': total_size, - 'size_formatted': self._format_size(total_size) - } - - def _format_size(self, size_bytes: int) -> str: - """格式化文件大小""" - if size_bytes == 0: - return "0 B" - - units = ["B", "KB", "MB", "GB"] - i = 0 - while size_bytes >= 1024 and i < len(units) - 1: - size_bytes /= 1024.0 - i += 1 - - return f"{size_bytes:.2f} {units[i]}" +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +""" +Docker镜像代理处理器 +支持Docker Registry API v2 +""" + +import os +import json +import time +import uuid +import urllib.request +import base64 +import hashlib +import hmac +from typing import Dict, List, Optional, Tuple +from datetime import datetime + + +class DockerMirror: + """Docker镜像代理""" + + def __init__(self, config: dict): + self.config = config + + # 配置 - 使用 storage_dir（基于 base_dir）；兼容 v2 分支传入的 upstream_url 键 + self.registry_url = (config.get('registry_url') or config.get('upstream_url') + or 'https://registry-1.docker.io') + self.mirror_url = config.get('mirror_url', '') + self.storage_dir = config.get('storage_dir', './downloads/docker') + self.base_dir = config.get('base_dir', './downloads') + + # 认证（可选） + self.username = config.get('username') + self.password = config.get('password') + + # 缓存配置（缺失时默认启用，避免 AttributeError） + self.cache_enabled = config.get('cache_enabled', True) + self.cache_ttl = config.get('cache_ttl', 3600) + + # 确保存储目录存在 + os.makedirs(self.storage_dir, exist_ok=True) + + def handle_request(self, handler, path: str) -> bool: + """ + 处理Docker镜像请求 + 路径格式: /v2/library/ubuntu/tags/list 或 /v2/library/ubuntu/manifests/latest + """ + try: + # 解析路径 + parts = path.strip('/').split('/') + + if len(parts) < 2 or parts[0] != 'v2': + handler.send_error(400, "Invalid Docker API path") + return False + + # 提取组件 + if parts[1] == 'library': + # 官方镜像 + image = 'library/' + '/'.join(parts[2:-2]) if len(parts) > 4 else 'library/' + parts[2] + action = parts[-2] # tags 或 manifests + reference = parts[-1] + else: + # 非官方镜像 + image = '/'.join(parts[1:-2]) + action = parts[-2] + reference = parts[-1] + + # 根据操作类型处理 + if action == 'tags' and reference == 'list': + return self._handle_tag_list(handler, image.rstrip('/tags')) + elif action == 'manifests': + return self._handle_manifest(handler, image, reference) + elif action == 'blobs': + return self._handle_blob(handler, image, reference) + elif action == 'token': + return self._handle_token(handler) + else: + handler.send_error(404, "Unknown action") + return False + + except Exception as e: + handler.send_error(500, str(e)) + return False + + def _handle_tag_list(self, handler, image: str) -> bool: + """处理标签列表请求""" + cache_key = f"tags:{image}" + cached = self._get_cache(handler, cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'application/json') + handler.send_header('Content-Length', str(len(cached))) + handler.end_headers() + handler.wfile.write(cached) + return True + + # 从上游获取 + url = f"{self.registry_url}/v2/{image}/tags/list" + + try: + data = self._fetch_from_upstream(url) + + # 缓存 + if self.cache_enabled: + self._set_cache(cache_key, data) + + handler.send_response(200) + handler.send_header('Content-Type', 'application/json') + handler.send_header('Content-Length', str(len(data))) + handler.end_headers() + handler.wfile.write(data) + return True + + except Exception as e: + handler.send_error(500, f"Failed to fetch tags: {str(e)}") + return False + + def _handle_manifest(self, handler, image: str, reference: str) -> bool: + """处理清单请求""" + cache_key = f"manifest:{image}:{reference}" + cached = self._get_cache(handler, cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'application/vnd.docker.distribution.manifest.v2+json') + handler.send_header('Content-Length', str(len(cached))) + handler.send_header('Docker-Content-Digest', f"sha256:{hashlib.sha256(cached).hexdigest()}") + handler.end_headers() + handler.wfile.write(cached) + return True + + # 从上游获取 + url = f"{self.registry_url}/v2/{image}/manifests/{reference}" + + try: + req = urllib.request.Request(url) + req.add_header('Accept', 'application/vnd.docker.distribution.manifest.v2+json') + + if self.username and self.password: + auth = base64.b64encode(f"{self.username}:{self.password}".encode()).decode() + req.add_header('Authorization', f"Basic {auth}") + + with urllib.request.urlopen(req) as response: + data = response.read() + + # 缓存 + if self.cache_enabled: + self._set_cache(cache_key, data) + + digest = f"sha256:{hashlib.sha256(data).hexdigest()}" + + handler.send_response(200) + handler.send_header('Content-Type', 'application/vnd.docker.distribution.manifest.v2+json') + handler.send_header('Content-Length', str(len(data))) + handler.send_header('Docker-Content-Digest', digest) + handler.end_headers() + handler.wfile.write(data) + return True + + except Exception as e: + handler.send_error(500, f"Failed to fetch manifest: {str(e)}") + return False + + def _handle_blob(self, handler, image: str, digest: str) -> bool: + """处理Blob层下载""" + # 移除 sha256: 前缀 + if digest.startswith('sha256:'): + digest = digest[7:] + + cache_key = f"blob:{digest}" + cached = self._get_cache(handler, cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'application/octet-stream') + handler.send_header('Content-Length', str(len(cached))) + handler.send_header('Docker-Content-Digest', f"sha256:{digest}") + handler.end_headers() + handler.wfile.write(cached) + return True + + # 从上游获取 + url = f"{self.registry_url}/v2/{image}/blobs/sha256:{digest}" + + try: + req = urllib.request.Request(url) + + if self.username and self.password: + auth = base64.b64encode(f"{self.username}:{self.password}".encode()).decode() + req.add_header('Authorization', f"Basic {auth}") + + with urllib.request.urlopen(req) as response: + data = response.read() + + # 缓存 + if self.cache_enabled: + self._set_cache(cache_key, data) + + handler.send_response(200) + handler.send_header('Content-Type', 'application/octet-stream') + handler.send_header('Content-Length', str(len(data))) + handler.send_header('Docker-Content-Digest', f"sha256:{digest}") + handler.end_headers() + handler.wfile.write(data) + return True + + except Exception as e: + handler.send_error(500, f"Failed to fetch blob: {str(e)}") + return False + + def _handle_token(self, handler) -> bool: + """处理Token请求 - 生成真实的访问令牌""" + # 解析认证信息 + auth_header = handler.headers.get('Authorization', '') + username = None + password = None + + if auth_header.startswith('Basic '): + try: + decoded = base64.b64decode(auth_header[6:]).decode('utf-8') + username, password = decoded.split(':', 1) + except Exception: + pass + + # 验证凭据（如果有） + if self.username and self.password: + if username != self.username or password != self.password: + handler.send_error(401, "Invalid credentials") + return False + + # 生成唯一的访问令牌 + token_id = str(uuid.uuid4()) + issued_at = int(time.time()) + expires_in = 300 # 5分钟 + expires_at = issued_at + expires_in + + # 创建令牌信息（简化版 JWT 结构） + token_data = { + "iss": "hyc-mirror", + "sub": username or "anonymous", + "aud": self.registry_url, + "iat": issued_at, + "exp": expires_at, + "access": [ + {"type": "repository", "actions": ["pull"]}, + {"type": "registry", "actions": ["catalog"]} + ] + } + + # 使用 HMAC-SHA256 对令牌进行简单签名 + secret_key = f"hyc-mirror-{self.registry_url}".encode() + signature = hmac.new( + secret_key, + f"{token_id}:{issued_at}".encode(), + hashlib.sha256 + ).hexdigest()[:32] + + full_token = f"{token_id}-{signature}" + + handler.send_json_response({ + "token": full_token, + "expires_in": expires_in, + "issued_at": issued_at + }) + return True + + def _fetch_from_upstream(self, url: str) -> bytes: + """从上游获取数据""" + req = urllib.request.Request(url) + + if self.username and self.password: + auth = base64.b64encode(f"{self.username}:{self.password}".encode()).decode() + req.add_header('Authorization', f"Basic {auth}") + + with urllib.request.urlopen(req, timeout=30) as response: + return response.read() + + def _get_cache(self, handler, cache_key: str) -> Optional[bytes]: + """获取缓存""" + if not self.cache_enabled: + return None + + cache_path = self._get_cache_path(cache_key) + meta_path = cache_path + '.meta' + + if not os.path.exists(cache_path): + return None + + # 检查是否过期 + if os.path.exists(meta_path): + try: + with open(meta_path, 'r') as f: + meta = json.load(f) + if time.time() > meta.get('expires', 0): + return None + except Exception: + pass + + try: + with open(cache_path, 'rb') as f: + return f.read() + except Exception: + return None + + def _set_cache(self, cache_key: str, data: bytes): + """设置缓存""" + cache_path = self._get_cache_path(cache_key) + meta_path = cache_path + '.meta' + + os.makedirs(os.path.dirname(cache_path), exist_ok=True) + + try: + with open(cache_path, 'wb') as f: + f.write(data) + + meta = { + 'cached_at': time.time(), + 'expires': time.time() + self.cache_ttl, + 'size': len(data) + } + + with open(meta_path, 'w') as f: + json.dump(meta, f) + + except Exception as e: + print(f"Docker缓存写入失败: {e}") + + def _get_cache_path(self, cache_key: str) -> str: + """获取缓存路径""" + subdir = cache_key[:2] + return os.path.join(self.storage_dir, subdir, cache_key) + + def get_cache_stats(self) -> dict: + """获取缓存统计""" + if not os.path.exists(self.storage_dir): + return {'files': 0, 'size': 0} + + total_size = 0 + file_count = 0 + + for root, dirs, files in os.walk(self.storage_dir): + for f in files: + if not f.endswith('.meta'): + file_count += 1 + total_size += os.path.getsize(os.path.join(root, f)) + + return { + 'files': file_count, + 'size': total_size, + 'size_formatted': self._format_size(total_size) + } + + def _format_size(self, size_bytes: int) -> str: + """格式化文件大小""" + if size_bytes == 0: + return "0 B" + + units = ["B", "KB", "MB", "GB"] + i = 0 + while size_bytes >= 1024 and i < len(units) - 1: + size_bytes /= 1024.0 + i += 1 + + return f"{size_bytes:.2f} {units[i]}" diff --git a/mirrors/go.py b/mirrors/go.py index 4db2d12..75ea49e 100644 --- a/mirrors/go.py +++ b/mirrors/go.py @@ -1,525 +1,529 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- - -""" -Go模块代理处理器 -支持Go模块代理协议 -""" - -import os -import json -import time -import urllib.request -import urllib.parse -import urllib.error -from typing import Dict, List, Optional -from datetime import datetime - - -class GoProxy: - """Go模块代理""" - - def __init__(self, config: dict): - self.config = config - - # 配置 - 使用 storage_dir（基于 base_dir） - self.upstream_url = config.get('upstream_url', 'https://proxy.golang.org') - self.storage_dir = config.get('storage_dir', './downloads/go') - self.base_dir = config.get('base_dir', './downloads') - self.mode = config.get('mode', 'proxy') # proxy | direct - - # 确保存储目录存在 - os.makedirs(self.storage_dir, exist_ok=True) - - def handle_request(self, handler, path: str) -> bool: - """ - 处理Go模块请求 - 路径格式: /golang.org/x/net/@v/list - /golang.org/x/net/@v/v1.0.0.info - /golang.org/x/net/@v/v1.0.0.zip - /golang.org/x/net/@latest - """ - try: - parts = path.strip('/').split('/') - - if len(parts) < 2: - return self._handle_index(handler) - - # 解析模块路径和操作 - module_parts = [] - - for i, part in enumerate(parts): - if part.startswith('@'): - # 找到操作部分 - module_path = '/'.join(parts[:i]) - action = parts[i:] - break - else: - # 没有找到操作符 - module_path = '/'.join(parts) - action = [] - - if not action: - handler.send_error(400, "Invalid Go module path") - return False - - action_type = action[0] - - if action_type == '@v': - # 版本相关操作 - if len(action) >= 3: - version = action[2] - return self._handle_version(handler, module_path, version) - elif len(action) == 2: - # /@v/list - return self._handle_version_list(handler, module_path) - - elif action_type == '@latest': - # /@latest - return self._handle_latest(handler, module_path) - - elif action_type == '@all': - # /@all - return self._handle_all(handler, module_path) - - elif action_type == '@list': - # /@list - return self._handle_module_list(handler, module_path) - - else: - handler.send_error(400, f"Unknown action: {action_type}") - return False - - except Exception as e: - handler.send_error(500, str(e)) - return False - - def _handle_index(self, handler) -> bool: - """处理索引请求""" - handler.send_json_response({ - 'proxy_url': self.upstream_url, - 'mode': self.mode, - 'cache_stats': self.get_cache_stats() - }) - return True - - def _handle_version_list(self, handler, module: str) -> bool: - """处理版本列表请求 /@v/list""" - cache_key = f"vlist:{module}" - - cached = self._get_cache(cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'text/plain; charset=utf-8') - handler.end_headers() - handler.wfile.write(cached) - return True - - # 从上游获取 - url = f"{self.upstream_url}/{module}/@v/list" - - try: - data = self._fetch(url) - - if self.cache_enabled: - self._set_cache(cache_key, data) - - handler.send_response(200) - handler.send_header('Content-Type', 'text/plain; charset=utf-8') - handler.end_headers() - handler.wfile.write(data) - return True - - except urllib.error.HTTPError as e: - if e.code == 404: - handler.send_error(404, f"Module not found: {module}") - else: - handler.send_error(502, f"Failed to fetch: {str(e)}") - return False - - def _handle_version_info(self, handler, module: str, version: str) -> bool: - """处理版本信息请求 /@v/version.info""" - cache_key = f"info:{module}:{version}" - - cached = self._get_cache(cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'application/json') - handler.end_headers() - handler.wfile.write(cached) - return True - - url = f"{self.upstream_url}/{module}/@v/{version}.info" - - try: - data = self._fetch(url) - - if self.cache_enabled: - self._set_cache(cache_key, data) - - handler.send_response(200) - handler.send_header('Content-Type', 'application/json') - handler.end_headers() - handler.wfile.write(data) - return True - - except urllib.error.HTTPError as e: - handler.send_error(502, f"Failed to fetch info: {str(e)}") - return False - - def _handle_version(self, handler, module: str, suffix: str) -> bool: - """处理版本相关请求""" - if suffix.endswith('.info'): - version = suffix[:-5] - return self._handle_version_info(handler, module, version) - - elif suffix.endswith('.zip'): - version = suffix[:-4] - return self._handle_zip(handler, module, version) - - elif suffix.endswith('.mod'): - version = suffix[:-4] - return self._handle_mod(handler, module, version) - - elif suffix.endswith('.sum'): - version = suffix[:-4] - return self._handle_sum(handler, module, version) - - else: - handler.send_error(400, f"Unknown suffix: {suffix}") - return False - - def _handle_latest(self, handler, module: str) -> bool: - """处理最新版本请求 /@latest""" - cache_key = f"latest:{module}" - - cached = self._get_cache(cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'application/json') - handler.end_headers() - handler.wfile.write(cached) - return True - - url = f"{self.upstream_url}/{module}/@latest" - - try: - data = self._fetch(url) - - if self.cache_enabled: - self._set_cache(cache_key, data) - - handler.send_response(200) - handler.send_header('Content-Type', 'application/json') - handler.end_headers() - handler.wfile.write(data) - return True - - except urllib.error.HTTPError as e: - handler.send_error(502, f"Failed to fetch latest: {str(e)}") - return False - - def _handle_zip(self, handler, module: str, version: str) -> bool: - """处理zip下载""" - cache_key = f"zip:{module}:{version}" - - cached = self._get_cache(cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'application/zip') - handler.send_header('Content-Length', str(len(cached))) - handler.end_headers() - handler.wfile.write(cached) - return True - - url = f"{self.upstream_url}/{module}/@v/{version}.zip" - - try: - data = self._fetch(url) - - if self.cache_enabled: - self._set_cache(cache_key, data) - - handler.send_response(200) - handler.send_header('Content-Type', 'application/zip') - handler.send_header('Content-Length', str(len(data))) - handler.end_headers() - handler.wfile.write(data) - return True - - except urllib.error.HTTPError as e: - handler.send_error(502, f"Failed to fetch zip: {str(e)}") - return False - - def _handle_mod(self, handler, module: str, version: str) -> bool: - """处理mod文件""" - cache_key = f"mod:{module}:{version}" - - cached = self._get_cache(cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'text/plain; charset=utf-8') - handler.send_header('Content-Length', str(len(cached))) - handler.end_headers() - handler.wfile.write(cached) - return True - - url = f"{self.upstream_url}/{module}/@v/{version}.mod" - - try: - data = self._fetch(url) - - if self.cache_enabled: - self._set_cache(cache_key, data) - - handler.send_response(200) - handler.send_header('Content-Type', 'text/plain; charset=utf-8') - handler.send_header('Content-Length', str(len(data))) - handler.end_headers() - handler.wfile.write(data) - return True - - except urllib.error.HTTPError as e: - handler.send_error(502, f"Failed to fetch mod: {str(e)}") - return False - - def _handle_sum(self, handler, module: str, version: str) -> bool: - """处理sum文件""" - cache_key = f"sum:{module}:{version}" - - cached = self._get_cache(cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'text/plain; charset=utf-8') - handler.end_headers() - handler.wfile.write(cached) - return True - - url = f"{self.upstream_url}/{module}/@v/{version}.sum" - - try: - data = self._fetch(url) - - if self.cache_enabled: - self._set_cache(cache_key, data) - - handler.send_response(200) - handler.send_header('Content-Type', 'text/plain; charset=utf-8') - handler.end_headers() - handler.wfile.write(data) - return True - - except urllib.error.HTTPError as e: - if e.code == 404: - # 没有sum文件时返回空 - handler.send_response(200) - handler.send_header('Content-Type', 'text/plain; charset=utf-8') - handler.end_headers() - handler.wfile.write(b'') - else: - handler.send_error(502, f"Failed to fetch sum: {str(e)}") - return False - - def _handle_all(self, handler, module: str) -> bool: - """处理/@all请求 - 返回模块及其所有依赖的zip包""" - cache_key = f"all:{module}" - - cached = self._get_cache(cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'application/zip') - handler.send_header('Content-Length', str(len(cached))) - handler.end_headers() - handler.wfile.write(cached) - return True - - # 从上游获取所有依赖的zip - url = f"{self.upstream_url}/{module}/@all.zip" - - try: - data = self._fetch(url) - - if not data: - handler.send_error(404, f"Module not found: {module}") - return False - - if self.cache_enabled: - self._set_cache(cache_key, data) - - handler.send_response(200) - handler.send_header('Content-Type', 'application/zip') - handler.send_header('Content-Length', str(len(data))) - handler.end_headers() - handler.wfile.write(data) - return True - - except urllib.error.HTTPError as e: - if e.code == 404: - handler.send_error(404, f"Module not found: {module}") - else: - handler.send_error(502, f"Failed to fetch @all: {str(e)}") - return False - - def _handle_module_list(self, handler, module: str) -> bool: - """处理/@list请求 - 返回模块及其依赖的路径列表""" - cache_key = f"list:{module}" - - cached = self._get_cache(cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'text/plain; charset=utf-8') - handler.end_headers() - handler.wfile.write(cached) - return True - - # 首先获取模块的 go.mod 文件以提取依赖 - mod_url = f"{self.upstream_url}/{module}/@v/{module}.mod" - - try: - mod_data = self._fetch(mod_url) - if not mod_data: - handler.send_error(404, f"Module not found: {module}") - return False - - # 解析go.mod获取依赖 - modules = [module] - mod_content = mod_data.decode('utf-8', errors='replace') - - # 提取require语句中的依赖 - import re - require_pattern = r'require\s+\(([^\)]+)\)' - inline_require_pattern = r'require\s+([^\s]+)\s+([^\s]+)' - - # 处理多行require - matches = re.findall(require_pattern, mod_content, re.DOTALL) - for match in matches: - for line in match.strip().split('\n'): - line = line.strip() - if line and not line.startswith('//'): - parts = line.split() - if parts: - modules.append(parts[0]) - - # 处理单行require - matches = re.findall(inline_require_pattern, mod_content) - for match in matches: - if match[0] not in modules: - modules.append(match[0]) - - # 生成列表输出 - list_output = '\n'.join(sorted(set(modules))) + '\n' - - if self.cache_enabled: - self._set_cache(cache_key, list_output.encode()) - - handler.send_response(200) - handler.send_header('Content-Type', 'text/plain; charset=utf-8') - handler.end_headers() - handler.wfile.write(list_output.encode()) - return True - - except urllib.error.HTTPError as e: - if e.code == 404: - handler.send_error(404, f"Module not found: {module}") - else: - handler.send_error(502, f"Failed to fetch @list: {str(e)}") - return False - - def _fetch(self, url: str) -> Optional[bytes]: - """从URL获取数据""" - try: - req = urllib.request.Request(url) - req.add_header('User-Agent', 'Go-Mirror/1.0') - - with urllib.request.urlopen(req, timeout=60) as response: - return response.read() - - except Exception: - return None - - def _get_cache(self, cache_key: str) -> Optional[bytes]: - """获取缓存""" - if not self.cache_enabled: - return None - - cache_path = self._get_cache_path(cache_key) - meta_path = cache_path + '.meta' - - if not os.path.exists(cache_path): - return None - - if os.path.exists(meta_path): - try: - with open(meta_path, 'r') as f: - meta = json.load(f) - if time.time() > meta.get('expires', 0): - return None - except Exception: - pass - - try: - with open(cache_path, 'rb') as f: - return f.read() - except Exception: - return None - - def _set_cache(self, cache_key: str, data: bytes): - """设置缓存""" - cache_path = self._get_cache_path(cache_key) - meta_path = cache_path + '.meta' - - os.makedirs(os.path.dirname(cache_path), exist_ok=True) - - try: - with open(cache_path, 'wb') as f: - f.write(data) - - meta = { - 'cached_at': time.time(), - 'expires': time.time() + self.cache_ttl, - 'size': len(data) - } - - with open(meta_path, 'w') as f: - json.dump(meta, f) - - except Exception as e: - print(f"Go缓存写入失败: {e}") - - def _get_cache_path(self, cache_key: str) -> str: - """获取缓存路径""" - subdir = cache_key[:2] - return os.path.join(self.storage_dir, subdir, cache_key) - - def get_cache_stats(self) -> dict: - """获取缓存统计""" - if not os.path.exists(self.storage_dir): - return {'files': 0, 'size': 0} - - total_size = 0 - file_count = 0 - - for root, dirs, files in os.walk(self.storage_dir): - for f in files: - if not f.endswith('.meta'): - file_count += 1 - total_size += os.path.getsize(os.path.join(root, f)) - - return { - 'files': file_count, - 'size': total_size, - 'size_formatted': self._format_size(total_size) - } - - def _format_size(self, size_bytes: int) -> str: - """格式化文件大小""" - if size_bytes == 0: - return "0 B" - - units = ["B", "KB", "MB", "GB"] - i = 0 - while size_bytes >= 1024 and i < len(units) - 1: - size_bytes /= 1024.0 - i += 1 - - return f"{size_bytes:.2f} {units[i]}" +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +""" +Go模块代理处理器 +支持Go模块代理协议 +""" + +import os +import json +import time +import urllib.request +import urllib.parse +import urllib.error +from typing import Dict, List, Optional +from datetime import datetime + + +class GoProxy: + """Go模块代理""" + + def __init__(self, config: dict): + self.config = config + + # 配置 - 使用 storage_dir（基于 base_dir） + self.upstream_url = config.get('upstream_url', 'https://proxy.golang.org') + self.storage_dir = config.get('storage_dir', './downloads/go') + self.base_dir = config.get('base_dir', './downloads') + self.mode = config.get('mode', 'proxy') # proxy | direct + + # 缓存配置（缺失时默认启用，避免 AttributeError） + self.cache_enabled = config.get('cache_enabled', True) + self.cache_ttl = config.get('cache_ttl', 3600) + + # 确保存储目录存在 + os.makedirs(self.storage_dir, exist_ok=True) + + def handle_request(self, handler, path: str) -> bool: + """ + 处理Go模块请求 + 路径格式: /golang.org/x/net/@v/list + /golang.org/x/net/@v/v1.0.0.info + /golang.org/x/net/@v/v1.0.0.zip + /golang.org/x/net/@latest + """ + try: + parts = path.strip('/').split('/') + + if len(parts) < 2: + return self._handle_index(handler) + + # 解析模块路径和操作 + module_parts = [] + + for i, part in enumerate(parts): + if part.startswith('@'): + # 找到操作部分 + module_path = '/'.join(parts[:i]) + action = parts[i:] + break + else: + # 没有找到操作符 + module_path = '/'.join(parts) + action = [] + + if not action: + handler.send_error(400, "Invalid Go module path") + return False + + action_type = action[0] + + if action_type == '@v': + # 版本相关操作 + if len(action) >= 3: + version = action[2] + return self._handle_version(handler, module_path, version) + elif len(action) == 2: + # /@v/list + return self._handle_version_list(handler, module_path) + + elif action_type == '@latest': + # /@latest + return self._handle_latest(handler, module_path) + + elif action_type == '@all': + # /@all + return self._handle_all(handler, module_path) + + elif action_type == '@list': + # /@list + return self._handle_module_list(handler, module_path) + + else: + handler.send_error(400, f"Unknown action: {action_type}") + return False + + except Exception as e: + handler.send_error(500, str(e)) + return False + + def _handle_index(self, handler) -> bool: + """处理索引请求""" + handler.send_json_response({ + 'proxy_url': self.upstream_url, + 'mode': self.mode, + 'cache_stats': self.get_cache_stats() + }) + return True + + def _handle_version_list(self, handler, module: str) -> bool: + """处理版本列表请求 /@v/list""" + cache_key = f"vlist:{module}" + + cached = self._get_cache(cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'text/plain; charset=utf-8') + handler.end_headers() + handler.wfile.write(cached) + return True + + # 从上游获取 + url = f"{self.upstream_url}/{module}/@v/list" + + try: + data = self._fetch(url) + + if self.cache_enabled: + self._set_cache(cache_key, data) + + handler.send_response(200) + handler.send_header('Content-Type', 'text/plain; charset=utf-8') + handler.end_headers() + handler.wfile.write(data) + return True + + except urllib.error.HTTPError as e: + if e.code == 404: + handler.send_error(404, f"Module not found: {module}") + else: + handler.send_error(502, f"Failed to fetch: {str(e)}") + return False + + def _handle_version_info(self, handler, module: str, version: str) -> bool: + """处理版本信息请求 /@v/version.info""" + cache_key = f"info:{module}:{version}" + + cached = self._get_cache(cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'application/json') + handler.end_headers() + handler.wfile.write(cached) + return True + + url = f"{self.upstream_url}/{module}/@v/{version}.info" + + try: + data = self._fetch(url) + + if self.cache_enabled: + self._set_cache(cache_key, data) + + handler.send_response(200) + handler.send_header('Content-Type', 'application/json') + handler.end_headers() + handler.wfile.write(data) + return True + + except urllib.error.HTTPError as e: + handler.send_error(502, f"Failed to fetch info: {str(e)}") + return False + + def _handle_version(self, handler, module: str, suffix: str) -> bool: + """处理版本相关请求""" + if suffix.endswith('.info'): + version = suffix[:-5] + return self._handle_version_info(handler, module, version) + + elif suffix.endswith('.zip'): + version = suffix[:-4] + return self._handle_zip(handler, module, version) + + elif suffix.endswith('.mod'): + version = suffix[:-4] + return self._handle_mod(handler, module, version) + + elif suffix.endswith('.sum'): + version = suffix[:-4] + return self._handle_sum(handler, module, version) + + else: + handler.send_error(400, f"Unknown suffix: {suffix}") + return False + + def _handle_latest(self, handler, module: str) -> bool: + """处理最新版本请求 /@latest""" + cache_key = f"latest:{module}" + + cached = self._get_cache(cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'application/json') + handler.end_headers() + handler.wfile.write(cached) + return True + + url = f"{self.upstream_url}/{module}/@latest" + + try: + data = self._fetch(url) + + if self.cache_enabled: + self._set_cache(cache_key, data) + + handler.send_response(200) + handler.send_header('Content-Type', 'application/json') + handler.end_headers() + handler.wfile.write(data) + return True + + except urllib.error.HTTPError as e: + handler.send_error(502, f"Failed to fetch latest: {str(e)}") + return False + + def _handle_zip(self, handler, module: str, version: str) -> bool: + """处理zip下载""" + cache_key = f"zip:{module}:{version}" + + cached = self._get_cache(cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'application/zip') + handler.send_header('Content-Length', str(len(cached))) + handler.end_headers() + handler.wfile.write(cached) + return True + + url = f"{self.upstream_url}/{module}/@v/{version}.zip" + + try: + data = self._fetch(url) + + if self.cache_enabled: + self._set_cache(cache_key, data) + + handler.send_response(200) + handler.send_header('Content-Type', 'application/zip') + handler.send_header('Content-Length', str(len(data))) + handler.end_headers() + handler.wfile.write(data) + return True + + except urllib.error.HTTPError as e: + handler.send_error(502, f"Failed to fetch zip: {str(e)}") + return False + + def _handle_mod(self, handler, module: str, version: str) -> bool: + """处理mod文件""" + cache_key = f"mod:{module}:{version}" + + cached = self._get_cache(cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'text/plain; charset=utf-8') + handler.send_header('Content-Length', str(len(cached))) + handler.end_headers() + handler.wfile.write(cached) + return True + + url = f"{self.upstream_url}/{module}/@v/{version}.mod" + + try: + data = self._fetch(url) + + if self.cache_enabled: + self._set_cache(cache_key, data) + + handler.send_response(200) + handler.send_header('Content-Type', 'text/plain; charset=utf-8') + handler.send_header('Content-Length', str(len(data))) + handler.end_headers() + handler.wfile.write(data) + return True + + except urllib.error.HTTPError as e: + handler.send_error(502, f"Failed to fetch mod: {str(e)}") + return False + + def _handle_sum(self, handler, module: str, version: str) -> bool: + """处理sum文件""" + cache_key = f"sum:{module}:{version}" + + cached = self._get_cache(cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'text/plain; charset=utf-8') + handler.end_headers() + handler.wfile.write(cached) + return True + + url = f"{self.upstream_url}/{module}/@v/{version}.sum" + + try: + data = self._fetch(url) + + if self.cache_enabled: + self._set_cache(cache_key, data) + + handler.send_response(200) + handler.send_header('Content-Type', 'text/plain; charset=utf-8') + handler.end_headers() + handler.wfile.write(data) + return True + + except urllib.error.HTTPError as e: + if e.code == 404: + # 没有sum文件时返回空 + handler.send_response(200) + handler.send_header('Content-Type', 'text/plain; charset=utf-8') + handler.end_headers() + handler.wfile.write(b'') + else: + handler.send_error(502, f"Failed to fetch sum: {str(e)}") + return False + + def _handle_all(self, handler, module: str) -> bool: + """处理/@all请求 - 返回模块及其所有依赖的zip包""" + cache_key = f"all:{module}" + + cached = self._get_cache(cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'application/zip') + handler.send_header('Content-Length', str(len(cached))) + handler.end_headers() + handler.wfile.write(cached) + return True + + # 从上游获取所有依赖的zip + url = f"{self.upstream_url}/{module}/@all.zip" + + try: + data = self._fetch(url) + + if not data: + handler.send_error(404, f"Module not found: {module}") + return False + + if self.cache_enabled: + self._set_cache(cache_key, data) + + handler.send_response(200) + handler.send_header('Content-Type', 'application/zip') + handler.send_header('Content-Length', str(len(data))) + handler.end_headers() + handler.wfile.write(data) + return True + + except urllib.error.HTTPError as e: + if e.code == 404: + handler.send_error(404, f"Module not found: {module}") + else: + handler.send_error(502, f"Failed to fetch @all: {str(e)}") + return False + + def _handle_module_list(self, handler, module: str) -> bool: + """处理/@list请求 - 返回模块及其依赖的路径列表""" + cache_key = f"list:{module}" + + cached = self._get_cache(cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'text/plain; charset=utf-8') + handler.end_headers() + handler.wfile.write(cached) + return True + + # 首先获取模块的 go.mod 文件以提取依赖 + mod_url = f"{self.upstream_url}/{module}/@v/{module}.mod" + + try: + mod_data = self._fetch(mod_url) + if not mod_data: + handler.send_error(404, f"Module not found: {module}") + return False + + # 解析go.mod获取依赖 + modules = [module] + mod_content = mod_data.decode('utf-8', errors='replace') + + # 提取require语句中的依赖 + import re + require_pattern = r'require\s+\(([^\)]+)\)' + inline_require_pattern = r'require\s+([^\s]+)\s+([^\s]+)' + + # 处理多行require + matches = re.findall(require_pattern, mod_content, re.DOTALL) + for match in matches: + for line in match.strip().split('\n'): + line = line.strip() + if line and not line.startswith('//'): + parts = line.split() + if parts: + modules.append(parts[0]) + + # 处理单行require + matches = re.findall(inline_require_pattern, mod_content) + for match in matches: + if match[0] not in modules: + modules.append(match[0]) + + # 生成列表输出 + list_output = '\n'.join(sorted(set(modules))) + '\n' + + if self.cache_enabled: + self._set_cache(cache_key, list_output.encode()) + + handler.send_response(200) + handler.send_header('Content-Type', 'text/plain; charset=utf-8') + handler.end_headers() + handler.wfile.write(list_output.encode()) + return True + + except urllib.error.HTTPError as e: + if e.code == 404: + handler.send_error(404, f"Module not found: {module}") + else: + handler.send_error(502, f"Failed to fetch @list: {str(e)}") + return False + + def _fetch(self, url: str) -> Optional[bytes]: + """从URL获取数据""" + try: + req = urllib.request.Request(url) + req.add_header('User-Agent', 'Go-Mirror/1.0') + + with urllib.request.urlopen(req, timeout=60) as response: + return response.read() + + except Exception: + return None + + def _get_cache(self, cache_key: str) -> Optional[bytes]: + """获取缓存""" + if not self.cache_enabled: + return None + + cache_path = self._get_cache_path(cache_key) + meta_path = cache_path + '.meta' + + if not os.path.exists(cache_path): + return None + + if os.path.exists(meta_path): + try: + with open(meta_path, 'r') as f: + meta = json.load(f) + if time.time() > meta.get('expires', 0): + return None + except Exception: + pass + + try: + with open(cache_path, 'rb') as f: + return f.read() + except Exception: + return None + + def _set_cache(self, cache_key: str, data: bytes): + """设置缓存""" + cache_path = self._get_cache_path(cache_key) + meta_path = cache_path + '.meta' + + os.makedirs(os.path.dirname(cache_path), exist_ok=True) + + try: + with open(cache_path, 'wb') as f: + f.write(data) + + meta = { + 'cached_at': time.time(), + 'expires': time.time() + self.cache_ttl, + 'size': len(data) + } + + with open(meta_path, 'w') as f: + json.dump(meta, f) + + except Exception as e: + print(f"Go缓存写入失败: {e}") + + def _get_cache_path(self, cache_key: str) -> str: + """获取缓存路径""" + subdir = cache_key[:2] + return os.path.join(self.storage_dir, subdir, cache_key) + + def get_cache_stats(self) -> dict: + """获取缓存统计""" + if not os.path.exists(self.storage_dir): + return {'files': 0, 'size': 0} + + total_size = 0 + file_count = 0 + + for root, dirs, files in os.walk(self.storage_dir): + for f in files: + if not f.endswith('.meta'): + file_count += 1 + total_size += os.path.getsize(os.path.join(root, f)) + + return { + 'files': file_count, + 'size': total_size, + 'size_formatted': self._format_size(total_size) + } + + def _format_size(self, size_bytes: int) -> str: + """格式化文件大小""" + if size_bytes == 0: + return "0 B" + + units = ["B", "KB", "MB", "GB"] + i = 0 + while size_bytes >= 1024 and i < len(units) - 1: + size_bytes /= 1024.0 + i += 1 + + return f"{size_bytes:.2f} {units[i]}" diff --git a/mirrors/npm.py b/mirrors/npm.py index 59503d2..7e5f1d1 100644 --- a/mirrors/npm.py +++ b/mirrors/npm.py @@ -1,280 +1,284 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- - -""" -npm镜像代理处理器 -支持Node.js包管理器 -""" - -import os -import json -import time -import urllib.request -import urllib.parse -from typing import Dict, List, Optional -from datetime import datetime - - -class NpmMirror: - """npm镜像代理""" - - def __init__(self, config: dict): - self.config = config - - # 配置 - 使用 storage_dir（基于 base_dir） - self.upstream_url = config.get('upstream_url', 'https://registry.npmjs.org') - self.storage_dir = config.get('storage_dir', './downloads/npm') - self.base_dir = config.get('base_dir', './downloads') - - # 确保存储目录存在 - os.makedirs(self.storage_dir, exist_ok=True) - - def handle_request(self, handler, path: str) -> bool: - """ - 处理npm请求 - 路径格式: /lodash 或 /-/package/lodash/dist - """ - try: - parts = path.strip('/').split('/') - - if not parts: - return self._handle_index(handler) - - if parts[0] == '-': - # Scoped package 或其他特殊请求 - if len(parts) >= 4 and parts[1] == 'package': - return self._handleScopedPackage(handler, parts[2], parts[3] if len(parts) > 3 else None) - elif len(parts) >= 3 and parts[1] == 'package': - return self._handle_package(handler, parts[2], None) - else: - handler.send_error(400, "Invalid npm API path") - return False - - elif parts[0] == '@': - # Scoped package - if len(parts) >= 2: - scope = parts[0] - package = '/'.join(parts[1:]) - return self._handle_scoped_package(handler, scope, package) - else: - handler.send_error(400, "Invalid scoped package") - return False - - elif parts[0] == '-/': - # npm特殊路径 - return self._handle_special(handler, '/'.join(parts)) - - elif len(parts) == 1: - # 单个包名 - return self._handle_package(handler, parts[0], None) - - else: - # 其他请求 - return self._handle_package(handler, parts[0], parts[1] if len(parts) > 1 else None) - - except Exception as e: - handler.send_error(500, str(e)) - return False - - def _handle_index(self, handler) -> bool: - """处理索引请求""" - handler.send_json_response({ - 'registry_url': self.upstream_url, - 'cache_stats': self.get_cache_stats() - }) - return True - - def _handle_package(self, handler, package: str, version: str = None) -> bool: - """处理包元数据请求""" - cache_key = f"package:{package}:{version or 'latest'}" - - cached = self._get_cache(cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'application/json') - handler.end_headers() - handler.wfile.write(cached) - return True - - # 从上游获取 - if version: - url = f"{self.upstream_url}/{package}/{version}" - else: - url = f"{self.upstream_url}/{package}/latest" - - try: - data = self._fetch(url) - - if self.cache_enabled: - self._set_cache(cache_key, data) - - handler.send_response(200) - handler.send_header('Content-Type', 'application/json') - handler.end_headers() - handler.wfile.write(data) - return True - - except urllib.error.HTTPError as e: - handler.send_error(404, f"Package not found: {package}") - return False - - def _handle_scoped_package(self, handler, scope: str, package: str) -> bool: - """处理scoped包""" - full_name = f"{scope}/{package}" - return self._handle_package(handler, full_name, None) - - def _handleScopedPackage(self, handler, scope: str, package: str) -> bool: - """处理特殊路径的scoped包""" - full_name = f"{scope}/{package}" - return self._handle_package(handler, full_name, None) - - def _handle_special(self, handler, path: str) -> bool: - """处理特殊npm路径""" - # 简化实现：转发到上游 - url = f"{self.upstream_url}/{path}" - - try: - data = self._fetch(url) - - handler.send_response(200) - handler.send_header('Content-Type', 'application/json') - handler.end_headers() - handler.wfile.write(data) - return True - - except Exception as e: - handler.send_error(502, f"Failed to fetch: {str(e)}") - return False - - def _handle_tarball(self, handler, package: str, filename: str) -> bool: - """处理tarball下载""" - cache_key = f"tarball:{package}:{filename}" - - cached = self._get_cache(cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'application/octet-stream') - handler.send_header('Content-Length', str(len(cached))) - handler.end_headers() - handler.wfile.write(cached) - return True - - # 从上游获取 - url = f"{self.upstream_url}/{package}/-/{filename}" - - try: - data = self._fetch(url) - - if self.cache_enabled: - self._set_cache(cache_key, data) - - handler.send_response(200) - handler.send_header('Content-Type', 'application/octet-stream') - handler.send_header('Content-Length', str(len(data))) - handler.end_headers() - handler.wfile.write(data) - return True - - except Exception as e: - handler.send_error(502, f"Failed to fetch tarball: {str(e)}") - return False - - def _fetch(self, url: str) -> Optional[bytes]: - """从URL获取数据""" - try: - req = urllib.request.Request(url) - req.add_header('User-Agent', 'npm-Mirror/1.0') - req.add_header('Accept', 'application/json') - - with urllib.request.urlopen(req, timeout=30) as response: - return response.read() - - except Exception: - return None - - def _get_cache(self, cache_key: str) -> Optional[bytes]: - """获取缓存""" - if not self.cache_enabled: - return None - - cache_path = self._get_cache_path(cache_key) - meta_path = cache_path + '.meta' - - if not os.path.exists(cache_path): - return None - - if os.path.exists(meta_path): - try: - with open(meta_path, 'r') as f: - meta = json.load(f) - if time.time() > meta.get('expires', 0): - return None - except Exception: - pass - - try: - with open(cache_path, 'rb') as f: - return f.read() - except Exception: - return None - - def _set_cache(self, cache_key: str, data: bytes): - """设置缓存""" - cache_path = self._get_cache_path(cache_key) - meta_path = cache_path + '.meta' - - os.makedirs(os.path.dirname(cache_path), exist_ok=True) - - try: - with open(cache_path, 'wb') as f: - f.write(data) - - meta = { - 'cached_at': time.time(), - 'expires': time.time() + self.cache_ttl, - 'size': len(data) - } - - with open(meta_path, 'w') as f: - json.dump(meta, f) - - except Exception as e: - print(f"npm缓存写入失败: {e}") - - def _get_cache_path(self, cache_key: str) -> str: - """获取缓存路径""" - subdir = cache_key[:2] - return os.path.join(self.storage_dir, subdir, cache_key) - - def get_cache_stats(self) -> dict: - """获取缓存统计""" - if not os.path.exists(self.storage_dir): - return {'files': 0, 'size': 0} - - total_size = 0 - file_count = 0 - - for root, dirs, files in os.walk(self.storage_dir): - for f in files: - if not f.endswith('.meta'): - file_count += 1 - total_size += os.path.getsize(os.path.join(root, f)) - - return { - 'files': file_count, - 'size': total_size, - 'size_formatted': self._format_size(total_size) - } - - def _format_size(self, size_bytes: int) -> str: - """格式化文件大小""" - if size_bytes == 0: - return "0 B" - - units = ["B", "KB", "MB", "GB"] - i = 0 - while size_bytes >= 1024 and i < len(units) - 1: - size_bytes /= 1024.0 - i += 1 - - return f"{size_bytes:.2f} {units[i]}" +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +""" +npm镜像代理处理器 +支持Node.js包管理器 +""" + +import os +import json +import time +import urllib.request +import urllib.parse +from typing import Dict, List, Optional +from datetime import datetime + + +class NpmMirror: + """npm镜像代理""" + + def __init__(self, config: dict): + self.config = config + + # 配置 - 使用 storage_dir（基于 base_dir） + self.upstream_url = config.get('upstream_url', 'https://registry.npmjs.org') + self.storage_dir = config.get('storage_dir', './downloads/npm') + self.base_dir = config.get('base_dir', './downloads') + + # 缓存配置（缺失时默认启用，避免 AttributeError） + self.cache_enabled = config.get('cache_enabled', True) + self.cache_ttl = config.get('cache_ttl', 3600) + + # 确保存储目录存在 + os.makedirs(self.storage_dir, exist_ok=True) + + def handle_request(self, handler, path: str) -> bool: + """ + 处理npm请求 + 路径格式: /lodash 或 /-/package/lodash/dist + """ + try: + parts = path.strip('/').split('/') + + if not parts: + return self._handle_index(handler) + + if parts[0] == '-': + # Scoped package 或其他特殊请求 + if len(parts) >= 4 and parts[1] == 'package': + return self._handleScopedPackage(handler, parts[2], parts[3] if len(parts) > 3 else None) + elif len(parts) >= 3 and parts[1] == 'package': + return self._handle_package(handler, parts[2], None) + else: + handler.send_error(400, "Invalid npm API path") + return False + + elif parts[0] == '@': + # Scoped package + if len(parts) >= 2: + scope = parts[0] + package = '/'.join(parts[1:]) + return self._handle_scoped_package(handler, scope, package) + else: + handler.send_error(400, "Invalid scoped package") + return False + + elif parts[0] == '-/': + # npm特殊路径 + return self._handle_special(handler, '/'.join(parts)) + + elif len(parts) == 1: + # 单个包名 + return self._handle_package(handler, parts[0], None) + + else: + # 其他请求 + return self._handle_package(handler, parts[0], parts[1] if len(parts) > 1 else None) + + except Exception as e: + handler.send_error(500, str(e)) + return False + + def _handle_index(self, handler) -> bool: + """处理索引请求""" + handler.send_json_response({ + 'registry_url': self.upstream_url, + 'cache_stats': self.get_cache_stats() + }) + return True + + def _handle_package(self, handler, package: str, version: str = None) -> bool: + """处理包元数据请求""" + cache_key = f"package:{package}:{version or 'latest'}" + + cached = self._get_cache(cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'application/json') + handler.end_headers() + handler.wfile.write(cached) + return True + + # 从上游获取 + if version: + url = f"{self.upstream_url}/{package}/{version}" + else: + url = f"{self.upstream_url}/{package}/latest" + + try: + data = self._fetch(url) + + if self.cache_enabled: + self._set_cache(cache_key, data) + + handler.send_response(200) + handler.send_header('Content-Type', 'application/json') + handler.end_headers() + handler.wfile.write(data) + return True + + except urllib.error.HTTPError as e: + handler.send_error(404, f"Package not found: {package}") + return False + + def _handle_scoped_package(self, handler, scope: str, package: str) -> bool: + """处理scoped包""" + full_name = f"{scope}/{package}" + return self._handle_package(handler, full_name, None) + + def _handleScopedPackage(self, handler, scope: str, package: str) -> bool: + """处理特殊路径的scoped包""" + full_name = f"{scope}/{package}" + return self._handle_package(handler, full_name, None) + + def _handle_special(self, handler, path: str) -> bool: + """处理特殊npm路径""" + # 简化实现：转发到上游 + url = f"{self.upstream_url}/{path}" + + try: + data = self._fetch(url) + + handler.send_response(200) + handler.send_header('Content-Type', 'application/json') + handler.end_headers() + handler.wfile.write(data) + return True + + except Exception as e: + handler.send_error(502, f"Failed to fetch: {str(e)}") + return False + + def _handle_tarball(self, handler, package: str, filename: str) -> bool: + """处理tarball下载""" + cache_key = f"tarball:{package}:{filename}" + + cached = self._get_cache(cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'application/octet-stream') + handler.send_header('Content-Length', str(len(cached))) + handler.end_headers() + handler.wfile.write(cached) + return True + + # 从上游获取 + url = f"{self.upstream_url}/{package}/-/{filename}" + + try: + data = self._fetch(url) + + if self.cache_enabled: + self._set_cache(cache_key, data) + + handler.send_response(200) + handler.send_header('Content-Type', 'application/octet-stream') + handler.send_header('Content-Length', str(len(data))) + handler.end_headers() + handler.wfile.write(data) + return True + + except Exception as e: + handler.send_error(502, f"Failed to fetch tarball: {str(e)}") + return False + + def _fetch(self, url: str) -> Optional[bytes]: + """从URL获取数据""" + try: + req = urllib.request.Request(url) + req.add_header('User-Agent', 'npm-Mirror/1.0') + req.add_header('Accept', 'application/json') + + with urllib.request.urlopen(req, timeout=30) as response: + return response.read() + + except Exception: + return None + + def _get_cache(self, cache_key: str) -> Optional[bytes]: + """获取缓存""" + if not self.cache_enabled: + return None + + cache_path = self._get_cache_path(cache_key) + meta_path = cache_path + '.meta' + + if not os.path.exists(cache_path): + return None + + if os.path.exists(meta_path): + try: + with open(meta_path, 'r') as f: + meta = json.load(f) + if time.time() > meta.get('expires', 0): + return None + except Exception: + pass + + try: + with open(cache_path, 'rb') as f: + return f.read() + except Exception: + return None + + def _set_cache(self, cache_key: str, data: bytes): + """设置缓存""" + cache_path = self._get_cache_path(cache_key) + meta_path = cache_path + '.meta' + + os.makedirs(os.path.dirname(cache_path), exist_ok=True) + + try: + with open(cache_path, 'wb') as f: + f.write(data) + + meta = { + 'cached_at': time.time(), + 'expires': time.time() + self.cache_ttl, + 'size': len(data) + } + + with open(meta_path, 'w') as f: + json.dump(meta, f) + + except Exception as e: + print(f"npm缓存写入失败: {e}") + + def _get_cache_path(self, cache_key: str) -> str: + """获取缓存路径""" + subdir = cache_key[:2] + return os.path.join(self.storage_dir, subdir, cache_key) + + def get_cache_stats(self) -> dict: + """获取缓存统计""" + if not os.path.exists(self.storage_dir): + return {'files': 0, 'size': 0} + + total_size = 0 + file_count = 0 + + for root, dirs, files in os.walk(self.storage_dir): + for f in files: + if not f.endswith('.meta'): + file_count += 1 + total_size += os.path.getsize(os.path.join(root, f)) + + return { + 'files': file_count, + 'size': total_size, + 'size_formatted': self._format_size(total_size) + } + + def _format_size(self, size_bytes: int) -> str: + """格式化文件大小""" + if size_bytes == 0: + return "0 B" + + units = ["B", "KB", "MB", "GB"] + i = 0 + while size_bytes >= 1024 and i < len(units) - 1: + size_bytes /= 1024.0 + i += 1 + + return f"{size_bytes:.2f} {units[i]}" diff --git a/mirrors/yum.py b/mirrors/yum.py index f0684a2..24940a7 100644 --- a/mirrors/yum.py +++ b/mirrors/yum.py @@ -1,384 +1,388 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- - -""" -YUM/DNF镜像代理处理器 -支持RHEL/CentOS/Rocky/AlmaLinux软件源 -""" - -import os -import json -import time -import gzip -import xml.etree.ElementTree as ET -import urllib.request -from typing import Dict, List, Optional -from datetime import datetime - - -class YUMMirror: - """YUM/DNF镜像代理""" - - def __init__(self, config: dict): - self.config = config - - # 配置 - 使用 storage_dir（基于 base_dir） - self.base_url = config.get('base_url', 'http://mirror.centos.org/centos') - self.storage_dir = config.get('storage_dir', './downloads/yum') - self.base_dir = config.get('base_dir', './downloads') - self.repo_id = config.get('repo_id', 'baseos') - self.arch = config.get('arch', 'x86_64') - - # 确保存储目录存在 - os.makedirs(self.storage_dir, exist_ok=True) - - def handle_request(self, handler, path: str) -> bool: - """ - 处理YUM请求 - 路径格式: /centos/7/updates/x86_64/repodata/repomd.xml - """ - try: - parts = path.strip('/').split('/') - - if len(parts) < 3: - return self._handle_index(handler) - - distro = parts[0] # centos, rocky, alma - version = parts[1] # 7, 8, 9 - repo = parts[2] # baseos, appstream, updates - rest = '/'.join(parts[3:]) - - # 确定文件类型 - if 'repomd.xml' in rest: - return self._handle_repomd(handler, distro, version, repo) - elif 'primary.xml.gz' in rest: - return self._handle_primary(handler, distro, version, repo, 'primary') - elif 'filelists.xml.gz' in rest: - return self._handle_filelists(handler, distro, version, repo, 'filelists') - elif 'other.xml.gz' in rest: - return self._handle_other(handler, distro, version, repo, 'other') - else: - return self._handle_repo_file(handler, distro, version, repo, rest) - - except Exception as e: - handler.send_error(500, str(e)) - return False - - def _handle_index(self, handler) -> bool: - """处理索引请求""" - handler.send_json_response({ - 'base_url': self.base_url, - 'repo_id': self.repo_id, - 'arch': self.arch, - 'cache_stats': self.get_cache_stats() - }) - return True - - def _handle_repomd(self, handler, distro: str, version: str, repo: str) -> bool: - """处理repomd.xml请求""" - cache_key = f"repomd:{distro}:{version}:{repo}" - - cached = self._get_cache(cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'application/xml') - handler.send_header('Content-Length', str(len(cached))) - handler.end_headers() - handler.wfile.write(cached) - return True - - # 从上游获取 - url = f"{self.base_url}/{version}/{repo}/{self.arch}/repodata/repomd.xml" - - try: - data = self._fetch(url) - - if self.cache_enabled: - self._set_cache(cache_key, data) - - handler.send_response(200) - handler.send_header('Content-Type', 'application/xml') - handler.send_header('Content-Length', str(len(data))) - handler.end_headers() - handler.wfile.write(data) - return True - - except Exception as e: - handler.send_error(502, f"Failed to fetch repomd: {str(e)}") - return False - - def _handle_primary(self, handler, distro: str, version: str, repo: str, db_type: str) -> bool: - """处理primary.xml.gz""" - cache_key = f"primary:{distro}:{version}:{repo}:{self.arch}" - - cached = self._get_cache(cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'application/x-gzip') - handler.send_header('Content-Length', str(len(cached))) - handler.end_headers() - handler.wfile.write(cached) - return True - - # 先获取repomd.xml找到对应的数据库文件 - repomd_url = f"{self.base_url}/{version}/{repo}/{self.arch}/repodata/repomd.xml" - - try: - repomd_data = self._fetch(repomd_url) - - # 解析repomd.xml找到primary文件 - root = ET.fromstring(repomd_data) - ns = {'repomd': 'http://linux.duke.edu/metadata/repo'} - - data_location = None - for elem in root.findall('.//repomd:data', ns): - if elem.get('type') == 'primary': - data_location = elem.find('repomd:location', ns).get('href') - break - - if data_location: - db_url = f"{self.base_url}/{version}/{repo}/{self.arch}/repodata/{data_location}" - data = self._fetch(db_url) - - if self.cache_enabled: - self._set_cache(cache_key, data) - - handler.send_response(200) - handler.send_header('Content-Type', 'application/x-gzip') - handler.send_header('Content-Length', str(len(data))) - handler.end_headers() - handler.wfile.write(data) - return True - - except Exception as e: - pass - - handler.send_error(502, "Failed to fetch primary database") - return False - - def _handle_filelists(self, handler, distro: str, version: str, repo: str, db_type: str) -> bool: - """处理filelists.xml.gz""" - cache_key = f"filelists:{distro}:{version}:{repo}:{self.arch}" - - cached = self._get_cache(cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'application/x-gzip') - handler.send_header('Content-Length', str(len(cached))) - handler.end_headers() - handler.wfile.write(cached) - return True - - # 先获取repomd.xml找到对应的数据库文件 - repomd_url = f"{self.base_url}/{version}/{repo}/{self.arch}/repodata/repomd.xml" - - try: - repomd_data = self._fetch(repomd_url) - - # 解析repomd.xml找到filelists文件 - root = ET.fromstring(repomd_data) - ns = {'repomd': 'http://linux.duke.edu/metadata/repo'} - - data_location = None - for elem in root.findall('.//repomd:data', ns): - if elem.get('type') == 'filelists': - data_location = elem.find('repomd:location', ns).get('href') - break - - if data_location: - db_url = f"{self.base_url}/{version}/{repo}/{self.arch}/repodata/{data_location}" - data = self._fetch(db_url) - - if self.cache_enabled: - self._set_cache(cache_key, data) - - handler.send_response(200) - handler.send_header('Content-Type', 'application/x-gzip') - handler.send_header('Content-Length', str(len(data))) - handler.end_headers() - handler.wfile.write(data) - return True - - except Exception as e: - pass - - handler.send_error(502, "Failed to fetch filelists database") - return False - - def _handle_other(self, handler, distro: str, version: str, repo: str, db_type: str) -> bool: - """处理other.xml.gz""" - cache_key = f"other:{distro}:{version}:{repo}:{self.arch}" - - cached = self._get_cache(cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'application/x-gzip') - handler.send_header('Content-Length', str(len(cached))) - handler.end_headers() - handler.wfile.write(cached) - return True - - # 先获取repomd.xml找到对应的数据库文件 - repomd_url = f"{self.base_url}/{version}/{repo}/{self.arch}/repodata/repomd.xml" - - try: - repomd_data = self._fetch(repomd_url) - - # 解析repomd.xml找到other文件 - root = ET.fromstring(repomd_data) - ns = {'repomd': 'http://linux.duke.edu/metadata/repo'} - - data_location = None - for elem in root.findall('.//repomd:data', ns): - if elem.get('type') == 'other': - data_location = elem.find('repomd:location', ns).get('href') - break - - if data_location: - db_url = f"{self.base_url}/{version}/{repo}/{self.arch}/repodata/{data_location}" - data = self._fetch(db_url) - - if self.cache_enabled: - self._set_cache(cache_key, data) - - handler.send_response(200) - handler.send_header('Content-Type', 'application/x-gzip') - handler.send_header('Content-Length', str(len(data))) - handler.end_headers() - handler.wfile.write(data) - return True - - except Exception as e: - pass - - handler.send_error(502, "Failed to fetch other database") - return False - - def _handle_repo_file(self, handler, distro: str, version: str, repo: str, path: str) -> bool: - """处理仓库中的其他文件""" - cache_key = f"file:{distro}:{version}:{repo}:{path.replace('/', ':')}" - - cached = self._get_cache(cache_key) - if cached: - handler.send_response(200) - handler.send_header('Content-Type', 'application/octet-stream') - handler.end_headers() - handler.wfile.write(cached) - return True - - url = f"{self.base_url}/{version}/{repo}/{self.arch}/{path}" - - try: - data = self._fetch(url) - - if self.cache_enabled: - self._set_cache(cache_key, data) - - handler.send_response(200) - handler.send_header('Content-Type', 'application/octet-stream') - handler.end_headers() - handler.wfile.write(data) - return True - - except Exception as e: - handler.send_error(404, f"File not found: {str(e)}") - return False - - def _fetch(self, url: str) -> Optional[bytes]: - """从URL获取数据""" - try: - req = urllib.request.Request(url) - req.add_header('User-Agent', 'YUM-Mirror/1.0') - - with urllib.request.urlopen(req, timeout=30) as response: - return response.read() - - except Exception: - return None - - def _get_cache(self, cache_key: str) -> Optional[bytes]: - """获取缓存""" - if not self.cache_enabled: - return None - - cache_path = self._get_cache_path(cache_key) - meta_path = cache_path + '.meta' - - if not os.path.exists(cache_path): - return None - - if os.path.exists(meta_path): - try: - with open(meta_path, 'r') as f: - meta = json.load(f) - if time.time() > meta.get('expires', 0): - return None - except Exception: - pass - - try: - with open(cache_path, 'rb') as f: - return f.read() - except Exception: - return None - - def _set_cache(self, cache_key: str, data: bytes): - """设置缓存""" - cache_path = self._get_cache_path(cache_key) - meta_path = cache_path + '.meta' - - os.makedirs(os.path.dirname(cache_path), exist_ok=True) - - try: - with open(cache_path, 'wb') as f: - f.write(data) - - meta = { - 'cached_at': time.time(), - 'expires': time.time() + self.cache_ttl, - 'size': len(data) - } - - with open(meta_path, 'w') as f: - json.dump(meta, f) - - except Exception as e: - print(f"YUM缓存写入失败: {e}") - - def _get_cache_path(self, cache_key: str) -> str: - """获取缓存路径""" - subdir = cache_key[:2] - return os.path.join(self.storage_dir, subdir, cache_key) - - def get_cache_stats(self) -> dict: - """获取缓存统计""" - if not os.path.exists(self.storage_dir): - return {'files': 0, 'size': 0} - - total_size = 0 - file_count = 0 - - for root, dirs, files in os.walk(self.storage_dir): - for f in files: - if not f.endswith('.meta'): - file_count += 1 - total_size += os.path.getsize(os.path.join(root, f)) - - return { - 'files': file_count, - 'size': total_size, - 'size_formatted': self._format_size(total_size) - } - - def _format_size(self, size_bytes: int) -> str: - """格式化文件大小""" - if size_bytes == 0: - return "0 B" - - units = ["B", "KB", "MB", "GB"] - i = 0 - while size_bytes >= 1024 and i < len(units) - 1: - size_bytes /= 1024.0 - i += 1 - - return f"{size_bytes:.2f} {units[i]}" +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +""" +YUM/DNF镜像代理处理器 +支持RHEL/CentOS/Rocky/AlmaLinux软件源 +""" + +import os +import json +import time +import gzip +import xml.etree.ElementTree as ET +import urllib.request +from typing import Dict, List, Optional +from datetime import datetime + + +class YUMMirror: + """YUM/DNF镜像代理""" + + def __init__(self, config: dict): + self.config = config + + # 配置 - 使用 storage_dir（基于 base_dir） + self.base_url = config.get('base_url', 'http://mirror.centos.org/centos') + self.storage_dir = config.get('storage_dir', './downloads/yum') + self.base_dir = config.get('base_dir', './downloads') + self.repo_id = config.get('repo_id', 'baseos') + self.arch = config.get('arch', 'x86_64') + + # 缓存配置（缺失时默认启用，避免 AttributeError） + self.cache_enabled = config.get('cache_enabled', True) + self.cache_ttl = config.get('cache_ttl', 3600) + + # 确保存储目录存在 + os.makedirs(self.storage_dir, exist_ok=True) + + def handle_request(self, handler, path: str) -> bool: + """ + 处理YUM请求 + 路径格式: /centos/7/updates/x86_64/repodata/repomd.xml + """ + try: + parts = path.strip('/').split('/') + + if len(parts) < 3: + return self._handle_index(handler) + + distro = parts[0] # centos, rocky, alma + version = parts[1] # 7, 8, 9 + repo = parts[2] # baseos, appstream, updates + rest = '/'.join(parts[3:]) + + # 确定文件类型 + if 'repomd.xml' in rest: + return self._handle_repomd(handler, distro, version, repo) + elif 'primary.xml.gz' in rest: + return self._handle_primary(handler, distro, version, repo, 'primary') + elif 'filelists.xml.gz' in rest: + return self._handle_filelists(handler, distro, version, repo, 'filelists') + elif 'other.xml.gz' in rest: + return self._handle_other(handler, distro, version, repo, 'other') + else: + return self._handle_repo_file(handler, distro, version, repo, rest) + + except Exception as e: + handler.send_error(500, str(e)) + return False + + def _handle_index(self, handler) -> bool: + """处理索引请求""" + handler.send_json_response({ + 'base_url': self.base_url, + 'repo_id': self.repo_id, + 'arch': self.arch, + 'cache_stats': self.get_cache_stats() + }) + return True + + def _handle_repomd(self, handler, distro: str, version: str, repo: str) -> bool: + """处理repomd.xml请求""" + cache_key = f"repomd:{distro}:{version}:{repo}" + + cached = self._get_cache(cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'application/xml') + handler.send_header('Content-Length', str(len(cached))) + handler.end_headers() + handler.wfile.write(cached) + return True + + # 从上游获取 + url = f"{self.base_url}/{version}/{repo}/{self.arch}/repodata/repomd.xml" + + try: + data = self._fetch(url) + + if self.cache_enabled: + self._set_cache(cache_key, data) + + handler.send_response(200) + handler.send_header('Content-Type', 'application/xml') + handler.send_header('Content-Length', str(len(data))) + handler.end_headers() + handler.wfile.write(data) + return True + + except Exception as e: + handler.send_error(502, f"Failed to fetch repomd: {str(e)}") + return False + + def _handle_primary(self, handler, distro: str, version: str, repo: str, db_type: str) -> bool: + """处理primary.xml.gz""" + cache_key = f"primary:{distro}:{version}:{repo}:{self.arch}" + + cached = self._get_cache(cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'application/x-gzip') + handler.send_header('Content-Length', str(len(cached))) + handler.end_headers() + handler.wfile.write(cached) + return True + + # 先获取repomd.xml找到对应的数据库文件 + repomd_url = f"{self.base_url}/{version}/{repo}/{self.arch}/repodata/repomd.xml" + + try: + repomd_data = self._fetch(repomd_url) + + # 解析repomd.xml找到primary文件 + root = ET.fromstring(repomd_data) + ns = {'repomd': 'http://linux.duke.edu/metadata/repo'} + + data_location = None + for elem in root.findall('.//repomd:data', ns): + if elem.get('type') == 'primary': + data_location = elem.find('repomd:location', ns).get('href') + break + + if data_location: + db_url = f"{self.base_url}/{version}/{repo}/{self.arch}/repodata/{data_location}" + data = self._fetch(db_url) + + if self.cache_enabled: + self._set_cache(cache_key, data) + + handler.send_response(200) + handler.send_header('Content-Type', 'application/x-gzip') + handler.send_header('Content-Length', str(len(data))) + handler.end_headers() + handler.wfile.write(data) + return True + + except Exception as e: + pass + + handler.send_error(502, "Failed to fetch primary database") + return False + + def _handle_filelists(self, handler, distro: str, version: str, repo: str, db_type: str) -> bool: + """处理filelists.xml.gz""" + cache_key = f"filelists:{distro}:{version}:{repo}:{self.arch}" + + cached = self._get_cache(cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'application/x-gzip') + handler.send_header('Content-Length', str(len(cached))) + handler.end_headers() + handler.wfile.write(cached) + return True + + # 先获取repomd.xml找到对应的数据库文件 + repomd_url = f"{self.base_url}/{version}/{repo}/{self.arch}/repodata/repomd.xml" + + try: + repomd_data = self._fetch(repomd_url) + + # 解析repomd.xml找到filelists文件 + root = ET.fromstring(repomd_data) + ns = {'repomd': 'http://linux.duke.edu/metadata/repo'} + + data_location = None + for elem in root.findall('.//repomd:data', ns): + if elem.get('type') == 'filelists': + data_location = elem.find('repomd:location', ns).get('href') + break + + if data_location: + db_url = f"{self.base_url}/{version}/{repo}/{self.arch}/repodata/{data_location}" + data = self._fetch(db_url) + + if self.cache_enabled: + self._set_cache(cache_key, data) + + handler.send_response(200) + handler.send_header('Content-Type', 'application/x-gzip') + handler.send_header('Content-Length', str(len(data))) + handler.end_headers() + handler.wfile.write(data) + return True + + except Exception as e: + pass + + handler.send_error(502, "Failed to fetch filelists database") + return False + + def _handle_other(self, handler, distro: str, version: str, repo: str, db_type: str) -> bool: + """处理other.xml.gz""" + cache_key = f"other:{distro}:{version}:{repo}:{self.arch}" + + cached = self._get_cache(cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'application/x-gzip') + handler.send_header('Content-Length', str(len(cached))) + handler.end_headers() + handler.wfile.write(cached) + return True + + # 先获取repomd.xml找到对应的数据库文件 + repomd_url = f"{self.base_url}/{version}/{repo}/{self.arch}/repodata/repomd.xml" + + try: + repomd_data = self._fetch(repomd_url) + + # 解析repomd.xml找到other文件 + root = ET.fromstring(repomd_data) + ns = {'repomd': 'http://linux.duke.edu/metadata/repo'} + + data_location = None + for elem in root.findall('.//repomd:data', ns): + if elem.get('type') == 'other': + data_location = elem.find('repomd:location', ns).get('href') + break + + if data_location: + db_url = f"{self.base_url}/{version}/{repo}/{self.arch}/repodata/{data_location}" + data = self._fetch(db_url) + + if self.cache_enabled: + self._set_cache(cache_key, data) + + handler.send_response(200) + handler.send_header('Content-Type', 'application/x-gzip') + handler.send_header('Content-Length', str(len(data))) + handler.end_headers() + handler.wfile.write(data) + return True + + except Exception as e: + pass + + handler.send_error(502, "Failed to fetch other database") + return False + + def _handle_repo_file(self, handler, distro: str, version: str, repo: str, path: str) -> bool: + """处理仓库中的其他文件""" + cache_key = f"file:{distro}:{version}:{repo}:{path.replace('/', ':')}" + + cached = self._get_cache(cache_key) + if cached: + handler.send_response(200) + handler.send_header('Content-Type', 'application/octet-stream') + handler.end_headers() + handler.wfile.write(cached) + return True + + url = f"{self.base_url}/{version}/{repo}/{self.arch}/{path}" + + try: + data = self._fetch(url) + + if self.cache_enabled: + self._set_cache(cache_key, data) + + handler.send_response(200) + handler.send_header('Content-Type', 'application/octet-stream') + handler.end_headers() + handler.wfile.write(data) + return True + + except Exception as e: + handler.send_error(404, f"File not found: {str(e)}") + return False + + def _fetch(self, url: str) -> Optional[bytes]: + """从URL获取数据""" + try: + req = urllib.request.Request(url) + req.add_header('User-Agent', 'YUM-Mirror/1.0') + + with urllib.request.urlopen(req, timeout=30) as response: + return response.read() + + except Exception: + return None + + def _get_cache(self, cache_key: str) -> Optional[bytes]: + """获取缓存""" + if not self.cache_enabled: + return None + + cache_path = self._get_cache_path(cache_key) + meta_path = cache_path + '.meta' + + if not os.path.exists(cache_path): + return None + + if os.path.exists(meta_path): + try: + with open(meta_path, 'r') as f: + meta = json.load(f) + if time.time() > meta.get('expires', 0): + return None + except Exception: + pass + + try: + with open(cache_path, 'rb') as f: + return f.read() + except Exception: + return None + + def _set_cache(self, cache_key: str, data: bytes): + """设置缓存""" + cache_path = self._get_cache_path(cache_key) + meta_path = cache_path + '.meta' + + os.makedirs(os.path.dirname(cache_path), exist_ok=True) + + try: + with open(cache_path, 'wb') as f: + f.write(data) + + meta = { + 'cached_at': time.time(), + 'expires': time.time() + self.cache_ttl, + 'size': len(data) + } + + with open(meta_path, 'w') as f: + json.dump(meta, f) + + except Exception as e: + print(f"YUM缓存写入失败: {e}") + + def _get_cache_path(self, cache_key: str) -> str: + """获取缓存路径""" + subdir = cache_key[:2] + return os.path.join(self.storage_dir, subdir, cache_key) + + def get_cache_stats(self) -> dict: + """获取缓存统计""" + if not os.path.exists(self.storage_dir): + return {'files': 0, 'size': 0} + + total_size = 0 + file_count = 0 + + for root, dirs, files in os.walk(self.storage_dir): + for f in files: + if not f.endswith('.meta'): + file_count += 1 + total_size += os.path.getsize(os.path.join(root, f)) + + return { + 'files': file_count, + 'size': total_size, + 'size_formatted': self._format_size(total_size) + } + + def _format_size(self, size_bytes: int) -> str: + """格式化文件大小""" + if size_bytes == 0: + return "0 B" + + units = ["B", "KB", "MB", "GB"] + i = 0 + while size_bytes >= 1024 and i < len(units) - 1: + size_bytes /= 1024.0 + i += 1 + + return f"{size_bytes:.2f} {units[i]}" diff --git a/requirements.txt b/requirements.txt index ec88ce1..1802840 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,7 +2,7 @@ psutil>=5.9.0 pyinstaller>=6.0.0 python-multipart>=0.0.6 paramiko>=2.11.0 -aiohttp>=3.9.0 +bcrypt>=4.0.0 cachetools>=5.3.0 requests>=2.31.0 croniter>=2.0.0