refactor: 修复告警

xiaomusic/httpserver.py

@@ -499,6 +499,15 @@ def access_key_verification(file_path, key, code):
 range_pattern = re.compile(r"bytes=(\d+)-(\d*)")
 
 
+def safe_redirect(config, url):
+    url = try_add_access_control_param(config, url)
+    url = url.replace("\\", "")
+    if not urllib.parse.urlparse(url).netloc and not urllib.parse.urlparse(url).scheme:
+        RedirectResponse(url=url)
+        return True
+    return False
+
+
 @app.get("/music/{file_path:path}")
 async def music_file(request: Request, file_path: str, key: str = "", code: str = ""):
     if not access_key_verification(f"/music/{file_path}", key, code):
@@ -517,8 +526,8 @@ async def music_file(request: Request, file_path: str, key: str = "", code: str
         temp_mp3_file = remove_id3_tags(absolute_file_path, config)
         if temp_mp3_file:
             log.info(f"ID3 tag removed {absolute_file_path} to {temp_mp3_file}")
-            url = try_add_access_control_param(config, f"/music/{temp_mp3_file}")
-            return RedirectResponse(url=url)
+            if safe_redirect(f"/music/{temp_mp3_file}"):
+                return
         else:
             log.info(f"No ID3 tag remove needed: {absolute_file_path}")
 
@@ -526,8 +535,8 @@ async def music_file(request: Request, file_path: str, key: str = "", code: str
         temp_mp3_file = convert_file_to_mp3(absolute_file_path, config)
         if temp_mp3_file:
             log.info(f"Converted file: {absolute_file_path} to {temp_mp3_file}")
-            url = try_add_access_control_param(config, f"/music/{temp_mp3_file}")
-            return RedirectResponse(url=url)
+            if safe_redirect(f"/music/{temp_mp3_file}"):
+                return
         else:
             log.warning(f"Failed to convert file to MP3 format: {absolute_file_path}")
 

xiaomusic/utils.py

@@ -463,6 +463,11 @@ def convert_file_to_mp3(input_file: str, config) -> str:
         log.info(f"File {input_file} = {out_file_path} . Skipping convert_file_to_mp3.")
         return None
 
+    absolute_music_path = os.path.abspath(music_path)
+    if not input_absolute_path.startswith(absolute_music_path):
+        log.error(f"Invalid input file path: {input_file}")
+        return None
+
     # 检查目标文件是否存在
     if os.path.exists(out_file_path):
         log.info(f"File {out_file_path} already exists. Skipping convert_file_to_mp3.")
@@ -471,7 +476,7 @@ def convert_file_to_mp3(input_file: str, config) -> str:
     command = [
         os.path.join(config.ffmpeg_location, "ffmpeg"),
         "-i",
-        input_file,
+        input_absolute_path,
         "-f",
         "mp3",
         "-vn",