From fb93949edd6a16c8061ad8245a7b4b53bc6ece2a Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Wed, 18 Mar 2026 17:36:46 +0800 Subject: [PATCH] fix: resolve ruff async path violations --- astrbot/core/star/star_manager.py | 17 +++++++++---- astrbot/dashboard/routes/chat.py | 38 +++++++++++++++++++----------- astrbot/dashboard/routes/config.py | 25 ++++++++++++-------- astrbot/dashboard/routes/stat.py | 33 ++++++++++++-------------- main.py | 6 +++-- 5 files changed, 70 insertions(+), 49 deletions(-) diff --git a/astrbot/core/star/star_manager.py b/astrbot/core/star/star_manager.py index c5379644a..ea599c3c3 100644 --- a/astrbot/core/star/star_manager.py +++ b/astrbot/core/star/star_manager.py @@ -1246,7 +1246,7 @@ class PluginManager: _, repo_name, _ = self.updator.parse_github_url(repo_url) repo_name = self.updator.format_name(repo_name) plugin_path = os.path.join(self.plugin_store_path, repo_name) - if os.path.exists(plugin_path): + if await anyio.Path(plugin_path).exists(): raise Exception( f"安装失败：目录 {os.path.basename(plugin_path)} 已存在。" ) @@ -1259,8 +1259,9 @@ class PluginManager: self.plugin_store_path, metadata_dir_name, ) - if target_plugin_path != plugin_path and os.path.exists( - target_plugin_path + if ( + target_plugin_path != plugin_path + and await anyio.Path(target_plugin_path).exists() ): raise Exception(f"安装失败：目录 {metadata_dir_name} 已存在。") if target_plugin_path != plugin_path: @@ -1649,7 +1650,10 @@ class PluginManager: self.plugin_store_path, metadata_dir_name, ) - if target_plugin_path != desti_dir and os.path.exists(target_plugin_path): + if ( + target_plugin_path != desti_dir + and await anyio.Path(target_plugin_path).exists() + ): raise Exception(f"安装失败：目录 {metadata_dir_name} 已存在。") if target_plugin_path != desti_dir: os.rename(desti_dir, target_plugin_path) @@ -1724,7 +1728,10 @@ class PluginManager: ) raise finally: - if temp_desti_dir != desti_dir and os.path.isdir(temp_desti_dir): + if ( + temp_desti_dir != desti_dir + and await anyio.Path(temp_desti_dir).is_dir() + ): try: remove_dir(temp_desti_dir) except Exception as e: diff --git a/astrbot/dashboard/routes/chat.py b/astrbot/dashboard/routes/chat.py index c79ad1e35..b4e7bb831 100644 --- a/astrbot/dashboard/routes/chat.py +++ b/astrbot/dashboard/routes/chat.py @@ -4,8 +4,10 @@ import os import re import uuid from contextlib import asynccontextmanager +from pathlib import Path from typing import cast +import anyio from quart import Response as QuartResponse from quart import g, make_response, request, send_file @@ -50,6 +52,10 @@ async def _poll_webchat_stream_result(back_queue, username: str): return result, False +def _resolve_path(path: str) -> Path: + return Path(path).resolve(strict=False) + + class ChatRoute(Route): def __init__( self, @@ -95,27 +101,29 @@ class ChatRoute(Route): try: file_path = os.path.join(self.attachments_dir, os.path.basename(filename)) - real_file_path = os.path.realpath(file_path) - real_imgs_dir = os.path.realpath(self.attachments_dir) + resolved_file_path = _resolve_path(file_path) + resolved_base_dir = _resolve_path(self.attachments_dir) - if not os.path.exists(real_file_path): + if not await anyio.Path(resolved_file_path).exists(): # try legacy file_path = os.path.join( self.legacy_img_dir, os.path.basename(filename) ) - if os.path.exists(file_path): - real_file_path = os.path.realpath(file_path) - real_imgs_dir = os.path.realpath(self.legacy_img_dir) + if await anyio.Path(file_path).exists(): + resolved_file_path = _resolve_path(file_path) + resolved_base_dir = _resolve_path(self.legacy_img_dir) - if not real_file_path.startswith(real_imgs_dir): + try: + resolved_file_path.relative_to(resolved_base_dir) + except ValueError: return Response().error("Invalid file path").__dict__ filename_ext = os.path.splitext(filename)[1].lower() if filename_ext == ".wav": - return await send_file(real_file_path, mimetype="audio/wav") + return await send_file(str(resolved_file_path), mimetype="audio/wav") if filename_ext[1:] in self.supported_imgs: - return await send_file(real_file_path, mimetype="image/jpeg") - return await send_file(real_file_path) + return await send_file(str(resolved_file_path), mimetype="image/jpeg") + return await send_file(str(resolved_file_path)) except (FileNotFoundError, OSError): return Response().error("File access error").__dict__ @@ -132,9 +140,11 @@ class ChatRoute(Route): return Response().error("Attachment not found").__dict__ file_path = attachment.path - real_file_path = os.path.realpath(file_path) + resolved_file_path = _resolve_path(file_path) - return await send_file(real_file_path, mimetype=attachment.mime_type) + return await send_file( + str(resolved_file_path), mimetype=attachment.mime_type + ) except (FileNotFoundError, OSError): return Response().error("File access error").__dict__ @@ -715,10 +725,10 @@ class ChatRoute(Route): try: attachments = await self.db.get_attachments(attachment_ids) for attachment in attachments: - if not os.path.exists(attachment.path): + if not await anyio.Path(attachment.path).exists(): continue try: - os.remove(attachment.path) + await anyio.Path(attachment.path).unlink() except OSError as e: logger.warning( f"Failed to delete attachment file {attachment.path}: {e}" diff --git a/astrbot/dashboard/routes/config.py b/astrbot/dashboard/routes/config.py index e057be780..960b48dcb 100644 --- a/astrbot/dashboard/routes/config.py +++ b/astrbot/dashboard/routes/config.py @@ -6,6 +6,7 @@ import traceback from pathlib import Path from typing import Any +import anyio from quart import request from astrbot.core import astrbot_config, file_token_service, logger @@ -40,6 +41,10 @@ from .util import ( MAX_FILE_BYTES = 500 * 1024 * 1024 +def _resolve_path(path: Path) -> Path: + return path.resolve(strict=False) + + def try_cast(value: Any, type_: str): if type_ == "int": try: @@ -1105,8 +1110,8 @@ class ConfigRoute(Route): if not files: return Response().error("No files uploaded").__dict__ - storage_root_path = Path(get_astrbot_plugin_data_path()).resolve(strict=False) - plugin_root_path = (storage_root_path / name).resolve(strict=False) + storage_root_path = _resolve_path(Path(get_astrbot_plugin_data_path())) + plugin_root_path = _resolve_path(storage_root_path / name) try: plugin_root_path.relative_to(storage_root_path) except ValueError: @@ -1133,7 +1138,7 @@ class ConfigRoute(Route): continue rel_path = f"files/{folder}/{filename}" - save_path = (plugin_root_path / rel_path).resolve(strict=False) + save_path = _resolve_path(plugin_root_path / rel_path) try: save_path.relative_to(plugin_root_path) except ValueError: @@ -1181,13 +1186,13 @@ class ConfigRoute(Route): if not md: return Response().error(f"Plugin {name} not found").__dict__ - storage_root_path = Path(get_astrbot_plugin_data_path()).resolve(strict=False) - plugin_root_path = (storage_root_path / name).resolve(strict=False) + storage_root_path = _resolve_path(Path(get_astrbot_plugin_data_path())) + plugin_root_path = _resolve_path(storage_root_path / name) try: plugin_root_path.relative_to(storage_root_path) except ValueError: return Response().error("Invalid name parameter").__dict__ - target_path = (plugin_root_path / rel_path).resolve(strict=False) + target_path = _resolve_path(plugin_root_path / rel_path) try: target_path.relative_to(plugin_root_path) except ValueError: @@ -1209,15 +1214,15 @@ class ConfigRoute(Route): if not meta or meta.get("type") != "file": return Response().error("Config item not found or not file type").__dict__ - storage_root_path = Path(get_astrbot_plugin_data_path()).resolve(strict=False) - plugin_root_path = (storage_root_path / name).resolve(strict=False) + storage_root_path = _resolve_path(Path(get_astrbot_plugin_data_path())) + plugin_root_path = _resolve_path(storage_root_path / name) try: plugin_root_path.relative_to(storage_root_path) except ValueError: return Response().error("Invalid name parameter").__dict__ folder = config_key_to_folder(key_path) - target_dir = (plugin_root_path / "files" / folder).resolve(strict=False) + target_dir = _resolve_path(plugin_root_path / "files" / folder) try: target_dir.relative_to(plugin_root_path) except ValueError: @@ -1377,7 +1382,7 @@ class ConfigRoute(Route): logo_file_path = os.path.join(plugin_dir, platform.logo_path) # 检查文件是否存在并注册令牌 - if os.path.exists(logo_file_path): + if await anyio.Path(logo_file_path).exists(): logo_token = await file_token_service.register_file( logo_file_path, expire_seconds=3600, diff --git a/astrbot/dashboard/routes/stat.py b/astrbot/dashboard/routes/stat.py index 532238ac7..2f3abb5fd 100644 --- a/astrbot/dashboard/routes/stat.py +++ b/astrbot/dashboard/routes/stat.py @@ -7,6 +7,7 @@ from functools import cmp_to_key from pathlib import Path import aiohttp +import anyio import psutil from quart import request @@ -22,6 +23,10 @@ from astrbot.core.utils.version_comparator import VersionComparator from .route import Response, Route, RouteContext +def _resolve_path(path: str | Path) -> Path: + return Path(path).resolve(strict=False) + + class StatRoute(Route): def __init__( self, @@ -210,41 +215,33 @@ class StatRoute(Route): filename = f"v{version}.md" project_path = get_astrbot_path() - changelogs_dir = os.path.join(project_path, "changelogs") - changelog_path = os.path.join(changelogs_dir, filename) - - # 规范化路径，防止符号链接攻击 - changelog_path = os.path.realpath(changelog_path) - changelogs_dir = os.path.realpath(changelogs_dir) + changelogs_dir = _resolve_path(Path(project_path) / "changelogs") + changelog_path = _resolve_path(changelogs_dir / filename) # 验证最终路径在预期的 changelogs 目录内（防止路径遍历） - # 确保规范化后的路径以 changelogs_dir 开头，且是目录内的文件 - changelog_path_normalized = os.path.normpath(changelog_path) - changelogs_dir_normalized = os.path.normpath(changelogs_dir) - - # 检查路径是否在预期目录内（必须是目录的子文件，不能是目录本身） - expected_prefix = changelogs_dir_normalized + os.sep - if not changelog_path_normalized.startswith(expected_prefix): + try: + changelog_path.relative_to(changelogs_dir) + except ValueError: logger.warning( f"Path traversal attempt detected: {version} -> {changelog_path}", ) return Response().error("Invalid version format").__dict__ - if not os.path.exists(changelog_path): + if not await anyio.Path(changelog_path).exists(): return ( Response() .error(f"Changelog for version {version} not found") .__dict__ ) - if not os.path.isfile(changelog_path): + if not await anyio.Path(changelog_path).is_file(): return ( Response() .error(f"Changelog for version {version} not found") .__dict__ ) - with open(changelog_path, encoding="utf-8") as f: - content = f.read() + async with await anyio.open_file(changelog_path, encoding="utf-8") as f: + content = await f.read() return Response().ok({"content": content, "version": version}).__dict__ except Exception as e: @@ -257,7 +254,7 @@ class StatRoute(Route): project_path = get_astrbot_path() changelogs_dir = os.path.join(project_path, "changelogs") - if not os.path.exists(changelogs_dir): + if not await anyio.Path(changelogs_dir).exists(): return Response().ok({"versions": []}).__dict__ versions = [] diff --git a/main.py b/main.py index bbeba2de9..7a9154301 100644 --- a/main.py +++ b/main.py @@ -5,6 +5,8 @@ import os import sys from pathlib import Path +import anyio + import runtime_bootstrap from astrbot.core import LogBroker, LogManager, db_helper, logger from astrbot.core.config.default import VERSION @@ -69,13 +71,13 @@ async def check_dashboard_files(webui_dir: str | None = None): """下载管理面板文件""" # 指定webui目录 if webui_dir: - if os.path.exists(webui_dir): + if await anyio.Path(webui_dir).exists(): logger.info(f"使用指定的 WebUI 目录: {webui_dir}") return webui_dir logger.warning(f"指定的 WebUI 目录 {webui_dir} 不存在，将使用默认逻辑。") data_dist_path = os.path.join(get_astrbot_data_path(), "dist") - if os.path.exists(data_dist_path): + if await anyio.Path(data_dist_path).exists(): v = await get_dashboard_version() if v is not None: # 存在文件