diff --git a/comfy_execution/media_enrichment.py b/comfy_execution/media_enrichment.py index f0149a0ce..c83beed62 100644 --- a/comfy_execution/media_enrichment.py +++ b/comfy_execution/media_enrichment.py @@ -16,29 +16,37 @@ import os def resolve_output_entry_path(entry) -> str | None: """Resolve a file-type output entry to an absolute path inside its base dir. - Returns ``None`` for non-file entries (no ``filename``/``type``), unknown - folder types, paths that escape the folder-type base directory, and paths - with no file on disk. + Returns ``None`` for non-file entries (no ``filename``/``type``, or + non-string fields), unknown folder types, paths that escape the + folder-type base directory (symlinks are resolved before the containment + check), and paths with no file on disk. """ import folder_paths if not isinstance(entry, dict) or "filename" not in entry or "type" not in entry: return None + filename = entry["filename"] + subfolder = entry.get("subfolder") or "" + if not isinstance(filename, str) or not isinstance(subfolder, str) or not isinstance(entry["type"], str): + return None base = folder_paths.get_directory_by_type(entry["type"]) if base is None: return None - base_abs = os.path.abspath(base) - abs_path = os.path.abspath(os.path.join(base_abs, entry.get("subfolder") or "", entry["filename"])) + # realpath (not abspath) so a symlink planted inside the base directory + # can't smuggle an outside target past the containment check; the base is + # resolved too since output/temp dirs are themselves commonly symlinks. + base_real = os.path.realpath(base) + real_path = os.path.realpath(os.path.join(base_real, subfolder, filename)) try: - if os.path.commonpath([base_abs, abs_path]) != base_abs: - logging.warning("Output entry path escapes base directory: %s", entry.get("filename")) + if os.path.commonpath([base_real, real_path]) != base_real: + logging.warning("Output entry path escapes base directory: %s", filename) return None except ValueError: - logging.warning("Output entry path escapes base directory: %s", entry.get("filename")) + logging.warning("Output entry path escapes base directory: %s", filename) return None - if not os.path.isfile(abs_path): + if not os.path.isfile(real_path): return None - return abs_path + return real_path def enrich_output_with_media_metadata(output_ui: dict) -> dict: @@ -52,8 +60,11 @@ def enrich_output_with_media_metadata(output_ui: dict) -> dict: """ try: from app.assets.services.media_metadata import extract_media_metadata - except ImportError: - logging.debug("Media metadata extraction unavailable; skipping enrichment") + except Exception: + # Broad on purpose: this enrichment is best-effort and is called + # unguarded from the output-processing path, so a dependency failing + # to import for any reason must degrade to a no-op, not block outputs. + logging.warning("Media metadata extraction unavailable; skipping enrichment", exc_info=True) return output_ui enriched = {} diff --git a/tests-unit/execution_test/test_media_enrichment.py b/tests-unit/execution_test/test_media_enrichment.py index 44cbfab7f..8116d29bb 100644 --- a/tests-unit/execution_test/test_media_enrichment.py +++ b/tests-unit/execution_test/test_media_enrichment.py @@ -128,6 +128,64 @@ class TestEnrichOutputWithMediaMetadata(unittest.TestCase): result = mod.enrich_output_with_media_metadata(output) self.assertNotIn("metadata", result["images"][0]) + def test_extractor_import_failure_beyond_importerror_degrades(self): + class ExplodingModule: + def __getattr__(self, name): + raise RuntimeError("dependency init failed") + + output = {"images": [{"filename": "a.mp4", "subfolder": "", "type": "output"}]} + mocked = { + "folder_paths": MagicMock(get_directory_by_type=MagicMock(return_value=_DEFAULT_BASE)), + "app.assets.services.media_metadata": ExplodingModule(), + } + with patch.dict("sys.modules", mocked), \ + patch("os.path.isfile", return_value=True): + import comfy_execution.media_enrichment as mod + result = mod.enrich_output_with_media_metadata(output) + self.assertNotIn("metadata", result["images"][0]) + + def test_non_string_fields_skipped(self): + output = { + "images": [ + {"filename": 42, "subfolder": "", "type": "output"}, + {"filename": "a.mp4", "subfolder": ["nested"], "type": "output"}, + {"filename": "b.mp4", "subfolder": "", "type": 7}, + ] + } + result, extract_mock = _call(output) + for entry in result["images"]: + self.assertNotIn("metadata", entry) + extract_mock.assert_not_called() + + def test_symlink_escape_rejected_real_fs(self): + import shutil + import tempfile + + root = tempfile.mkdtemp(prefix="media-enrichment-symlink-") + try: + base = os.path.join(root, "output") + os.makedirs(base) + secret = os.path.join(root, "secret.txt") + with open(secret, "w") as f: + f.write("outside") + os.symlink(secret, os.path.join(base, "clip.mp4")) + inside = os.path.join(base, "real.mp4") + with open(inside, "w") as f: + f.write("inside") + + mocked = {"folder_paths": MagicMock(get_directory_by_type=MagicMock(return_value=base))} + # No isfile patching — this test runs against the real filesystem. + with patch.dict("sys.modules", mocked): + import comfy_execution.media_enrichment as mod + escaped = mod.resolve_output_entry_path( + {"filename": "clip.mp4", "subfolder": "", "type": "output"}) + contained = mod.resolve_output_entry_path( + {"filename": "real.mp4", "subfolder": "", "type": "output"}) + self.assertIsNone(escaped, "symlink pointing outside the base must be rejected") + self.assertEqual(contained, os.path.realpath(inside)) + finally: + shutil.rmtree(root, ignore_errors=True) + def test_extractor_error_does_not_block_sibling_entries(self): call_count = [0]