From ffae0286644032b9bc0d72e48196f748485dfbe1 Mon Sep 17 00:00:00 2001 From: HYC Fixer Date: Sun, 30 Aug 2026 12:16:14 +0800 Subject: [PATCH] =?UTF-8?q?P0-A:=20=E4=BF=AE=E5=A4=8D=E9=89=B4=E6=9D=83?= =?UTF-8?q?=E4=BD=93=E7=B3=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - check_endpoint_auth 路径归一化(无前导斜杠),v1/v2 端点级鉴权真正生效 - ADMIN_API_ENDPOINTS 重建:修复 files 单复数、补 mirrors/alerts/prewarm/api-docs/user-password - handler check_auth 方法感知:公开端点仅 GET/HEAD,写操作默认要求认证(fail-closed) - 补全受保护端点(monitor/activity/login-logs/mirrors/alerts/prewarm/file 元数据) - _check_basic_auth/_check_token_auth 改用 hmac.compare_digest - do_POST/do_HEAD 解析路径时剥离查询串;do_HEAD 增加 is_safe_path 检查 - is_safe_path 改用 realpath 防符号链接绕过 --- api/v1.py | 11 +++++ api/v2.py | 4 +- core/api_auth.py | 102 +++++++++++++++++++++++++++----------- core/utils.py | 9 ++-- handlers/http_handler.py | 104 +++++++++++++++++++++++++++------------ 5 files changed, 164 insertions(+), 66 deletions(-) diff --git a/api/v1.py b/api/v1.py index fa586ae..9252d9f 100644 --- a/api/v1.py +++ b/api/v1.py @@ -85,6 +85,17 @@ class APIv1: }) return + if auth_check.get('permission'): + if not auth_manager.check_permission(auth_result, auth_check['permission']): + handler.send_json_response({ + "error": "权限不足", + "code": "FORBIDDEN", + "required_permission": auth_check['permission'] + }, 403) + return + + handler.auth_result = auth_result + # 文件管理API (GET /api/v1/files 不需要认证) if path == 'files': if method == 'GET': diff --git a/api/v2.py b/api/v2.py index 401900d..128f766 100644 --- a/api/v2.py +++ b/api/v2.py @@ -83,7 +83,6 @@ class APIv2(APIv1): public_endpoints = [ 'admin/auth/verify', 'user/login', - 'user/password', 'search/enhanced', 'search/by-tag', 'search/by-date', @@ -100,7 +99,8 @@ class APIv2(APIv1): # 如果需要认证（不是公开端点且auth_type不是none） if auth_manager and not is_public and not skip_auth: - auth_check = check_endpoint_auth(method, path, auth_manager) + # 传入完整路径（api/v2/ 前缀），与 ADMIN_API_ENDPOINTS 规则匹配 + auth_check = check_endpoint_auth(method, f"api/v2/{path}", auth_manager) if auth_check['required']: auth_result = auth_manager.validate_request(handler, 'admin') if not auth_result.get('authenticated'): diff --git a/core/api_auth.py b/core/api_auth.py index 26e726f..1820eb7 100644 --- a/core/api_auth.py +++ b/core/api_auth.py @@ -520,39 +520,53 @@ def require_auth(required_level: str = "admin", permission: str = None): # === 需要认证的API端点定义 === +# 规则格式: 'METHOD:/api/vN/path' 或 'METHOD:/api/vN/prefix/'（前缀规则,尾斜杠） +# 匹配时路径统一归一化为无前导斜杠形式,与 v1/v2 传入的 api_action 对齐 ADMIN_API_ENDPOINTS = { # 同步管理 - 'POST:/api/v2/sync/*': 'sync:manage', - 'POST:/api/v2/sync/*/start': 'sync:start', - 'POST:/api/v2/sync/*/stop': 'sync:stop', - 'DELETE:/api/v2/sync/*': 'sync:manage', + 'POST:/api/v2/sync/': 'sync:manage', + 'DELETE:/api/v2/sync/': 'sync:manage', # 缓存管理 'POST:/api/v2/cache/clean': 'cache:manage', - 'DELETE:/api/v2/cache/*': 'cache:manage', + 'POST:/api/v2/cache/prewarm/': 'cache:manage', + 'DELETE:/api/v2/cache/prewarm/': 'cache:manage', # Webhook管理 'POST:/api/v2/webhooks': 'webhook:create', - 'PUT:/api/v2/webhooks/*': 'webhook:update', - 'DELETE:/api/v2/webhooks/*': 'webhook:delete', - 'POST:/api/v2/webhooks/*/trigger': 'webhook:trigger', + 'PUT:/api/v2/webhooks/': 'webhook:update', + 'DELETE:/api/v2/webhooks/': 'webhook:delete', + 'POST:/api/v2/webhooks/': 'webhook:trigger', # 服务器配置 'PUT:/api/v2/config': 'config:manage', 'POST:/api/v2/server/reload': 'server:reload', + 'POST:/api/v2/server/': 'server:manage', - # 文件管理（高危操作） - 'DELETE:/api/v2/files/*': 'files:delete', - 'PUT:/api/v2/files/*/rename': 'files:rename', + # 文件管理（高危操作: 删除/重命名/元数据/版本） + 'DELETE:/api/v1/file/': 'files:delete', + 'DELETE:/api/v2/file/': 'files:delete', + 'PUT:/api/v2/file/': 'files:update', + 'POST:/api/v2/file/': 'files:update', # 用户管理 'POST:/api/v2/users': 'users:create', - 'DELETE:/api/v2/users/*': 'users:delete', - 'PUT:/api/v2/users/*': 'users:update', + 'DELETE:/api/v2/users/': 'users:delete', + 'PUT:/api/v2/users/': 'users:update', + 'POST:/api/v2/user/password': 'users:update', + + # 镜像管理（写 settings.json,必须鉴权） + 'POST:/api/v2/mirrors': 'mirrors:manage', + 'PUT:/api/v2/mirrors/': 'mirrors:manage', + 'DELETE:/api/v2/mirrors/': 'mirrors:manage', + + # 告警配置与API文档生成 + 'PUT:/api/v2/alerts': 'config:manage', + 'POST:/api/v2/alerts': 'config:manage', + 'POST:/api/v2/api-docs/generate': 'config:manage', # === API v1 文件操作认证 === - 'DELETE:/api/v1/file/*': 'files:delete', 'PUT:/api/v1/mkdir': 'files:create', 'POST:/api/v1/upload': 'files:upload', 'POST:/api/v1/batch': 'files:batch', @@ -560,26 +574,56 @@ ADMIN_API_ENDPOINTS = { } +def _normalize_endpoint_patterns(): + """把 ADMIN_API_ENDPOINTS 归一化为 (精确表, 前缀表) + + 规则格式: 'METHOD:/api/vN/path' + - 无通配符 → 精确匹配 (method, path) + - 尾斜杠或尾* → 前缀匹配, 匹配 norm_path.startswith(prefix + '/') + """ + exact = {} + prefix = [] + for pattern, permission in ADMIN_API_ENDPOINTS.items(): + pat_method, pat_path = pattern.split(':', 1) + pat_path = pat_path.lstrip('/') + if '*' in pattern or pat_path.endswith('/'): + if pat_path.endswith('*'): + pat_path = pat_path.rstrip('*') + pat_path = pat_path.rstrip('/') + prefix.append((pat_method, pat_path, permission)) + else: + exact[(pat_method, pat_path)] = permission + return exact, prefix + + +_EXACT_ENDPOINTS, _PREFIX_ENDPOINTS = _normalize_endpoint_patterns() + + def check_endpoint_auth(method: str, path: str, auth_manager: APIAuthManager) -> dict: - """检查端点是否需要认证""" - key = f"{method}:{path}" - if key in ADMIN_API_ENDPOINTS: + """检查端点是否需要认证 + + 路径统一归一化为无前导斜杠形式（如 'api/v2/sync/sources'）， + 与 v1/v2 传入的 api_action 保持一致。 + """ + norm_path = path.lstrip('/') + + # 精确匹配 + key = (method, norm_path) + if key in _EXACT_ENDPOINTS: return { "required": True, - "permission": ADMIN_API_ENDPOINTS[key] + "permission": _EXACT_ENDPOINTS[key] } - for pattern, permission in ADMIN_API_ENDPOINTS.items(): - if '*' in pattern: - pat_method, pat_path = pattern.split(':', 1) - if method == pat_method or pat_method == '*': - if pat_path.endswith('*'): - prefix = pat_path.rstrip('*').rstrip('/') - if path.startswith(prefix): - return { - "required": True, - "permission": permission - } + # 前缀匹配 + for pat_method, pat_path, permission in _PREFIX_ENDPOINTS: + if method != pat_method and pat_method != '*': + continue + if norm_path.startswith(pat_path + '/'): + return { + "required": True, + "permission": permission + } return { "required": False, diff --git a/core/utils.py b/core/utils.py index 36346b8..3551853 100644 --- a/core/utils.py +++ b/core/utils.py @@ -84,10 +84,13 @@ def sanitize_filename(filename: str) -> str: def is_safe_path(base_dir: str, path: str) -> bool: - """检查路径是否安全（防止目录遍历）""" + """检查路径是否安全（防止目录遍历） + + 使用 realpath 解析符号链接，避免 base_dir 内 symlink 指向外部目录的绕过。 + """ try: - abs_path = os.path.abspath(path) - abs_base = os.path.abspath(base_dir) + abs_path = os.path.realpath(path) + abs_base = os.path.realpath(base_dir) common_path = os.path.commonpath([abs_path, abs_base]) return common_path == abs_base except ValueError: diff --git a/handlers/http_handler.py b/handlers/http_handler.py index fda6252..febf84d 100644 --- a/handlers/http_handler.py +++ b/handlers/http_handler.py @@ -254,7 +254,7 @@ class MirrorServerHandler(BaseHTTPRequestHandler): return handler def check_auth(self, path=None): - """检查认证""" + """检查认证（方法感知: 读操作与写操作区分对待）""" # 获取检查路径 if path is not None: check_path = path @@ -263,6 +263,10 @@ class MirrorServerHandler(BaseHTTPRequestHandler): else: check_path = '' + # 去掉查询串（do_POST/do_HEAD 传入的路径可能带 ?query） + if '?' in check_path: + check_path = check_path.split('?', 1)[0] + # 根路径直接放行 if not check_path or check_path == '/': return True @@ -270,10 +274,20 @@ class MirrorServerHandler(BaseHTTPRequestHandler): if not self.config: return True - # 公开端点(不需要认证)- 文件只读操作 - public_endpoints = [ - # 登录 + auth_type = self.config.get('auth_type', 'none') + if auth_type == 'none': + return True + + method = (getattr(self, 'command', '') or 'GET').upper() + + # 任何方法都公开的端点（登录、认证状态查询等） + public_any = [ 'api/v2/user/login', + 'api/v2/admin/auth/verify', + ] + + # 只读公开端点（仅 GET/HEAD 放行） + public_get = [ # 文件只读:列表/搜索/下载/mirror/mc 'api/v1/files', 'api/v1/file/', @@ -288,17 +302,20 @@ class MirrorServerHandler(BaseHTTPRequestHandler): 'api/v2/health', 'api/v2/stats/', 'api/v2/cache/stats', + 'api/v2/cache/popular', + 'api/v2/api-docs.yaml', + # 镜像加速源代理下载（只读） + 'api/v2/mirrors/pypi', + 'api/v2/mirrors/npm', + 'api/v2/mirrors/go', + 'api/v2/mirrors/docker', ] - # 检查是否是公开端点 - for endpoint in public_endpoints: - if check_path == endpoint or check_path.startswith(endpoint + '/'): - return True - - # 需要认证的端点 - 所有修改操作 + # 受保护前缀（GET/HEAD 命中也需要认证；写操作另有兜底） protected_endpoints = [ # 用户操作 'api/v2/user/password', # 改密码 + 'api/v2/user/login-logs', # 登录日志（含IP） 'api/v2/users', # 文件修改操作 'api/v1/upload', @@ -314,27 +331,41 @@ class MirrorServerHandler(BaseHTTPRequestHandler): 'api/v2/config', 'api/v2/server/', 'api/v2/cache/clean', + 'api/v2/cache/prewarm', 'api/v2/webhooks', 'api/v2/sync/', - 'api/v2/file/', # 文件删除/重命名 + 'api/v2/file/', # 文件删除/重命名/元数据/版本/缩略图 + 'api/v2/mirrors', # 镜像管理（写 settings.json） + 'api/v2/alerts', + 'api/v2/activity', + 'api/v2/monitor', ] - # 检查是否是需要认证的端点 - is_protected = False - for endpoint in protected_endpoints: - if check_path.startswith(endpoint): - is_protected = True - break + # 任何方法都公开 + for endpoint in public_any: + if check_path == endpoint or check_path.startswith(endpoint + '/'): + return True - # 如果不是受保护端点,直接放行 - if not is_protected: + if method in ('GET', 'HEAD'): + # 只读公开端点放行 + for endpoint in public_get: + if check_path == endpoint or check_path.startswith(endpoint + '/'): + return True + # 其余 GET: 命中受保护前缀才需要认证 + for endpoint in protected_endpoints: + if check_path.startswith(endpoint): + return self._do_auth(auth_type) return True - # 只有受保护端点才需要认证检查 - auth_type = self.config.get('auth_type', 'none') - if auth_type == 'none': - return True - elif auth_type == 'basic': + # 写方法（POST/PUT/DELETE）: API 路径一律要求认证，防止新增端点漏配 + if check_path.startswith('api/'): + return self._do_auth(auth_type) + + return True + + def _do_auth(self, auth_type): + """执行指定类型的认证检查""" + if auth_type == 'basic': return self._check_basic_auth() elif auth_type == 'token': return self._check_token_auth() @@ -342,6 +373,7 @@ class MirrorServerHandler(BaseHTTPRequestHandler): def _check_basic_auth(self): """检查基本认证""" + import hmac auth_header = self.headers.get('Authorization') if not auth_header or not auth_header.startswith('Basic '): self.send_auth_required() @@ -352,8 +384,9 @@ class MirrorServerHandler(BaseHTTPRequestHandler): username, password = auth_decoded.split(':', 1) expected_user = self.config.get('auth_user', 'admin') if self.config else 'admin' expected_pass = self.config.get('auth_pass', 'admin123') if self.config else 'admin123' - - if username == expected_user and password == expected_pass: + + # 恒定时间比较，防时序攻击 + if hmac.compare_digest(username, expected_user) and hmac.compare_digest(password, expected_pass): return True else: self.send_auth_required() @@ -390,7 +423,8 @@ class MirrorServerHandler(BaseHTTPRequestHandler): # 检查是否是静态 token expected_token = self.config.get('auth_token') if self.config else None - if token and token == expected_token: + import hmac + if token and expected_token and hmac.compare_digest(token, expected_token): return True # 验证失败 @@ -504,7 +538,10 @@ class MirrorServerHandler(BaseHTTPRequestHandler): if self.api_router is None and self.config is not None: MirrorServerHandler.api_router = APIRouter(self.config) - path = unquote(self.path).lstrip('/') + # 解析路径（去掉查询串，避免把 ?query 拼进 API 路径） + parsed_path = urlparse(self.path) + path = unquote(parsed_path.path).lstrip('/') + query = parsed_path.query # 调试模式输出完整路径 (debug-http) if self._is_debug_enabled('http'): @@ -516,8 +553,6 @@ class MirrorServerHandler(BaseHTTPRequestHandler): return if path.startswith("api/"): - parsed_path = urlparse(self.path) - query = parsed_path.query self.api_router.handle_request(self, 'POST', path, query) else: self.send_error(405) @@ -601,11 +636,16 @@ class MirrorServerHandler(BaseHTTPRequestHandler): """处理HEAD请求""" if not self.check_auth(): return - path = unquote(self.path).lstrip('/') + # 解析路径（去掉查询串），并复用 serve_path 的安全检查防路径穿越 + parsed_path = urlparse(self.path) + rel_path = unquote(parsed_path.path).lstrip('/') if self.config is None: self.send_error(500) return - file_path = os.path.join(self.config['base_dir'], path) + file_path = os.path.join(self.config['base_dir'], rel_path) + if not is_safe_path(self.config['base_dir'], file_path): + self.send_error(403, "Access denied") + return if os.path.isfile(file_path): self.send_file_headers(file_path) else: