Merge pull request #4417 from pipecat-ai/mb/resolve-runner-filepath

tests/test_runner_downloads.py

@@ -0,0 +1,84 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import os
+import tempfile
+import unittest
+from pathlib import Path
+
+from fastapi import HTTPException
+
+from pipecat.runner.run import _resolve_download_path
+
+
+class TestRunnerDownloads(unittest.TestCase):
+    def test_resolve_download_path_allows_files_inside_folder(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            downloads = Path(tmpdir) / "downloads"
+            nested = downloads / "nested"
+            nested.mkdir(parents=True)
+            file_path = nested / "recording.txt"
+            file_path.write_text("session transcript")
+
+            resolved = _resolve_download_path(str(downloads), "nested/recording.txt")
+
+            self.assertEqual(resolved, file_path.resolve())
+
+    def test_resolve_download_path_blocks_parent_traversal(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            root = Path(tmpdir)
+            downloads = root / "downloads"
+            downloads.mkdir()
+            (root / "secret.txt").write_text("secret")
+
+            with self.assertRaises(HTTPException) as context:
+                _resolve_download_path(str(downloads), "../secret.txt")
+
+            self.assertEqual(context.exception.status_code, 403)
+
+    def test_resolve_download_path_blocks_decoded_encoded_slashes(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            root = Path(tmpdir)
+            downloads = root / "media"
+            downloads.mkdir()
+            outside = root / "outside"
+            outside.mkdir()
+            (outside / "secret.txt").write_text("secret")
+
+            with self.assertRaises(HTTPException) as context:
+                _resolve_download_path(str(downloads), "../outside/secret.txt")
+
+            self.assertEqual(context.exception.status_code, 403)
+
+    def test_resolve_download_path_blocks_absolute_paths(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            downloads = Path(tmpdir) / "downloads"
+            downloads.mkdir()
+
+            with self.assertRaises(HTTPException) as context:
+                _resolve_download_path(str(downloads), "/etc/passwd")
+
+            self.assertEqual(context.exception.status_code, 403)
+
+    @unittest.skipUnless(hasattr(os, "symlink"), "os.symlink is not available")
+    def test_resolve_download_path_blocks_symlink_escape(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            root = Path(tmpdir)
+            downloads = root / "downloads"
+            outside = root / "outside"
+            downloads.mkdir()
+            outside.mkdir()
+            (outside / "secret.txt").write_text("secret")
+
+            try:
+                os.symlink(outside, downloads / "linked")
+            except OSError as e:
+                self.skipTest(f"Unable to create symlink: {e}")
+
+            with self.assertRaises(HTTPException) as context:
+                _resolve_download_path(str(downloads), "linked/secret.txt")
+
+            self.assertEqual(context.exception.status_code, 403)