fix(execution): harden output-entry media enrichment per review

- Resolve symlinks (realpath on base and candidate) before the
  containment check, so a symlink planted inside output/ can't smuggle
  an outside target past commonpath; also covers output dirs that are
  themselves symlinks. Applies to both enrichers via the shared helper.
- Validate filename/subfolder/type are strings before path resolution
  so a malformed entry skips cleanly instead of raising TypeError.
- Broaden the lazy-import guard beyond ImportError: the enrichment is
  called unguarded on the output path and must degrade to a no-op when
  a dependency fails to import for any reason.
- Tests: real-filesystem symlink escape/containment, non-string fields,
  non-ImportError import failure.
This commit is contained in:
Simon Pinfold
2026-08-07 21:05:46 -07:00
parent 6bc642537f
commit 51a38a6379
2 changed files with 81 additions and 12 deletions

View File

@@ -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 = {}

View File

@@ -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]