From b0e60ec4b88e3e94509f81f9a1ccf0b8f4ea517c Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Wed, 7 Oct 2026 18:44:16 -0700 Subject: [PATCH] fix: no absolute server paths in API responses Settings, diagnostics, duplicate/output scans and the logging endpoints returned absolute filesystem paths, and several error envelopes echoed str(e) which for OSError carries the full path. - _public_path() renders a path relative to the ComfyUI base directory (folder_paths.base_path, else the output dir's parent) or to the parent of an opted-in extra gallery root, falling back to the basename; _public_error() reduces OSError to strerror + basename. - GET /settings returns gallery roots and monitored directories in that form; POST /settings resolves relative roots against the same anchors (GalleryConfig.resolve_gallery_root) so the GET output round-trips. - Diagnostics output_dirs, monitored_directories and the missing-DB message no longer carry absolute paths; the image monitor is reached through the package-or-top-level import fallback instead of a three-dot relative import that failed outside the package. - Duplicate scan 'path' is now relative to the output root and the delete endpoint accepts relative paths. - /logs/files 'path' and /logs/stats 'log_directory' use the public form; unexpected errors answer with a generic message. --- py/api/__init__.py | 34 ++++++++++ py/api/admin.py | 114 +++++++++++++++++++------------- py/api/logging_routes.py | 30 ++++++--- py/config.py | 76 ++++++++++++++++++++- tests/test_api_admin.py | 127 ++++++++++++++++++++++++++++++++++++ tests/test_api_core.py | 122 ++++++++++++++++++++++++++++++++++ tests/test_api_logging.py | 38 +++++++++++ tests/test_gallery_roots.py | 14 +++- 8 files changed, 494 insertions(+), 61 deletions(-) create mode 100644 tests/test_api_core.py diff --git a/py/api/__init__.py b/py/api/__init__.py index c0c424dc..9606d960 100644 --- a/py/api/__init__.py +++ b/py/api/__init__.py @@ -44,6 +44,37 @@ def _get_project_root(): return os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +def _public_path(path) -> str: + """Render a server path for an API response without revealing the layout. + + Returns the path relative to the ComfyUI base directory (or to the parent + of an opted-in extra gallery root), as a POSIX-style string, or just the + basename when the path is not under any of those anchors. + """ + if not path: + return "" + from ..config import GalleryConfig + + try: + canonical = Path(os.path.normcase(os.path.realpath(str(path)))) + except (OSError, ValueError): + return os.path.basename(str(path)) + for anchor in GalleryConfig.path_anchors(): + if canonical.is_relative_to(Path(anchor)): + return canonical.relative_to(Path(anchor)).as_posix() or "." + return canonical.name + + +def _public_error(exc: BaseException) -> str: + """Error text safe for API responses: OSError paths reduced to basenames.""" + if isinstance(exc, OSError): + parts = [exc.strerror or type(exc).__name__] + if exc.filename: + parts.append(os.path.basename(str(exc.filename))) + return ": ".join(parts) + return str(exc) or type(exc).__name__ + + # ── Gzip compression middleware ──────────────────────────────────────── _GZIP_MIN_SIZE = 1024 # Only compress bodies larger than 1 KB _GZIP_TYPES = frozenset( @@ -140,6 +171,9 @@ class PromptManagerAPI( self.logger.info("PromptManager API initialization completed") + _public_path = staticmethod(_public_path) + _public_error = staticmethod(_public_error) + async def _run_in_executor(self, func, *args, **kwargs): """Run a blocking function in the default thread pool executor. diff --git a/py/api/admin.py b/py/api/admin.py index 12e68e6b..122dde21 100644 --- a/py/api/admin.py +++ b/py/api/admin.py @@ -24,6 +24,20 @@ except ImportError: from database.operations import PromptDatabase +def _image_monitor_module(): + """The utils.image_monitor module under either package identity.""" + try: + from ...utils import image_monitor + except ImportError: + from utils import image_monitor + return image_monitor + + +def _current_image_monitor(): + """The running ImageMonitor singleton, or None when not started.""" + return getattr(_image_monitor_module(), "_monitor_instance", None) + + class AdminRoutesMixin: """Mixin providing admin, diagnostics, and maintenance API endpoints.""" @@ -90,12 +104,9 @@ class AdminRoutesMixin: ) except Exception as e: - self.logger.error(f"Scan duplicates error: {e}") + self.logger.error(f"Scan duplicates error: {e}", exc_info=True) return web.json_response( - { - "success": False, - "error": f"Failed to scan duplicate images: {str(e)}", - }, + {"success": False, "error": "Failed to scan duplicate images"}, status=500, ) @@ -113,9 +124,9 @@ class AdminRoutesMixin: ) except Exception as e: - self.logger.error(f"Cleanup error: {e}") + self.logger.error(f"Cleanup error: {e}", exc_info=True) return web.json_response( - {"success": False, "error": f"Failed to cleanup duplicates: {str(e)}"}, + {"success": False, "error": "Failed to cleanup duplicates"}, status=500, ) @@ -189,7 +200,7 @@ class AdminRoutesMixin: image_info = { "id": str(hash(str(media_path))), "filename": media_path.name, - "path": str(media_path), + "path": str(rel_path), "relative_path": str(rel_path), "url": f"/prompt_manager/images/serve/{rel_path.as_posix()}", "thumbnail_url": thumbnail_url, @@ -266,6 +277,8 @@ class AdminRoutesMixin: output_path = Path(output_dir) file_path = Path(image_path) + if not file_path.is_absolute(): + file_path = output_path / file_path # Security check - ensure file is within output directory try: @@ -307,7 +320,7 @@ class AdminRoutesMixin: except Exception as e: self.logger.error(f"Error deleting file {image_path}: {e}") - failed_files.append(f"{image_path} ({str(e)})") + failed_files.append(f"{image_path} ({self._public_error(e)})") failed_count += 1 response_data = { @@ -324,12 +337,9 @@ class AdminRoutesMixin: return web.json_response(response_data) except Exception as e: - self.logger.error(f"Delete duplicate images error: {e}") + self.logger.error(f"Delete duplicate images error: {e}", exc_info=True) return web.json_response( - { - "success": False, - "error": f"Failed to delete duplicate images: {str(e)}", - }, + {"success": False, "error": "Failed to delete duplicate images"}, status=500, ) @@ -341,9 +351,9 @@ class AdminRoutesMixin: return web.json_response({"success": True, "stats": stats}) except Exception as e: - self.logger.error(f"Stats error: {e}") + self.logger.error(f"Stats error: {e}", exc_info=True) return web.json_response( - {"success": False, "error": f"Failed to get statistics: {str(e)}"}, + {"success": False, "error": "Failed to get statistics"}, status=500, ) @@ -375,27 +385,27 @@ class AdminRoutesMixin: if GalleryConfig.MONITORING_DIRECTORIES: monitored_dirs = GalleryConfig.MONITORING_DIRECTORIES + root_paths = [ + self._public_path(d) for d in GalleryConfig.MONITORING_DIRECTORIES + ] return web.json_response( { "success": True, "settings": { "result_timeout": PromptManagerConfig.RESULT_TIMEOUT, "webui_display_mode": PromptManagerConfig.WEBUI_DISPLAY_MODE, - "gallery_root_paths": list( - GalleryConfig.MONITORING_DIRECTORIES - ), - "gallery_root_path": ( - GalleryConfig.MONITORING_DIRECTORIES[0] - if GalleryConfig.MONITORING_DIRECTORIES - else "" - ), - "monitored_directories": monitored_dirs, + "gallery_root_paths": root_paths, + "gallery_root_path": root_paths[0] if root_paths else "", + "monitored_directories": [ + self._public_path(d) for d in monitored_dirs + ], }, } ) except Exception as e: + self.logger.error(f"Get settings error: {e}", exc_info=True) return web.json_response( - {"success": False, "error": f"Failed to get settings: {str(e)}"}, + {"success": False, "error": "Failed to get settings"}, status=500, ) @@ -477,7 +487,7 @@ class AdminRoutesMixin: None, f"Invalid gallery root '{os.path.basename(entry)}': {reason}", ) - roots.append(entry) + roots.append(GalleryConfig.resolve_gallery_root(entry)) return roots, None def _apply_gallery_roots(self, roots): @@ -548,12 +558,14 @@ class AdminRoutesMixin: else: results["database"] = { "status": "error", - "message": f"Database file not found: {db_path}", + "message": ( + f"Database file not found: {os.path.basename(db_path)}" + ), } except Exception as e: results["database"] = { "status": "error", - "message": f"Database error: {str(e)}", + "message": f"Database error: {self._public_error(e)}", } # Check dependencies @@ -616,16 +628,18 @@ class AdminRoutesMixin: results["comfyui_output"] = { "status": "ok" if output_dirs else "warning", - "output_dirs": output_dirs, + "output_dirs": [self._public_path(d) for d in output_dirs], } # Check image monitor status try: - from ...utils import image_monitor as im_mod - - monitor = im_mod._monitor_instance + monitor = _current_image_monitor() if monitor is not None: - monitor_status = monitor.get_status() + monitor_status = dict(monitor.get_status()) + monitor_status["monitored_directories"] = [ + self._public_path(d) + for d in monitor_status.get("monitored_directories", []) + ] results["image_monitor"] = { "status": ( "ok" if monitor_status.get("observer_alive") else "error" @@ -640,14 +654,16 @@ class AdminRoutesMixin: except Exception as e: results["image_monitor"] = { "status": "error", - "message": f"Failed to get monitor status: {str(e)}", + "message": f"Failed to get monitor status: {self._public_error(e)}", } return web.json_response({"success": True, "diagnostics": results}) except Exception as e: self.logger.error(f"Diagnostics error: {e}", exc_info=True) - return web.json_response({"success": False, "error": str(e)}, status=500) + return web.json_response( + {"success": False, "error": "Diagnostics failed"}, status=500 + ) async def test_image_link(self, request): """Test creating an image link.""" @@ -695,14 +711,18 @@ class AdminRoutesMixin: "success": False, "result": { "status": "error", - "message": f"Failed to create test link: {str(e)}", + "message": ( + f"Failed to create test link: {self._public_error(e)}" + ), }, } ) except Exception as e: self.logger.error(f"Test link error: {e}", exc_info=True) - return web.json_response({"success": False, "error": str(e)}, status=500) + return web.json_response( + {"success": False, "error": "Test link failed"}, status=500 + ) async def run_maintenance(self, request): """Perform comprehensive database maintenance and optimization.""" @@ -731,7 +751,7 @@ class AdminRoutesMixin: except Exception as e: results["cleanup_duplicates"] = { "success": False, - "error": str(e), + "error": self._public_error(e), "message": "Failed to cleanup duplicates", } @@ -745,7 +765,7 @@ class AdminRoutesMixin: except Exception as e: results["vacuum"] = { "success": False, - "error": str(e), + "error": self._public_error(e), "message": "Failed to vacuum database", } @@ -760,7 +780,7 @@ class AdminRoutesMixin: except Exception as e: results["cleanup_orphaned_images"] = { "success": False, - "error": str(e), + "error": self._public_error(e), "message": "Failed to cleanup orphaned images", } @@ -775,7 +795,7 @@ class AdminRoutesMixin: except Exception as e: results["check_hash_duplicates"] = { "success": False, - "error": str(e), + "error": self._public_error(e), "message": "Failed to check hash duplicates", } @@ -790,7 +810,7 @@ class AdminRoutesMixin: except Exception as e: results["statistics"] = { "success": False, - "error": str(e), + "error": self._public_error(e), "message": "Failed to get database statistics", } @@ -805,7 +825,7 @@ class AdminRoutesMixin: except Exception as e: results["prune_orphaned_prompts"] = { "success": False, - "error": str(e), + "error": self._public_error(e), "message": "Failed to prune orphaned prompts", } @@ -821,7 +841,7 @@ class AdminRoutesMixin: except Exception as e: results["check_consistency"] = { "success": False, - "error": str(e), + "error": self._public_error(e), "message": "Failed to check database consistency", } @@ -844,7 +864,7 @@ class AdminRoutesMixin: except Exception as e: self.logger.error(f"Maintenance error: {e}", exc_info=True) return web.json_response( - {"success": False, "error": f"Maintenance failed: {str(e)}"}, status=500 + {"success": False, "error": "Maintenance failed"}, status=500 ) async def backup_database(self, request): @@ -1020,7 +1040,7 @@ class AdminRoutesMixin: yield f"data: {json.dumps({'type': 'error', 'message': 'No output directories found. Configure scan directories in Settings.'})}\n\n" return - dir_names = [str(d) for d in output_dirs] + dir_names = [self._public_path(d) for d in output_dirs] yield f"data: {json.dumps({'type': 'progress', 'progress': 0, 'status': f'Scanning {len(output_dirs)} directory(ies) for media files...', 'processed': 0, 'found': 0})}\n\n" media_files = await self._run_in_executor( diff --git a/py/api/logging_routes.py b/py/api/logging_routes.py index 9c052987..70796caa 100644 --- a/py/api/logging_routes.py +++ b/py/api/logging_routes.py @@ -131,14 +131,18 @@ class LoggingRoutesMixin: except Exception as e: self.logger.error(f"Get logs error: {e}") return web.json_response( - {"success": False, "error": str(e), "logs": []}, status=500 + {"success": False, "error": self._public_error(e), "logs": []}, + status=500, ) async def get_log_files(self, request): """Get information about available log files.""" try: logger_manager = self._get_logger_manager() - log_files = logger_manager.get_log_files() + log_files = [ + {**entry, "path": self._public_path(entry.get("path"))} + for entry in logger_manager.get_log_files() + ] return web.json_response( {"success": True, "files": log_files, "count": len(log_files)} @@ -147,7 +151,8 @@ class LoggingRoutesMixin: except Exception as e: self.logger.error(f"Get log files error: {e}") return web.json_response( - {"success": False, "error": str(e), "files": []}, status=500 + {"success": False, "error": self._public_error(e), "files": []}, + status=500, ) async def download_log_file(self, request): @@ -182,7 +187,9 @@ class LoggingRoutesMixin: except Exception as e: self.logger.error(f"Download log file error: {e}") - return web.json_response({"success": False, "error": str(e)}, status=500) + return web.json_response( + {"success": False, "error": self._public_error(e)}, status=500 + ) async def truncate_logs(self, request): """Truncate all log files.""" @@ -200,7 +207,9 @@ class LoggingRoutesMixin: except Exception as e: self.logger.error(f"Truncate logs error: {e}") - return web.json_response({"success": False, "error": str(e)}, status=500) + return web.json_response( + {"success": False, "error": self._public_error(e)}, status=500 + ) async def get_log_config(self, request): """Get current logging configuration.""" @@ -212,7 +221,9 @@ class LoggingRoutesMixin: except Exception as e: self.logger.error(f"Get log config error: {e}") - return web.json_response({"success": False, "error": str(e)}, status=500) + return web.json_response( + {"success": False, "error": self._public_error(e)}, status=500 + ) async def update_log_config(self, request): """Update logging configuration (whitelisted keys only).""" @@ -251,10 +262,13 @@ class LoggingRoutesMixin: """Get logging statistics.""" try: logger_manager = self._get_logger_manager() - stats = logger_manager.get_log_stats() + stats = dict(logger_manager.get_log_stats()) + stats["log_directory"] = self._public_path(stats.get("log_directory")) return web.json_response({"success": True, "stats": stats}) except Exception as e: self.logger.error(f"Get log stats error: {e}") - return web.json_response({"success": False, "error": str(e)}, status=500) + return web.json_response( + {"success": False, "error": self._public_error(e)}, status=500 + ) diff --git a/py/config.py b/py/config.py index 00e6e3f2..6d49e06c 100644 --- a/py/config.py +++ b/py/config.py @@ -73,11 +73,39 @@ def _is_filesystem_root(path: str) -> bool: return os.path.dirname(path) == path -def _comfyui_directories() -> List[str]: - """Directories reported by ComfyUI's folder_paths module, if importable.""" +def _import_folder_paths(): + """ComfyUI's folder_paths module, or None outside ComfyUI.""" try: import folder_paths except ImportError: + return None + return folder_paths + + +def _comfyui_base_directory() -> Optional[str]: + """ComfyUI's base directory: folder_paths.base_path, else the output dir's parent.""" + folder_paths = _import_folder_paths() + if folder_paths is None: + return None + base = getattr(folder_paths, "base_path", None) + if isinstance(base, str) and base: + return base + getter = getattr(folder_paths, "get_output_directory", None) + if getter is None: + return None + try: + output_dir = getter() + except Exception: + return None + if isinstance(output_dir, str) and output_dir: + return os.path.dirname(os.path.realpath(output_dir)) + return None + + +def _comfyui_directories() -> List[str]: + """Directories reported by ComfyUI's folder_paths module, if importable.""" + folder_paths = _import_folder_paths() + if folder_paths is None: return [] found = [] @@ -272,6 +300,48 @@ class GalleryConfig: parents.append(canonical) return parents + @classmethod + def path_anchors(cls) -> List[str]: + """Canonical directories that relative gallery paths are resolved against. + + The ComfyUI base directory comes first, followed by the parent of each + directory listed in PROMPT_MANAGER_EXTRA_GALLERY_ROOTS, so an extra + root is addressed by its own name (``gallery/sub``) rather than by an + absolute path. The same anchors drive the public (relative) form of + paths in API responses. + """ + anchors = [] + base = _comfyui_base_directory() + candidates = [base] if base else [] + candidates.extend(os.path.dirname(root) for root in _extra_gallery_roots()) + for candidate in candidates: + try: + canonical = _canonical_path(candidate) + except (OSError, ValueError): + continue + if canonical not in anchors: + anchors.append(canonical) + return anchors + + @classmethod + def resolve_gallery_root(cls, path: str) -> str: + """Canonical absolute path for a gallery root given in any accepted form. + + Absolute paths are canonicalised as-is. Relative paths are tried + against each :meth:`path_anchors` entry and the first existing match + wins; otherwise the first anchor (or the current directory when there + is none) is used, so the caller's existence check reports it. + """ + path = path.strip() + if os.path.isabs(path): + return _canonical_path(path) + anchors = cls.path_anchors() + for anchor in anchors: + candidate = os.path.join(anchor, path) + if os.path.exists(candidate): + return _canonical_path(candidate) + return _canonical_path(os.path.join(anchors[0], path) if anchors else path) + @classmethod def validate_gallery_root(cls, path: Any) -> Tuple[bool, str]: """Check whether ``path`` may be used as a gallery root. @@ -284,7 +354,7 @@ class GalleryConfig: return False, "Gallery root must be a non-empty path" try: - canonical = _canonical_path(path.strip()) + canonical = cls.resolve_gallery_root(path) except (OSError, ValueError): return False, "Gallery root could not be resolved" diff --git a/tests/test_api_admin.py b/tests/test_api_admin.py index bcf74ba1..9534063a 100644 --- a/tests/test_api_admin.py +++ b/tests/test_api_admin.py @@ -221,5 +221,132 @@ class TestSaveSettingsGalleryRoots(AdminAPITestCase): self.assertEqual(resp.status, 400) +class FakeMonitor: + def __init__(self, dirs): + self.monitored_directories = list(dirs) + + def get_status(self): + return { + "running": True, + "monitored_directories": self.monitored_directories, + "handler_active": True, + "observer_alive": True, + } + + +class TestNoAbsolutePathsInResponses(AdminAPITestCase): + """No admin response may reveal the server's absolute filesystem layout.""" + + def _install_monitor(self, dirs): + import utils.image_monitor as im_mod + + orig = im_mod._monitor_instance + im_mod._monitor_instance = FakeMonitor(dirs) + self.addCleanup(setattr, im_mod, "_monitor_instance", orig) + + async def _body(self, method, path, **kwargs): + resp = await self.client.request(method, path, **kwargs) + return resp, await resp.text() + + async def test_settings_show_roots_relative_to_comfyui(self): + renders = self.output_dir / "renders" + renders.mkdir() + GalleryConfig.MONITORING_DIRECTORIES = [str(renders)] + self._install_monitor([str(renders)]) + + resp, body = await self._body("GET", "/prompt_manager/settings") + + self.assertEqual(resp.status, 200) + self.assertNotIn(self.tmpdir, body) + data = json.loads(body) + self.assertEqual(data["settings"]["gallery_root_paths"], ["output/renders"]) + self.assertEqual(data["settings"]["gallery_root_path"], "output/renders") + self.assertEqual(data["settings"]["monitored_directories"], ["output/renders"]) + + async def test_settings_relative_root_round_trips(self): + """What GET returns can be POSTed back unchanged from any CWD.""" + renders = self.output_dir / "renders" + renders.mkdir() + + resp = await self.client.request( + "POST", + "/prompt_manager/settings", + json={"gallery_root_paths": ["output/renders"]}, + ) + + self.assertEqual(resp.status, 200) + self.assertEqual( + [os.path.normcase(p) for p in GalleryConfig.MONITORING_DIRECTORIES], + [os.path.normcase(os.path.realpath(renders))], + ) + + async def test_diagnostics_paths_are_relative(self): + GalleryConfig.MONITORING_DIRECTORIES = [str(self.output_dir)] + self._install_monitor([str(self.output_dir)]) + + resp, body = await self._body("GET", "/prompt_manager/diagnostics") + + self.assertEqual(resp.status, 200) + self.assertNotIn(self.tmpdir, body) + data = json.loads(body) + self.assertEqual( + data["diagnostics"]["comfyui_output"]["output_dirs"], ["output"] + ) + self.assertEqual( + data["diagnostics"]["image_monitor"]["monitored_directories"], ["output"] + ) + + async def test_diagnostics_missing_db_message_has_no_path(self): + self.api.db.model.db_path = os.path.join(self.tmpdir, "missing", "prompts.db") + + resp, body = await self._body("GET", "/prompt_manager/diagnostics") + + self.assertNotIn(self.tmpdir, body) + data = json.loads(body) + self.assertEqual(data["diagnostics"]["database"]["status"], "error") + self.assertIn("prompts.db", data["diagnostics"]["database"]["message"]) + + async def test_scan_duplicates_paths_are_relative_to_output(self): + (self.output_dir / "a.png").write_bytes(b"same-bytes") + (self.output_dir / "sub").mkdir() + (self.output_dir / "sub" / "b.png").write_bytes(b"same-bytes") + GalleryConfig.MONITORING_DIRECTORIES = [str(self.output_dir)] + + resp, body = await self._body("GET", "/prompt_manager/scan_duplicates") + + self.assertEqual(resp.status, 200) + self.assertNotIn(self.tmpdir, body) + data = json.loads(body) + self.assertEqual(len(data["duplicates"]), 1) + paths = sorted(img["path"] for img in data["duplicates"][0]["images"]) + self.assertEqual(paths, ["a.png", os.path.join("sub", "b.png")]) + + async def test_delete_duplicates_accepts_relative_paths(self): + target = self.output_dir / "sub" / "b.png" + target.parent.mkdir() + target.write_bytes(b"bytes") + GalleryConfig.MONITORING_DIRECTORIES = [str(self.output_dir)] + + resp, body = await self._body( + "POST", + "/prompt_manager/delete_duplicate_images", + json={"image_paths": [os.path.join("sub", "b.png")]}, + ) + + self.assertEqual(resp.status, 200) + data = json.loads(body) + self.assertEqual(data["deleted_count"], 1) + self.assertFalse(target.exists()) + + async def test_scan_images_stream_has_no_absolute_paths(self): + GalleryConfig.MONITORING_DIRECTORIES = [str(self.output_dir)] + + resp, body = await self._body("POST", "/prompt_manager/scan") + + self.assertEqual(resp.status, 200) + self.assertNotIn(self.tmpdir, body) + self.assertIn('"directories": ["output"]', body) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_api_core.py b/tests/test_api_core.py new file mode 100644 index 00000000..3736f975 --- /dev/null +++ b/tests/test_api_core.py @@ -0,0 +1,122 @@ +"""Tests for the core API module (py/api/__init__.py): helpers and base routes.""" + +import os +import shutil +import sys +import tempfile +import types +import unittest +from pathlib import Path +from unittest.mock import MagicMock + +_mock_server = MagicMock() +_mock_server.PromptServer.instance.routes = MagicMock() +sys.modules["server"] = _mock_server + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from py.api import _public_error, _public_path # noqa: E402 + +EXTRA_ROOTS_ENV = "PROMPT_MANAGER_EXTRA_GALLERY_ROOTS" + + +class FolderPathsFixture(unittest.TestCase): + """Temp ComfyUI tree exposed through a stub folder_paths module.""" + + def setUp(self): + self.tmpdir = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self.tmpdir, True) + self.comfy_dir = Path(self.tmpdir) / "ComfyUI" + self.output_dir = self.comfy_dir / "output" + self.output_dir.mkdir(parents=True) + + self._orig_folder_paths = sys.modules.get("folder_paths") + self.addCleanup(self._restore_folder_paths) + self._orig_extra = os.environ.pop(EXTRA_ROOTS_ENV, None) + self.addCleanup(self._restore_extra_env) + + self.install_folder_paths(base_path=str(self.comfy_dir)) + + def install_folder_paths(self, **attrs): + attrs.setdefault("get_output_directory", lambda: str(self.output_dir)) + sys.modules["folder_paths"] = types.SimpleNamespace(**attrs) + + def _restore_folder_paths(self): + if self._orig_folder_paths is None: + sys.modules.pop("folder_paths", None) + else: + sys.modules["folder_paths"] = self._orig_folder_paths + + def _restore_extra_env(self): + if self._orig_extra is None: + os.environ.pop(EXTRA_ROOTS_ENV, None) + else: + os.environ[EXTRA_ROOTS_ENV] = self._orig_extra + + +class TestPublicPath(FolderPathsFixture): + + def test_path_under_base_is_relative_posix(self): + target = self.output_dir / "renders" / "a.png" + self.assertEqual(_public_path(str(target)), "output/renders/a.png") + + def test_base_itself_is_dot(self): + self.assertEqual(_public_path(str(self.comfy_dir)), ".") + + def test_path_outside_base_is_basename_only(self): + secret = Path(self.tmpdir) / "secret" / "passwords.txt" + secret.parent.mkdir() + secret.write_text("x") + self.assertEqual(_public_path(str(secret)), "passwords.txt") + self.assertNotIn(self.tmpdir, _public_path(str(secret))) + + def test_path_object_accepted(self): + self.assertEqual(_public_path(self.output_dir), "output") + + def test_falls_back_to_output_parent_without_base_path(self): + self.install_folder_paths() # no base_path attribute + self.assertEqual(_public_path(str(self.output_dir / "x.png")), "output/x.png") + + def test_env_extra_root_keeps_its_own_name(self): + extra = Path(self.tmpdir) / "nas" / "gallery" + (extra / "sub").mkdir(parents=True) + os.environ[EXTRA_ROOTS_ENV] = str(extra) + self.assertEqual(_public_path(str(extra / "sub")), "gallery/sub") + + def test_without_folder_paths_returns_basename(self): + sys.modules.pop("folder_paths", None) + self.assertEqual(_public_path(str(self.output_dir / "img.png")), "img.png") + + def test_empty_or_none_gives_empty_string(self): + self.assertEqual(_public_path(""), "") + self.assertEqual(_public_path(None), "") + + +class TestPublicError(FolderPathsFixture): + + def test_oserror_uses_strerror_and_basename(self): + err = FileNotFoundError( + 2, "No such file or directory", str(self.tmpdir) + "/a/b.png" + ) + text = _public_error(err) + self.assertIn("No such file or directory", text) + self.assertIn("b.png", text) + self.assertNotIn(self.tmpdir, text) + + def test_oserror_without_filename(self): + err = PermissionError(13, "Permission denied") + self.assertEqual(_public_error(err), "Permission denied") + + def test_oserror_without_strerror_uses_type_name(self): + err = OSError() + self.assertEqual(_public_error(err), "OSError") + + def test_non_oserror_is_str(self): + self.assertEqual(_public_error(ValueError("bad value")), "bad value") + + def test_non_oserror_with_empty_message_uses_type_name(self): + self.assertEqual(_public_error(RuntimeError()), "RuntimeError") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_api_logging.py b/tests/test_api_logging.py index 12938bb7..1fb78b75 100644 --- a/tests/test_api_logging.py +++ b/tests/test_api_logging.py @@ -1,5 +1,6 @@ """Route tests for the logging API (py/api/logging_routes.py).""" +import json import os import shutil import sys @@ -192,5 +193,42 @@ class TestUpdateLogConfigWhitelist(LoggingAPITestCase): self.assertEqual(self.manager.update_calls, []) +class TestNoAbsolutePathsInLogResponses(LoggingAPITestCase): + + async def test_log_files_have_no_absolute_path(self): + (self.log_dir / "prompt_manager.log").write_text("hello") + + resp = await self.client.request("GET", "/prompt_manager/logs/files") + + body = await resp.text() + self.assertEqual(resp.status, 200) + self.assertNotIn(self.tmpdir, body) + data = json.loads(body) + self.assertEqual(data["files"][0]["filename"], "prompt_manager.log") + + async def test_log_stats_directory_is_not_absolute(self): + resp = await self.client.request("GET", "/prompt_manager/logs/stats") + + body = await resp.text() + self.assertEqual(resp.status, 200) + self.assertNotIn(self.tmpdir, body) + self.assertEqual(json.loads(body)["stats"]["log_directory"], "logs") + + async def test_error_from_manager_does_not_leak_path(self): + def boom(): + raise FileNotFoundError( + 2, "No such file", os.path.join(self.tmpdir, "x.log") + ) + + self.manager.get_log_stats = boom + + resp = await self.client.request("GET", "/prompt_manager/logs/stats") + + body = await resp.text() + self.assertEqual(resp.status, 500) + self.assertNotIn(self.tmpdir, body) + self.assertIn("x.log", json.loads(body)["error"]) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_gallery_roots.py b/tests/test_gallery_roots.py index 61199f32..4a83ac25 100644 --- a/tests/test_gallery_roots.py +++ b/tests/test_gallery_roots.py @@ -181,14 +181,22 @@ class TestValidateGalleryRootAccepts(GalleryRootTestCase): self.assertTrue(ok_a, reason_a) self.assertTrue(ok_b, reason_b) - def test_relative_path_is_resolved_before_checking(self): + def test_relative_path_resolves_against_comfyui_base_not_cwd(self): sub = self.output_dir / "rel" sub.mkdir() orig_cwd = os.getcwd() - os.chdir(str(self.output_dir)) + os.chdir(self.tmpdir) self.addCleanup(os.chdir, orig_cwd) - ok, reason = GalleryConfig.validate_gallery_root("rel") + + ok, reason = GalleryConfig.validate_gallery_root("output/rel") self.assertTrue(ok, reason) + self.assertEqual( + GalleryConfig.resolve_gallery_root("output/rel"), + os.path.normcase(os.path.realpath(sub)), + ) + # A path that only exists relative to the CWD is not accepted. + (Path(self.tmpdir) / "rel").mkdir() + self.assertFalse(GalleryConfig.validate_gallery_root("rel")[0]) def test_accepts_without_folder_paths_when_env_lists_dir(self): sys.modules.pop("folder_paths", None)