From 85e76f42d09510d1b50cd460bf5307a910308af4 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Tue, 27 Oct 2026 00:38:32 -0700 Subject: [PATCH] fix(diagnostics): never anchor a report path above the home folder; serve only regular log files Review findings on #329 PR 1: - report_path never uses a filesystem root or a strict ancestor of the home folder (/home, C:\Users) as its anchor, so a path inside home needs an anchor at or below it and otherwise reads "outside ComfyUI"; no path starts with a user name. The scrub matches the home folder only as whole path parts (in full, without its root, and its name). - A download serves only a regular file (a FIFO, socket or device is a 404) and opens it non-blocking. - One unreadable timestamp no longer breaks the log listing, stats, download, review or Clear. - The report tests read a temporary log folder, never the process's own. --- py/api/diagnostics_report.py | 73 +++++++++++++++++------ py/api/log_cleanup.py | 6 +- py/api/logging_routes.py | 2 + tests/test_api_diagnostics_report.py | 87 +++++++++++++++++++++++++++- tests/test_api_logging.py | 8 +++ tests/test_utils_logging_config.py | 15 +++++ utils/logging_config.py | 9 ++- 7 files changed, 175 insertions(+), 25 deletions(-) diff --git a/py/api/diagnostics_report.py b/py/api/diagnostics_report.py index b642895..7e70fa4 100644 --- a/py/api/diagnostics_report.py +++ b/py/api/diagnostics_report.py @@ -28,7 +28,7 @@ import os import platform import posixpath import re -from pathlib import Path +from pathlib import Path, PurePosixPath, PureWindowsPath from aiohttp import web @@ -52,6 +52,36 @@ MIN_KEY_LENGTH = 8 _WINDOWS_ABSOLUTE = re.compile(r"^[A-Za-z]:[\\/]") +def anchored_path(canonical, anchors, home, pathmod=os.path): + """``canonical`` relative to the closest usable anchor, else ``outside ComfyUI``. + + All three are already canonical (``normcase``) in ``pathmod``'s spelling. + A filesystem root is never usable, nor is a strict ancestor of ``home`` + (``/home``, ``C:\\Users``), whose relative paths would start with a user + name; so a path inside the home folder needs an anchor at or below it. + """ + flavor = PureWindowsPath if pathmod is ntpath else PurePosixPath + target = flavor(canonical) + home_path = flavor(home) if home else None + usable = [] + for anchor in map(flavor, anchors): + if anchor.parent == anchor: + continue + if home_path is not None and home_path != anchor: + if home_path.is_relative_to(anchor): + continue + if target.is_relative_to(anchor): + usable.append(anchor) + if not usable: + return OUTSIDE + anchor = max(usable, key=lambda a: len(a.parts)) + return target.relative_to(anchor).as_posix() or "." + + +def _canonical(path): + return os.path.normcase(os.path.realpath(str(path))) + + def report_path(path): """``path`` relative to a public anchor (POSIX style), else ``outside ComfyUI``.""" if not path: @@ -59,17 +89,13 @@ def report_path(path): from ..config import GalleryConfig try: - canonical = Path(os.path.normcase(os.path.realpath(str(path)))) + canonical = _canonical(path) + anchors = [_canonical(anchor) for anchor in GalleryConfig.path_anchors()] + home = os.path.expanduser("~") + home = _canonical(home) if home and home != "~" else None except (OSError, ValueError): return OUTSIDE - # The closest anchor wins; a filesystem root (an output folder directly - # under / or D:\\) never does, or every path would read as public. - anchors = [Path(anchor) for anchor in GalleryConfig.path_anchors()] - anchors = [a for a in anchors if a.parent != a and canonical.is_relative_to(a)] - if not anchors: - return OUTSIDE - anchor = max(anchors, key=lambda a: len(a.parts)) - return canonical.relative_to(anchor).as_posix() or "." + return anchored_path(canonical, anchors, home) def _looks_absolute(text): @@ -81,25 +107,34 @@ def _looks_absolute(text): ) -def _home_spellings(): - """The home folder as it may be spelled in a string (never ``/`` itself).""" +# The home folder is matched only as whole path parts, never inside a word. +_PART = r"(?= MIN_HOME_NAME: + spellings.add(name) + return tuple( + re.compile(_PART.format(re.escape(spelling)), re.IGNORECASE) + for spelling in spellings + ) def scrub(report, key): """``report`` with leaked paths and the key replaced, counted in ``redacted``.""" key = key if isinstance(key, str) and len(key) >= MIN_KEY_LENGTH else None - homes = _home_spellings() + homes = _home_patterns() count = 0 def clean(value): @@ -108,7 +143,7 @@ def scrub(report, key): if ( _looks_absolute(value) or (key and key in value) - or any(home in value for home in homes) + or any(home.search(value) for home in homes) ): count += 1 return REMOVED diff --git a/py/api/log_cleanup.py b/py/api/log_cleanup.py index 83eb5ea..f624838 100644 --- a/py/api/log_cleanup.py +++ b/py/api/log_cleanup.py @@ -57,7 +57,11 @@ def size_label(size): def time_label(mtime_ns): """The local modification time as ``YYYY-MM-DD HH:MM``.""" - return datetime.datetime.fromtimestamp(mtime_ns / 1e9).strftime("%Y-%m-%d %H:%M") + try: + moment = datetime.datetime.fromtimestamp(mtime_ns / 1e9) + except (OSError, OverflowError, ValueError): # a time before 1970 on Windows + return "unknown time" + return moment.strftime("%Y-%m-%d %H:%M") def _count(count, singular, plural): diff --git a/py/api/logging_routes.py b/py/api/logging_routes.py index 40e3780..8918986 100644 --- a/py/api/logging_routes.py +++ b/py/api/logging_routes.py @@ -102,6 +102,8 @@ def _read_log_bytes(path): before this returns. """ flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0) + # A FIFO swapped in after the listing must not block the executor thread. + flags |= getattr(os, "O_NONBLOCK", 0) try: fd = os.open(str(path), flags) except OSError as error: diff --git a/tests/test_api_diagnostics_report.py b/tests/test_api_diagnostics_report.py index d5ba20a..41fdb90 100644 --- a/tests/test_api_diagnostics_report.py +++ b/tests/test_api_diagnostics_report.py @@ -34,9 +34,9 @@ from py.config import ( ) # noqa: E402 try: - from tests.test_api_admin import AdminAPITestCase # noqa: E402 + from tests.test_api_log_cleanup import LogCleanupTestCase # noqa: E402 except ImportError: # discovered with tests/ as the top-level directory - from test_api_admin import AdminAPITestCase # noqa: E402 + from test_api_log_cleanup import LogCleanupTestCase # noqa: E402 REPORT = "/prompt_manager/system/diagnostics" KEY = "civitai-secret-0123456789abcdef" @@ -102,7 +102,9 @@ def _strings(value): yield from _strings(item) -class DiagnosticsReportTestCase(AdminAPITestCase): +class DiagnosticsReportTestCase(LogCleanupTestCase): + """The admin app with its logger manager on a temporary log folder.""" + async def get_application(self): app = await super().get_application() saved = IntegrationConfig.get_config() @@ -229,6 +231,85 @@ class TestReport(DiagnosticsReportTestCase): cleaned = scrub({"a": f"{relative}/area/prompts.db", "redacted": 0}, None) self.assertEqual(cleaned, {"a": "[removed]", "redacted": 1}) + def test_the_home_folder_is_matched_as_a_whole_path_part(self): + with mock.patch.object( + diagnostics_report.os.path, "expanduser", return_value="/root" + ): + cleaned = scrub( + { + "a": "output/rootfiles", + "b": "output/root/x", + "c": "see /root/x", + "redacted": 0, + }, + None, + ) + self.assertEqual(cleaned["a"], "output/rootfiles") + # The home folder's name is a whole path part here: removed. + self.assertEqual(cleaned["b"], "[removed]") + self.assertEqual(cleaned["c"], "[removed]") + self.assertEqual(cleaned["redacted"], 2) + + async def test_report_reads_only_the_temporary_log_folder(self): + self.write("prompt_manager.log.1", b"x" * 1234) + report, _text = await self.report() + stats = report["logging"]["stats"] + self.assertEqual(stats["log_files_count"], 2) + self.assertEqual( + stats["total_log_size"], + 1234 + os.path.getsize(self.path("prompt_manager.log")), + ) + + def test_an_ancestor_of_home_is_never_an_anchor_on_linux(self): + anchored = diagnostics_report.anchored_path + home = "/home/someone" + anchors = ["/home", "/opt/ComfyUI"] + self.assertEqual( + anchored("/home/someone/renders/a.png", anchors, home, posixpath), OUTSIDE + ) + self.assertEqual( + anchored("/home/other/renders", anchors, home, posixpath), OUTSIDE + ) + self.assertEqual( + anchored("/opt/ComfyUI/output", anchors, home, posixpath), "output" + ) + + def test_an_ancestor_of_home_is_never_an_anchor_on_windows(self): + anchored = diagnostics_report.anchored_path + home = ntpath.normcase("C:\\Users\\someone") + anchors = [ntpath.normcase(a) for a in ("C:\\Users", "C:\\", "D:\\ComfyUI")] + path = ntpath.normcase("C:\\Users\\someone\\renders\\a.png") + self.assertEqual(anchored(path, anchors, home, ntpath), OUTSIDE) + inside = ntpath.normcase("D:\\ComfyUI\\output\\x") + self.assertEqual(anchored(inside, anchors, home, ntpath), "output/x") + + def test_an_anchor_that_is_home_names_no_user(self): + anchored = diagnostics_report.anchored_path + home = "/home/someone" + self.assertEqual( + anchored("/home/someone/ComfyUI/output", [home], home, posixpath), + "ComfyUI/output", + ) + self.assertEqual( + anchored( + "/home/someone/ComfyUI/output", + [home, "/home/someone/ComfyUI"], + home, + posixpath, + ), + "output", + ) + + def test_scrub_removes_the_home_folder_name(self): + with mock.patch.object( + diagnostics_report.os.path, "expanduser", return_value="/home/someone" + ): + cleaned = scrub( + {"a": "someone/renders", "b": "someones", "redacted": 0}, None + ) + self.assertEqual(cleaned["a"], "[removed]") + self.assertEqual(cleaned["b"], "someones") + def test_scrub_removes_a_leaked_path_and_counts_it(self): leaked = { "a": "/home/someone/x", diff --git a/tests/test_api_logging.py b/tests/test_api_logging.py index b98d9a7..8e0e825 100644 --- a/tests/test_api_logging.py +++ b/tests/test_api_logging.py @@ -464,6 +464,14 @@ class TestLogFilesAndDownload(LoggingAPITestCase): self.assertEqual(resp.status, 404) self.assertNotIn("swapped in", await resp.text()) + @unittest.skipUnless(hasattr(os, "mkfifo"), "FIFOs are POSIX only") + async def test_download_refuses_a_fifo_without_blocking(self): + os.mkfifo(self.log_dir / "pipe.log") + resp = await self.client.request( + "GET", "/prompt_manager/logs/download/pipe.log" + ) + self.assertEqual(resp.status, 404) + async def test_download_reads_off_the_loop(self): (self.log_dir / "prompt_manager.log").write_bytes(b"line\n") seen = [] diff --git a/tests/test_utils_logging_config.py b/tests/test_utils_logging_config.py index ed7d765..3b07911 100644 --- a/tests/test_utils_logging_config.py +++ b/tests/test_utils_logging_config.py @@ -712,6 +712,21 @@ class TestRotatedCopyRule(IsolatedLoggerTestCase): set(row), {"filename", "path", "size", "modified", "is_main", "kind"} ) + def test_an_unreadable_timestamp_does_not_break_the_scan(self): + self._write("prompt_manager.log.1") + real = logging_config.datetime + + class Clock: + @staticmethod + def fromtimestamp(value): + raise OSError(22, "Invalid argument") + + with mock.patch.object(logging_config, "datetime", Clock): + rows = {r["name"]: r for r in self.manager.scan_log_folder()} + self.assertIs(logging_config.datetime, real) + self.assertEqual(rows["prompt_manager.log.1"]["kind"], "rotated") + self.assertEqual(rows["prompt_manager.log.1"]["modified"], "") + def test_stats_exclude_links(self): self._write("prompt_manager.log.1", b"r" * 10) before = self.manager.get_log_stats() diff --git a/utils/logging_config.py b/utils/logging_config.py index 43931f7..97a3763 100644 --- a/utils/logging_config.py +++ b/utils/logging_config.py @@ -99,6 +99,10 @@ def scan_log_folder(log_dir) -> List[Dict[str, Any]]: kind = _kind(name, path, st) except OSError: continue + try: + modified = datetime.fromtimestamp(st.st_mtime).isoformat() + except (OSError, OverflowError, ValueError): # a time before 1970 on Windows + modified = "" rows.append( { "name": name, @@ -106,8 +110,9 @@ def scan_log_folder(log_dir) -> List[Dict[str, Any]]: "size": st.st_size, "mtime_ns": st.st_mtime_ns, "ino": st.st_ino, - "modified": datetime.fromtimestamp(st.st_mtime).isoformat(), + "modified": modified, "_dir": stat.S_ISDIR(st.st_mode) and kind != "link", + "_regular": stat.S_ISREG(st.st_mode) and kind != "link", } ) return rows @@ -135,7 +140,7 @@ def log_file_entry(log_dir, name) -> Path: raise FileNotFoundError("Log file not found") for row in scan_log_folder(log_dir): if row["name"] == name and row["kind"] in ("active", "rotated", "other"): - if row["_dir"]: + if not row["_regular"]: # a folder, FIFO, socket or device break entry = LogFilePath(log_dir, row["name"]) entry.listed_ino = row["ino"]