From dbf7fef140cbef531e6ba3c8d0086a2b7b1b0290 Mon Sep 17 00:00:00 2001 From: Glary-Bot Date: Wed, 10 Jun 2026 19:10:23 +0000 Subject: [PATCH 1/5] fix(frontend): clean up @latest frontend downloads (CORE-285) Old ComfyUI_frontend releases pulled in via --front-end-version @latest were never removed, causing web_custom_versions/ to grow unbounded (one user reported 5.11 GB). When the user requests an auto-resolving specifier (@latest or @prerelease), record which concrete version was materialized in a small .auto_managed.json next to the per-provider folders. On a subsequent auto-managed run, prune any previously-tracked folder that is no longer the current one. Explicitly pinned versions (e.g. @v1.46.0) are left alone, and if a user explicitly pins a version that had previously been downloaded under @latest, it is promoted out of the auto-managed set so it survives future cleanup. --- app/frontend_management.py | 154 +++++++++++++++++- tests-unit/app_test/frontend_manager_test.py | 162 +++++++++++++++++++ 2 files changed, 315 insertions(+), 1 deletion(-) diff --git a/app/frontend_management.py b/app/frontend_management.py index 8e84e8dd9075..fcf94c558dcc 100644 --- a/app/frontend_management.py +++ b/app/frontend_management.py @@ -1,7 +1,9 @@ import argparse +import json import logging import os import re +import shutil import sys import tempfile import zipfile @@ -205,6 +207,129 @@ def download_release_asset_zip(release: Release, destination_path: str) -> None: class FrontendManager: CUSTOM_FRONTENDS_ROOT = str(Path(__file__).parents[1] / "web_custom_versions") + # Version specifiers that resolve to a moving target on each invocation. + # Versions downloaded via these specifiers are tracked in the per-provider + # metadata file so that stale copies can be pruned when a new release + # becomes the current one. Explicitly pinned versions (e.g. ``@1.46.0`` or + # ``@v1.46.0``) are left alone so users can keep them around indefinitely + # for things like bisecting frontend regressions. + AUTO_MANAGED_VERSION_SPECIFIERS = ("latest", "prerelease") + + # File written next to per-provider version folders that records which + # versions were downloaded via an auto-managed specifier. Hidden so it does + # not show up as a sibling release in casual ``ls`` output. + AUTO_MANAGED_METADATA_FILENAME = ".auto_managed.json" + + @classmethod + def _provider_dir(cls, repo_owner: str, repo_name: str) -> Path: + return Path(cls.CUSTOM_FRONTENDS_ROOT) / f"{repo_owner}_{repo_name}" + + @classmethod + def _auto_managed_metadata_path(cls, repo_owner: str, repo_name: str) -> Path: + return cls._provider_dir(repo_owner, repo_name) / cls.AUTO_MANAGED_METADATA_FILENAME + + @classmethod + def _read_auto_managed_versions(cls, repo_owner: str, repo_name: str) -> list[str]: + """Return the list of versions previously downloaded under an + auto-managed specifier for this provider. Missing or unreadable + metadata is treated as an empty list.""" + metadata_path = cls._auto_managed_metadata_path(repo_owner, repo_name) + if not metadata_path.exists(): + return [] + try: + with open(metadata_path, "r", encoding="utf-8") as fh: + data = json.load(fh) + except (OSError, ValueError) as exc: + logging.warning( + "Could not read frontend auto-managed metadata at %s: %s", + metadata_path, + exc, + ) + return [] + versions = data.get("auto_managed", []) + if not isinstance(versions, list): + return [] + return [str(v) for v in versions if isinstance(v, (str, int))] + + @classmethod + def _write_auto_managed_versions( + cls, repo_owner: str, repo_name: str, versions: list[str] + ) -> None: + """Persist the auto-managed version list atomically. Deduped and + sorted for stability so the file is friendly to diffs.""" + metadata_path = cls._auto_managed_metadata_path(repo_owner, repo_name) + metadata_path.parent.mkdir(parents=True, exist_ok=True) + payload = {"auto_managed": sorted(set(versions))} + # Atomic write via temp file + rename so a crashed process can't leave + # a half-written metadata file behind. + tmp_path = metadata_path.with_suffix(metadata_path.suffix + ".tmp") + try: + with open(tmp_path, "w", encoding="utf-8") as fh: + json.dump(payload, fh, indent=2, sort_keys=True) + os.replace(tmp_path, metadata_path) + except OSError as exc: + logging.warning( + "Could not write frontend auto-managed metadata at %s: %s", + metadata_path, + exc, + ) + if tmp_path.exists(): + try: + tmp_path.unlink() + except OSError: + pass + + @classmethod + def _prune_auto_managed_versions( + cls, repo_owner: str, repo_name: str, keep_version: str + ) -> None: + """Remove all auto-managed version folders for this provider other + than ``keep_version`` and update the metadata to only list it. + + Folders that aren't currently tracked as auto-managed (i.e. explicitly + pinned downloads) are never touched. + """ + tracked = cls._read_auto_managed_versions(repo_owner, repo_name) + if not tracked and keep_version is None: + return + + provider_dir = cls._provider_dir(repo_owner, repo_name) + for stale_version in tracked: + if stale_version == keep_version: + continue + stale_path = provider_dir / stale_version + if not stale_path.exists(): + continue + try: + shutil.rmtree(stale_path) + logging.info( + "Removed stale auto-managed frontend version: %s", + stale_path, + ) + except OSError as exc: + logging.warning( + "Failed to remove stale frontend version at %s: %s", + stale_path, + exc, + ) + + new_tracked = [keep_version] if keep_version else [] + cls._write_auto_managed_versions(repo_owner, repo_name, new_tracked) + + @classmethod + def _untrack_auto_managed_version( + cls, repo_owner: str, repo_name: str, version: str + ) -> None: + """Drop ``version`` from the auto-managed list without deleting its + folder. Used when a user explicitly pins a version that previously + had been downloaded under ``@latest`` / ``@prerelease`` so the next + auto cleanup pass doesn't wipe it out.""" + tracked = cls._read_auto_managed_versions(repo_owner, repo_name) + if version not in tracked: + return + tracked = [v for v in tracked if v != version] + cls._write_auto_managed_versions(repo_owner, repo_name, tracked) + @classmethod def get_required_frontend_version(cls) -> str: """Get the required frontend package version.""" @@ -372,6 +497,7 @@ def init_frontend_unsafe( return cls.default_frontend_path() repo_owner, repo_name, version = cls.parse_version_string(version_string) + is_auto_managed = version in cls.AUTO_MANAGED_VERSION_SPECIFIERS if version.startswith("v"): expected_path = str( @@ -383,6 +509,12 @@ def init_frontend_unsafe( logging.info( f"Using existing copy of specific frontend version tag: {repo_owner}/{repo_name}@{version}" ) + # User explicitly pinned this exact version: promote it out of + # the auto-managed set so future @latest cleanups won't wipe + # it out. + cls._untrack_auto_managed_version( + repo_owner, repo_name, version.lstrip("v") + ) return expected_path logging.info( @@ -396,7 +528,8 @@ def init_frontend_unsafe( web_root = str( Path(cls.CUSTOM_FRONTENDS_ROOT) / provider.folder_name / semantic_version ) - if not os.path.exists(web_root): + download_succeeded = os.path.exists(web_root) + if not download_succeeded: try: os.makedirs(web_root, exist_ok=True) logging.info( @@ -407,10 +540,29 @@ def init_frontend_unsafe( ) logging.debug(release) download_release_asset_zip(release, destination_path=web_root) + download_succeeded = True finally: # Clean up the directory if it is empty, i.e. the download failed if not os.listdir(web_root): os.rmdir(web_root) + download_succeeded = False + + if download_succeeded: + if is_auto_managed: + # Wipe out previously-tracked auto-managed versions and record + # the current one. This is what keeps disk usage bounded when + # users run with ``--front-end-version @latest`` over a + # long period of time (CORE-285). + cls._prune_auto_managed_versions( + repo_owner, repo_name, semantic_version + ) + else: + # An explicit version request matched a folder that had been + # downloaded under @latest previously. Promote it so it is no + # longer subject to auto-cleanup. + cls._untrack_auto_managed_version( + repo_owner, repo_name, semantic_version + ) return web_root diff --git a/tests-unit/app_test/frontend_manager_test.py b/tests-unit/app_test/frontend_manager_test.py index 8c8a2eb48550..f62f902a57c3 100644 --- a/tests-unit/app_test/frontend_manager_test.py +++ b/tests-unit/app_test/frontend_manager_test.py @@ -1,4 +1,6 @@ import argparse +from pathlib import Path + import pytest from requests.exceptions import HTTPError from unittest.mock import patch, mock_open @@ -287,3 +289,163 @@ def test_get_installed_templates_version_not_installed(): # Assert assert version is None + + +# --------------------------------------------------------------------------- +# Auto-managed @latest / @prerelease cleanup (CORE-285) +# --------------------------------------------------------------------------- + + +@pytest.fixture +def custom_frontends_root(tmp_path, monkeypatch): + """Point ``FrontendManager.CUSTOM_FRONTENDS_ROOT`` at a fresh tmp dir.""" + root = tmp_path / "web_custom_versions" + root.mkdir() + monkeypatch.setattr(FrontendManager, "CUSTOM_FRONTENDS_ROOT", str(root)) + return root + + +def _make_version_dir(root, owner, repo, version): + """Create ``/_//index.html`` and return path.""" + folder = root / f"{owner}_{repo}" / version + folder.mkdir(parents=True, exist_ok=True) + (folder / "index.html").write_text("") + return folder + + +def test_auto_managed_metadata_roundtrip(custom_frontends_root): + FrontendManager._write_auto_managed_versions("o", "r", ["1.0.0", "1.1.0", "1.0.0"]) + assert FrontendManager._read_auto_managed_versions("o", "r") == ["1.0.0", "1.1.0"] + + +def test_read_auto_managed_versions_missing(custom_frontends_root): + assert FrontendManager._read_auto_managed_versions("o", "r") == [] + + +def test_read_auto_managed_versions_corrupt(custom_frontends_root): + provider_dir = custom_frontends_root / "o_r" + provider_dir.mkdir() + (provider_dir / FrontendManager.AUTO_MANAGED_METADATA_FILENAME).write_text( + "not json" + ) + assert FrontendManager._read_auto_managed_versions("o", "r") == [] + + +def test_prune_auto_managed_versions_removes_stale_and_keeps_pinned( + custom_frontends_root, +): + # Two versions previously fetched via @latest, plus an explicitly pinned one. + _make_version_dir(custom_frontends_root, "o", "r", "1.0.0") + _make_version_dir(custom_frontends_root, "o", "r", "1.1.0") + pinned = _make_version_dir(custom_frontends_root, "o", "r", "1.2.0") + FrontendManager._write_auto_managed_versions("o", "r", ["1.0.0", "1.1.0"]) + + # User runs @latest again and it resolves to 1.1.0 — older auto-managed + # 1.0.0 should be deleted, pinned 1.2.0 should remain untouched. + FrontendManager._prune_auto_managed_versions("o", "r", keep_version="1.1.0") + + provider_dir = custom_frontends_root / "o_r" + assert not (provider_dir / "1.0.0").exists() + assert (provider_dir / "1.1.0").exists() + assert pinned.exists() + assert FrontendManager._read_auto_managed_versions("o", "r") == ["1.1.0"] + + +def test_untrack_auto_managed_version_does_not_delete_folder(custom_frontends_root): + version_dir = _make_version_dir(custom_frontends_root, "o", "r", "1.0.0") + FrontendManager._write_auto_managed_versions("o", "r", ["1.0.0", "1.1.0"]) + + FrontendManager._untrack_auto_managed_version("o", "r", "1.0.0") + + assert version_dir.exists() + assert FrontendManager._read_auto_managed_versions("o", "r") == ["1.1.0"] + + +def test_init_frontend_latest_prunes_previous_auto_managed_versions( + custom_frontends_root, mock_provider, mock_releases +): + # Pre-existing folders: 1.0.0 was previously downloaded via @latest, 1.1.5 + # was explicitly pinned by the user. Now @latest resolves to 2.0.0. + _make_version_dir(custom_frontends_root, "test-owner", "test-repo", "1.0.0") + pinned = _make_version_dir( + custom_frontends_root, "test-owner", "test-repo", "1.1.5" + ) + FrontendManager._write_auto_managed_versions( + "test-owner", "test-repo", ["1.0.0"] + ) + + # Stub out the actual download so we just create the destination dir. + def fake_download(release, destination_path): + Path(destination_path).mkdir(parents=True, exist_ok=True) + (Path(destination_path) / "index.html").write_text("") + + with patch( + "app.frontend_management.download_release_asset_zip", + side_effect=fake_download, + ): + result = FrontendManager.init_frontend_unsafe( + "test-owner/test-repo@latest", mock_provider + ) + + provider_dir = custom_frontends_root / "test-owner_test-repo" + # 2.0.0 was downloaded and tracked. + assert Path(result) == provider_dir / "2.0.0" + assert (provider_dir / "2.0.0").exists() + assert FrontendManager._read_auto_managed_versions( + "test-owner", "test-repo" + ) == ["2.0.0"] + # Old auto-managed 1.0.0 was pruned. + assert not (provider_dir / "1.0.0").exists() + # Pinned 1.1.5 was left alone. + assert pinned.exists() + + +def test_init_frontend_explicit_version_promotes_out_of_auto_managed( + custom_frontends_root, mock_provider +): + # 1.0.0 was previously downloaded via @latest. + _make_version_dir(custom_frontends_root, "test-owner", "test-repo", "1.0.0") + FrontendManager._write_auto_managed_versions( + "test-owner", "test-repo", ["1.0.0"] + ) + + # User now explicitly pins it. The `v`-prefixed early-return path runs. + result = FrontendManager.init_frontend_unsafe( + "test-owner/test-repo@v1.0.0", mock_provider + ) + + provider_dir = custom_frontends_root / "test-owner_test-repo" + assert Path(result) == provider_dir / "1.0.0" + # It should no longer be tracked as auto-managed, so a future @latest run + # won't sweep it away. + assert FrontendManager._read_auto_managed_versions( + "test-owner", "test-repo" + ) == [] + # The folder is still on disk. + assert (provider_dir / "1.0.0").exists() + + +def test_init_frontend_explicit_version_no_v_prefix_promotes_out_of_auto_managed( + custom_frontends_root, mock_provider +): + # 1.0.0 was previously downloaded via @latest, and is also already on + # disk so the GitHub-resolution path is a no-op for download. + _make_version_dir(custom_frontends_root, "test-owner", "test-repo", "1.0.0") + FrontendManager._write_auto_managed_versions( + "test-owner", "test-repo", ["1.0.0"] + ) + + # No `v` prefix → goes through the GitHub release lookup path. The folder + # already exists, so download is skipped. + with patch( + "app.frontend_management.download_release_asset_zip" + ) as mock_download_zip: + FrontendManager.init_frontend_unsafe( + "test-owner/test-repo@1.0.0", mock_provider + ) + mock_download_zip.assert_not_called() + + # It should be promoted out of auto-managed even when the folder is reused. + assert FrontendManager._read_auto_managed_versions( + "test-owner", "test-repo" + ) == [] From bb2c1db8c70d16f090b4abffa0dd818380620b2a Mon Sep 17 00:00:00 2001 From: Glary-Bot Date: Wed, 10 Jun 2026 19:16:36 +0000 Subject: [PATCH 2/5] harden: validate metadata shape and refuse out-of-dir cleanup paths Addresses review feedback on the auto-managed metadata helpers: - json.load() on the metadata file can return non-dict values (e.g. a bare list or a string); guard the root type before calling .get(). - A tampered or hand-edited .auto_managed.json could contain entries like '../escape'. The previous code happily fed those into rmtree. Filter such entries out at both read time and write time so they never reach disk or cleanup, and add a belt-and-suspenders path containment check in _prune_auto_managed_versions that requires the resolved target to live strictly under the resolved provider dir. --- app/frontend_management.py | 78 ++++++++++++++++++-- tests-unit/app_test/frontend_manager_test.py | 62 ++++++++++++++++ 2 files changed, 135 insertions(+), 5 deletions(-) diff --git a/app/frontend_management.py b/app/frontend_management.py index fcf94c558dcc..cdae228dc22b 100644 --- a/app/frontend_management.py +++ b/app/frontend_management.py @@ -228,11 +228,26 @@ def _provider_dir(cls, repo_owner: str, repo_name: str) -> Path: def _auto_managed_metadata_path(cls, repo_owner: str, repo_name: str) -> Path: return cls._provider_dir(repo_owner, repo_name) / cls.AUTO_MANAGED_METADATA_FILENAME + # A version directory name must look like a simple semver-ish token. We + # use this as a defensive allowlist when interpreting metadata so a + # malformed or tampered ``.auto_managed.json`` cannot point cleanup at + # paths outside the provider directory (e.g. ``../somewhere``). + _VERSION_DIRNAME_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$") + + @classmethod + def _is_safe_version_dirname(cls, name: str) -> bool: + if not isinstance(name, str): + return False + if name in (".", "..") or "/" in name or "\\" in name or "\x00" in name: + return False + return bool(cls._VERSION_DIRNAME_PATTERN.match(name)) + @classmethod def _read_auto_managed_versions(cls, repo_owner: str, repo_name: str) -> list[str]: """Return the list of versions previously downloaded under an - auto-managed specifier for this provider. Missing or unreadable - metadata is treated as an empty list.""" + auto-managed specifier for this provider. Missing, unreadable, or + otherwise malformed metadata is treated as an empty list so a bad + file never blocks startup or directs cleanup at unrelated paths.""" metadata_path = cls._auto_managed_metadata_path(repo_owner, repo_name) if not metadata_path.exists(): return [] @@ -246,20 +261,31 @@ def _read_auto_managed_versions(cls, repo_owner: str, repo_name: str) -> list[st exc, ) return [] + if not isinstance(data, dict): + logging.warning( + "Frontend auto-managed metadata at %s has unexpected shape; ignoring.", + metadata_path, + ) + return [] versions = data.get("auto_managed", []) if not isinstance(versions, list): return [] - return [str(v) for v in versions if isinstance(v, (str, int))] + # Filter out anything that doesn't look like a safe version dirname + # so a tampered file can't point us at, say, ``../../etc``. + return [v for v in versions if cls._is_safe_version_dirname(v)] @classmethod def _write_auto_managed_versions( cls, repo_owner: str, repo_name: str, versions: list[str] ) -> None: """Persist the auto-managed version list atomically. Deduped and - sorted for stability so the file is friendly to diffs.""" + sorted for stability so the file is friendly to diffs. Any entry that + doesn't look like a safe version dirname is dropped before write so + the on-disk metadata always contains valid values.""" metadata_path = cls._auto_managed_metadata_path(repo_owner, repo_name) metadata_path.parent.mkdir(parents=True, exist_ok=True) - payload = {"auto_managed": sorted(set(versions))} + safe_versions = [v for v in versions if cls._is_safe_version_dirname(v)] + payload = {"auto_managed": sorted(set(safe_versions))} # Atomic write via temp file + rename so a crashed process can't leave # a half-written metadata file behind. tmp_path = metadata_path.with_suffix(metadata_path.suffix + ".tmp") @@ -294,12 +320,54 @@ def _prune_auto_managed_versions( return provider_dir = cls._provider_dir(repo_owner, repo_name) + try: + provider_dir_resolved = provider_dir.resolve() + except OSError as exc: + logging.warning( + "Could not resolve provider directory %s for cleanup: %s", + provider_dir, + exc, + ) + return + for stale_version in tracked: if stale_version == keep_version: continue + # ``_read_auto_managed_versions`` already filters tracked entries + # through ``_is_safe_version_dirname``, but re-check here so that + # this helper is also safe when called with externally-supplied + # version lists (and so a defense-in-depth audit can confirm the + # rmtree target lives under the provider directory). + if not cls._is_safe_version_dirname(stale_version): + logging.warning( + "Refusing to clean up suspicious frontend version name: %r", + stale_version, + ) + continue stale_path = provider_dir / stale_version if not stale_path.exists(): continue + try: + stale_resolved = stale_path.resolve() + except OSError as exc: + logging.warning( + "Could not resolve stale frontend path %s: %s", + stale_path, + exc, + ) + continue + # Ensure the resolved target lives strictly under the resolved + # provider directory (so symlinks / path tricks can't escape). + if ( + stale_resolved == provider_dir_resolved + or provider_dir_resolved not in stale_resolved.parents + ): + logging.warning( + "Refusing to remove path outside provider dir: %s (provider=%s)", + stale_resolved, + provider_dir_resolved, + ) + continue try: shutil.rmtree(stale_path) logging.info( diff --git a/tests-unit/app_test/frontend_manager_test.py b/tests-unit/app_test/frontend_manager_test.py index f62f902a57c3..2e9a31f1924b 100644 --- a/tests-unit/app_test/frontend_manager_test.py +++ b/tests-unit/app_test/frontend_manager_test.py @@ -331,6 +331,68 @@ def test_read_auto_managed_versions_corrupt(custom_frontends_root): assert FrontendManager._read_auto_managed_versions("o", "r") == [] +@pytest.mark.parametrize("payload", ["[]", "null", '"oops"', "42"]) +def test_read_auto_managed_versions_non_dict_root(custom_frontends_root, payload): + """Valid JSON whose root isn't a dict must not raise — bug pointed out in + Oracle review: ``data.get(...)`` would throw on non-dict roots.""" + provider_dir = custom_frontends_root / "o_r" + provider_dir.mkdir() + (provider_dir / FrontendManager.AUTO_MANAGED_METADATA_FILENAME).write_text(payload) + assert FrontendManager._read_auto_managed_versions("o", "r") == [] + + +def test_read_auto_managed_versions_drops_unsafe_names(custom_frontends_root): + """Tampered metadata containing path-traversal-y names must be ignored + at read time so cleanup never even sees them.""" + provider_dir = custom_frontends_root / "o_r" + provider_dir.mkdir() + (provider_dir / FrontendManager.AUTO_MANAGED_METADATA_FILENAME).write_text( + '{"auto_managed": ["1.0.0", "../escape", "/etc", "..", ".", "ok-1.2"]}' + ) + assert FrontendManager._read_auto_managed_versions("o", "r") == ["1.0.0", "ok-1.2"] + + +def test_write_auto_managed_versions_drops_unsafe_names(custom_frontends_root): + """Even if a caller passes a tainted list, the file we persist must + only contain safe names.""" + FrontendManager._write_auto_managed_versions( + "o", "r", ["1.0.0", "../escape", "/etc/passwd", "..", "."] + ) + assert FrontendManager._read_auto_managed_versions("o", "r") == ["1.0.0"] + + +def test_prune_refuses_to_delete_outside_provider_dir( + custom_frontends_root, monkeypatch +): + """Defense in depth: even if a hostile name slips past the read/write + filters somehow, ``_prune_auto_managed_versions`` must refuse to rmtree + anything outside the provider directory.""" + # Set up a sibling directory that must NOT be touched. + sibling = custom_frontends_root / "do-not-touch" + sibling.mkdir() + (sibling / "marker").write_text("keep me") + + provider_dir = custom_frontends_root / "o_r" + provider_dir.mkdir() + _make_version_dir(custom_frontends_root, "o", "r", "1.0.0") + + # Bypass the read filter to simulate a corrupt-but-parseable list slipping + # through (e.g. older code wrote it before this hardening landed). + monkeypatch.setattr( + FrontendManager, + "_read_auto_managed_versions", + classmethod(lambda cls, owner, repo: ["1.0.0", "../do-not-touch"]), + ) + + FrontendManager._prune_auto_managed_versions("o", "r", keep_version="2.0.0") + + # Sibling untouched. + assert sibling.exists() + assert (sibling / "marker").read_text() == "keep me" + # In-bounds folder still gets cleaned. + assert not (provider_dir / "1.0.0").exists() + + def test_prune_auto_managed_versions_removes_stale_and_keeps_pinned( custom_frontends_root, ): From 0a2b9f8864ae2adbe1ba0d0e6bfbd1eaf81f793e Mon Sep 17 00:00:00 2001 From: Glary-Bot Date: Thu, 11 Jun 2026 20:14:47 +0000 Subject: [PATCH 3/5] trim verbose comments and docstrings --- app/frontend_management.py | 71 +++++--------------- tests-unit/app_test/frontend_manager_test.py | 39 +---------- 2 files changed, 17 insertions(+), 93 deletions(-) diff --git a/app/frontend_management.py b/app/frontend_management.py index cdae228dc22b..8da20ca3ef22 100644 --- a/app/frontend_management.py +++ b/app/frontend_management.py @@ -207,19 +207,16 @@ def download_release_asset_zip(release: Release, destination_path: str) -> None: class FrontendManager: CUSTOM_FRONTENDS_ROOT = str(Path(__file__).parents[1] / "web_custom_versions") - # Version specifiers that resolve to a moving target on each invocation. - # Versions downloaded via these specifiers are tracked in the per-provider - # metadata file so that stale copies can be pruned when a new release - # becomes the current one. Explicitly pinned versions (e.g. ``@1.46.0`` or - # ``@v1.46.0``) are left alone so users can keep them around indefinitely - # for things like bisecting frontend regressions. + # Specifiers that resolve to a moving target. Downloads made via these are + # tracked so old copies can be pruned; explicit pins (@1.46.0, @v1.46.0) + # are left alone. AUTO_MANAGED_VERSION_SPECIFIERS = ("latest", "prerelease") - - # File written next to per-provider version folders that records which - # versions were downloaded via an auto-managed specifier. Hidden so it does - # not show up as a sibling release in casual ``ls`` output. AUTO_MANAGED_METADATA_FILENAME = ".auto_managed.json" + # Allowlist for version directory names. Prevents a tampered metadata + # file from steering cleanup at paths like "../etc". + _VERSION_DIRNAME_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$") + @classmethod def _provider_dir(cls, repo_owner: str, repo_name: str) -> Path: return Path(cls.CUSTOM_FRONTENDS_ROOT) / f"{repo_owner}_{repo_name}" @@ -228,12 +225,6 @@ def _provider_dir(cls, repo_owner: str, repo_name: str) -> Path: def _auto_managed_metadata_path(cls, repo_owner: str, repo_name: str) -> Path: return cls._provider_dir(repo_owner, repo_name) / cls.AUTO_MANAGED_METADATA_FILENAME - # A version directory name must look like a simple semver-ish token. We - # use this as a defensive allowlist when interpreting metadata so a - # malformed or tampered ``.auto_managed.json`` cannot point cleanup at - # paths outside the provider directory (e.g. ``../somewhere``). - _VERSION_DIRNAME_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$") - @classmethod def _is_safe_version_dirname(cls, name: str) -> bool: if not isinstance(name, str): @@ -244,10 +235,8 @@ def _is_safe_version_dirname(cls, name: str) -> bool: @classmethod def _read_auto_managed_versions(cls, repo_owner: str, repo_name: str) -> list[str]: - """Return the list of versions previously downloaded under an - auto-managed specifier for this provider. Missing, unreadable, or - otherwise malformed metadata is treated as an empty list so a bad - file never blocks startup or directs cleanup at unrelated paths.""" + """Versions tracked as auto-managed for this provider. Any error or + malformed content yields an empty list.""" metadata_path = cls._auto_managed_metadata_path(repo_owner, repo_name) if not metadata_path.exists(): return [] @@ -270,24 +259,17 @@ def _read_auto_managed_versions(cls, repo_owner: str, repo_name: str) -> list[st versions = data.get("auto_managed", []) if not isinstance(versions, list): return [] - # Filter out anything that doesn't look like a safe version dirname - # so a tampered file can't point us at, say, ``../../etc``. return [v for v in versions if cls._is_safe_version_dirname(v)] @classmethod def _write_auto_managed_versions( cls, repo_owner: str, repo_name: str, versions: list[str] ) -> None: - """Persist the auto-managed version list atomically. Deduped and - sorted for stability so the file is friendly to diffs. Any entry that - doesn't look like a safe version dirname is dropped before write so - the on-disk metadata always contains valid values.""" metadata_path = cls._auto_managed_metadata_path(repo_owner, repo_name) metadata_path.parent.mkdir(parents=True, exist_ok=True) safe_versions = [v for v in versions if cls._is_safe_version_dirname(v)] payload = {"auto_managed": sorted(set(safe_versions))} - # Atomic write via temp file + rename so a crashed process can't leave - # a half-written metadata file behind. + # Atomic write: tmp + rename so a crash can't leave a half-written file. tmp_path = metadata_path.with_suffix(metadata_path.suffix + ".tmp") try: with open(tmp_path, "w", encoding="utf-8") as fh: @@ -309,12 +291,8 @@ def _write_auto_managed_versions( def _prune_auto_managed_versions( cls, repo_owner: str, repo_name: str, keep_version: str ) -> None: - """Remove all auto-managed version folders for this provider other - than ``keep_version`` and update the metadata to only list it. - - Folders that aren't currently tracked as auto-managed (i.e. explicitly - pinned downloads) are never touched. - """ + """Remove tracked auto-managed folders other than ``keep_version`` + and rewrite the metadata to list only it.""" tracked = cls._read_auto_managed_versions(repo_owner, repo_name) if not tracked and keep_version is None: return @@ -333,11 +311,8 @@ def _prune_auto_managed_versions( for stale_version in tracked: if stale_version == keep_version: continue - # ``_read_auto_managed_versions`` already filters tracked entries - # through ``_is_safe_version_dirname``, but re-check here so that - # this helper is also safe when called with externally-supplied - # version lists (and so a defense-in-depth audit can confirm the - # rmtree target lives under the provider directory). + # Defense in depth: read already filters, but re-check here so + # the rmtree target is provably under the provider dir. if not cls._is_safe_version_dirname(stale_version): logging.warning( "Refusing to clean up suspicious frontend version name: %r", @@ -356,8 +331,6 @@ def _prune_auto_managed_versions( exc, ) continue - # Ensure the resolved target lives strictly under the resolved - # provider directory (so symlinks / path tricks can't escape). if ( stale_resolved == provider_dir_resolved or provider_dir_resolved not in stale_resolved.parents @@ -388,10 +361,8 @@ def _prune_auto_managed_versions( def _untrack_auto_managed_version( cls, repo_owner: str, repo_name: str, version: str ) -> None: - """Drop ``version`` from the auto-managed list without deleting its - folder. Used when a user explicitly pins a version that previously - had been downloaded under ``@latest`` / ``@prerelease`` so the next - auto cleanup pass doesn't wipe it out.""" + """Drop ``version`` from auto-managed tracking without touching its + folder, so a subsequent explicit pin survives later @latest cleanups.""" tracked = cls._read_auto_managed_versions(repo_owner, repo_name) if version not in tracked: return @@ -577,9 +548,6 @@ def init_frontend_unsafe( logging.info( f"Using existing copy of specific frontend version tag: {repo_owner}/{repo_name}@{version}" ) - # User explicitly pinned this exact version: promote it out of - # the auto-managed set so future @latest cleanups won't wipe - # it out. cls._untrack_auto_managed_version( repo_owner, repo_name, version.lstrip("v") ) @@ -617,17 +585,10 @@ def init_frontend_unsafe( if download_succeeded: if is_auto_managed: - # Wipe out previously-tracked auto-managed versions and record - # the current one. This is what keeps disk usage bounded when - # users run with ``--front-end-version @latest`` over a - # long period of time (CORE-285). cls._prune_auto_managed_versions( repo_owner, repo_name, semantic_version ) else: - # An explicit version request matched a folder that had been - # downloaded under @latest previously. Promote it so it is no - # longer subject to auto-cleanup. cls._untrack_auto_managed_version( repo_owner, repo_name, semantic_version ) diff --git a/tests-unit/app_test/frontend_manager_test.py b/tests-unit/app_test/frontend_manager_test.py index 2e9a31f1924b..aa2e35c3f45e 100644 --- a/tests-unit/app_test/frontend_manager_test.py +++ b/tests-unit/app_test/frontend_manager_test.py @@ -291,14 +291,11 @@ def test_get_installed_templates_version_not_installed(): assert version is None -# --------------------------------------------------------------------------- # Auto-managed @latest / @prerelease cleanup (CORE-285) -# --------------------------------------------------------------------------- @pytest.fixture def custom_frontends_root(tmp_path, monkeypatch): - """Point ``FrontendManager.CUSTOM_FRONTENDS_ROOT`` at a fresh tmp dir.""" root = tmp_path / "web_custom_versions" root.mkdir() monkeypatch.setattr(FrontendManager, "CUSTOM_FRONTENDS_ROOT", str(root)) @@ -306,7 +303,6 @@ def custom_frontends_root(tmp_path, monkeypatch): def _make_version_dir(root, owner, repo, version): - """Create ``/_//index.html`` and return path.""" folder = root / f"{owner}_{repo}" / version folder.mkdir(parents=True, exist_ok=True) (folder / "index.html").write_text("") @@ -333,8 +329,6 @@ def test_read_auto_managed_versions_corrupt(custom_frontends_root): @pytest.mark.parametrize("payload", ["[]", "null", '"oops"', "42"]) def test_read_auto_managed_versions_non_dict_root(custom_frontends_root, payload): - """Valid JSON whose root isn't a dict must not raise — bug pointed out in - Oracle review: ``data.get(...)`` would throw on non-dict roots.""" provider_dir = custom_frontends_root / "o_r" provider_dir.mkdir() (provider_dir / FrontendManager.AUTO_MANAGED_METADATA_FILENAME).write_text(payload) @@ -342,8 +336,6 @@ def test_read_auto_managed_versions_non_dict_root(custom_frontends_root, payload def test_read_auto_managed_versions_drops_unsafe_names(custom_frontends_root): - """Tampered metadata containing path-traversal-y names must be ignored - at read time so cleanup never even sees them.""" provider_dir = custom_frontends_root / "o_r" provider_dir.mkdir() (provider_dir / FrontendManager.AUTO_MANAGED_METADATA_FILENAME).write_text( @@ -353,8 +345,6 @@ def test_read_auto_managed_versions_drops_unsafe_names(custom_frontends_root): def test_write_auto_managed_versions_drops_unsafe_names(custom_frontends_root): - """Even if a caller passes a tainted list, the file we persist must - only contain safe names.""" FrontendManager._write_auto_managed_versions( "o", "r", ["1.0.0", "../escape", "/etc/passwd", "..", "."] ) @@ -364,10 +354,6 @@ def test_write_auto_managed_versions_drops_unsafe_names(custom_frontends_root): def test_prune_refuses_to_delete_outside_provider_dir( custom_frontends_root, monkeypatch ): - """Defense in depth: even if a hostile name slips past the read/write - filters somehow, ``_prune_auto_managed_versions`` must refuse to rmtree - anything outside the provider directory.""" - # Set up a sibling directory that must NOT be touched. sibling = custom_frontends_root / "do-not-touch" sibling.mkdir() (sibling / "marker").write_text("keep me") @@ -376,8 +362,6 @@ def test_prune_refuses_to_delete_outside_provider_dir( provider_dir.mkdir() _make_version_dir(custom_frontends_root, "o", "r", "1.0.0") - # Bypass the read filter to simulate a corrupt-but-parseable list slipping - # through (e.g. older code wrote it before this hardening landed). monkeypatch.setattr( FrontendManager, "_read_auto_managed_versions", @@ -386,24 +370,19 @@ def test_prune_refuses_to_delete_outside_provider_dir( FrontendManager._prune_auto_managed_versions("o", "r", keep_version="2.0.0") - # Sibling untouched. assert sibling.exists() assert (sibling / "marker").read_text() == "keep me" - # In-bounds folder still gets cleaned. assert not (provider_dir / "1.0.0").exists() def test_prune_auto_managed_versions_removes_stale_and_keeps_pinned( custom_frontends_root, ): - # Two versions previously fetched via @latest, plus an explicitly pinned one. _make_version_dir(custom_frontends_root, "o", "r", "1.0.0") _make_version_dir(custom_frontends_root, "o", "r", "1.1.0") pinned = _make_version_dir(custom_frontends_root, "o", "r", "1.2.0") FrontendManager._write_auto_managed_versions("o", "r", ["1.0.0", "1.1.0"]) - # User runs @latest again and it resolves to 1.1.0 — older auto-managed - # 1.0.0 should be deleted, pinned 1.2.0 should remain untouched. FrontendManager._prune_auto_managed_versions("o", "r", keep_version="1.1.0") provider_dir = custom_frontends_root / "o_r" @@ -426,8 +405,6 @@ def test_untrack_auto_managed_version_does_not_delete_folder(custom_frontends_ro def test_init_frontend_latest_prunes_previous_auto_managed_versions( custom_frontends_root, mock_provider, mock_releases ): - # Pre-existing folders: 1.0.0 was previously downloaded via @latest, 1.1.5 - # was explicitly pinned by the user. Now @latest resolves to 2.0.0. _make_version_dir(custom_frontends_root, "test-owner", "test-repo", "1.0.0") pinned = _make_version_dir( custom_frontends_root, "test-owner", "test-repo", "1.1.5" @@ -436,7 +413,6 @@ def test_init_frontend_latest_prunes_previous_auto_managed_versions( "test-owner", "test-repo", ["1.0.0"] ) - # Stub out the actual download so we just create the destination dir. def fake_download(release, destination_path): Path(destination_path).mkdir(parents=True, exist_ok=True) (Path(destination_path) / "index.html").write_text("") @@ -450,55 +426,43 @@ def fake_download(release, destination_path): ) provider_dir = custom_frontends_root / "test-owner_test-repo" - # 2.0.0 was downloaded and tracked. assert Path(result) == provider_dir / "2.0.0" assert (provider_dir / "2.0.0").exists() assert FrontendManager._read_auto_managed_versions( "test-owner", "test-repo" ) == ["2.0.0"] - # Old auto-managed 1.0.0 was pruned. assert not (provider_dir / "1.0.0").exists() - # Pinned 1.1.5 was left alone. assert pinned.exists() def test_init_frontend_explicit_version_promotes_out_of_auto_managed( custom_frontends_root, mock_provider ): - # 1.0.0 was previously downloaded via @latest. _make_version_dir(custom_frontends_root, "test-owner", "test-repo", "1.0.0") FrontendManager._write_auto_managed_versions( "test-owner", "test-repo", ["1.0.0"] ) - # User now explicitly pins it. The `v`-prefixed early-return path runs. result = FrontendManager.init_frontend_unsafe( "test-owner/test-repo@v1.0.0", mock_provider ) provider_dir = custom_frontends_root / "test-owner_test-repo" assert Path(result) == provider_dir / "1.0.0" - # It should no longer be tracked as auto-managed, so a future @latest run - # won't sweep it away. assert FrontendManager._read_auto_managed_versions( "test-owner", "test-repo" ) == [] - # The folder is still on disk. assert (provider_dir / "1.0.0").exists() -def test_init_frontend_explicit_version_no_v_prefix_promotes_out_of_auto_managed( +def test_init_frontend_explicit_no_v_prefix_promotes_out_of_auto_managed( custom_frontends_root, mock_provider ): - # 1.0.0 was previously downloaded via @latest, and is also already on - # disk so the GitHub-resolution path is a no-op for download. _make_version_dir(custom_frontends_root, "test-owner", "test-repo", "1.0.0") FrontendManager._write_auto_managed_versions( "test-owner", "test-repo", ["1.0.0"] ) - # No `v` prefix → goes through the GitHub release lookup path. The folder - # already exists, so download is skipped. with patch( "app.frontend_management.download_release_asset_zip" ) as mock_download_zip: @@ -507,7 +471,6 @@ def test_init_frontend_explicit_version_no_v_prefix_promotes_out_of_auto_managed ) mock_download_zip.assert_not_called() - # It should be promoted out of auto-managed even when the folder is reused. assert FrontendManager._read_auto_managed_versions( "test-owner", "test-repo" ) == [] From 273314a690f3532dd08047b90d3de15947603988 Mon Sep 17 00:00:00 2001 From: Glary-Bot Date: Thu, 11 Jun 2026 20:17:22 +0000 Subject: [PATCH 4/5] trim remaining comments; keep only the security-relevant one --- app/frontend_management.py | 17 ++--------------- 1 file changed, 2 insertions(+), 15 deletions(-) diff --git a/app/frontend_management.py b/app/frontend_management.py index 8da20ca3ef22..fb2c639a8bd4 100644 --- a/app/frontend_management.py +++ b/app/frontend_management.py @@ -207,14 +207,8 @@ def download_release_asset_zip(release: Release, destination_path: str) -> None: class FrontendManager: CUSTOM_FRONTENDS_ROOT = str(Path(__file__).parents[1] / "web_custom_versions") - # Specifiers that resolve to a moving target. Downloads made via these are - # tracked so old copies can be pruned; explicit pins (@1.46.0, @v1.46.0) - # are left alone. AUTO_MANAGED_VERSION_SPECIFIERS = ("latest", "prerelease") AUTO_MANAGED_METADATA_FILENAME = ".auto_managed.json" - - # Allowlist for version directory names. Prevents a tampered metadata - # file from steering cleanup at paths like "../etc". _VERSION_DIRNAME_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$") @classmethod @@ -235,8 +229,6 @@ def _is_safe_version_dirname(cls, name: str) -> bool: @classmethod def _read_auto_managed_versions(cls, repo_owner: str, repo_name: str) -> list[str]: - """Versions tracked as auto-managed for this provider. Any error or - malformed content yields an empty list.""" metadata_path = cls._auto_managed_metadata_path(repo_owner, repo_name) if not metadata_path.exists(): return [] @@ -269,7 +261,6 @@ def _write_auto_managed_versions( metadata_path.parent.mkdir(parents=True, exist_ok=True) safe_versions = [v for v in versions if cls._is_safe_version_dirname(v)] payload = {"auto_managed": sorted(set(safe_versions))} - # Atomic write: tmp + rename so a crash can't leave a half-written file. tmp_path = metadata_path.with_suffix(metadata_path.suffix + ".tmp") try: with open(tmp_path, "w", encoding="utf-8") as fh: @@ -291,8 +282,6 @@ def _write_auto_managed_versions( def _prune_auto_managed_versions( cls, repo_owner: str, repo_name: str, keep_version: str ) -> None: - """Remove tracked auto-managed folders other than ``keep_version`` - and rewrite the metadata to list only it.""" tracked = cls._read_auto_managed_versions(repo_owner, repo_name) if not tracked and keep_version is None: return @@ -311,8 +300,8 @@ def _prune_auto_managed_versions( for stale_version in tracked: if stale_version == keep_version: continue - # Defense in depth: read already filters, but re-check here so - # the rmtree target is provably under the provider dir. + # Re-check even though read already filters: keeps rmtree provably + # bounded under provider_dir regardless of caller. if not cls._is_safe_version_dirname(stale_version): logging.warning( "Refusing to clean up suspicious frontend version name: %r", @@ -361,8 +350,6 @@ def _prune_auto_managed_versions( def _untrack_auto_managed_version( cls, repo_owner: str, repo_name: str, version: str ) -> None: - """Drop ``version`` from auto-managed tracking without touching its - folder, so a subsequent explicit pin survives later @latest cleanups.""" tracked = cls._read_auto_managed_versions(repo_owner, repo_name) if version not in tracked: return From 672b4a255f3b081b2469012e087cf88279e2f5cb Mon Sep 17 00:00:00 2001 From: DrJKL Date: Thu, 11 Jun 2026 14:24:44 -0700 Subject: [PATCH 5/5] Simplify auto-managed frontend cleanup with on-disk markers Replace the .auto_managed.json sidecar with a .auto_managed/ marker directory. Tracked version names now come from real single-component dirents, removing the untrusted-input parsing, path-traversal checks, and rmtree boundary guards that the JSON design required. Also fix the dead Optional handling in _prune_auto_managed_versions, extract _ensure_release_downloaded to drop the download_succeeded flag, and reuse _provider_dir in init_frontend_unsafe. Amp-Thread-ID: https://ampcode.com/threads/T-019eb879-1e6f-77aa-abb3-4d229d18061f Co-authored-by: Amp --- app/frontend_management.py | 241 +++++++------------ tests-unit/app_test/frontend_manager_test.py | 82 ++----- 2 files changed, 117 insertions(+), 206 deletions(-) diff --git a/app/frontend_management.py b/app/frontend_management.py index fb2c639a8bd4..7e456bcda9f3 100644 --- a/app/frontend_management.py +++ b/app/frontend_management.py @@ -1,5 +1,4 @@ import argparse -import json import logging import os import re @@ -208,153 +207,95 @@ class FrontendManager: CUSTOM_FRONTENDS_ROOT = str(Path(__file__).parents[1] / "web_custom_versions") AUTO_MANAGED_VERSION_SPECIFIERS = ("latest", "prerelease") - AUTO_MANAGED_METADATA_FILENAME = ".auto_managed.json" - _VERSION_DIRNAME_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$") + AUTO_MANAGED_MARKER_DIRNAME = ".auto_managed" @classmethod def _provider_dir(cls, repo_owner: str, repo_name: str) -> Path: return Path(cls.CUSTOM_FRONTENDS_ROOT) / f"{repo_owner}_{repo_name}" @classmethod - def _auto_managed_metadata_path(cls, repo_owner: str, repo_name: str) -> Path: - return cls._provider_dir(repo_owner, repo_name) / cls.AUTO_MANAGED_METADATA_FILENAME - - @classmethod - def _is_safe_version_dirname(cls, name: str) -> bool: - if not isinstance(name, str): - return False - if name in (".", "..") or "/" in name or "\\" in name or "\x00" in name: - return False - return bool(cls._VERSION_DIRNAME_PATTERN.match(name)) + def _auto_managed_marker_dir(cls, repo_owner: str, repo_name: str) -> Path: + return cls._provider_dir(repo_owner, repo_name) / cls.AUTO_MANAGED_MARKER_DIRNAME @classmethod def _read_auto_managed_versions(cls, repo_owner: str, repo_name: str) -> list[str]: - metadata_path = cls._auto_managed_metadata_path(repo_owner, repo_name) - if not metadata_path.exists(): - return [] + """Return versions ComfyUI auto-downloaded for @latest / @prerelease. + + Each tracked version is an empty marker file under the provider's + ``.auto_managed`` directory. Because the names come straight from real + single-component directory entries, there is no untrusted parsing and no + path-traversal surface to defend against. + """ + marker_dir = cls._auto_managed_marker_dir(repo_owner, repo_name) try: - with open(metadata_path, "r", encoding="utf-8") as fh: - data = json.load(fh) - except (OSError, ValueError) as exc: - logging.warning( - "Could not read frontend auto-managed metadata at %s: %s", - metadata_path, - exc, - ) + return sorted(entry.name for entry in marker_dir.iterdir() if entry.is_file()) + except FileNotFoundError: return [] - if not isinstance(data, dict): + except OSError as exc: logging.warning( - "Frontend auto-managed metadata at %s has unexpected shape; ignoring.", - metadata_path, + "Could not read frontend auto-managed markers at %s: %s", + marker_dir, + exc, ) return [] - versions = data.get("auto_managed", []) - if not isinstance(versions, list): - return [] - return [v for v in versions if cls._is_safe_version_dirname(v)] @classmethod - def _write_auto_managed_versions( - cls, repo_owner: str, repo_name: str, versions: list[str] - ) -> None: - metadata_path = cls._auto_managed_metadata_path(repo_owner, repo_name) - metadata_path.parent.mkdir(parents=True, exist_ok=True) - safe_versions = [v for v in versions if cls._is_safe_version_dirname(v)] - payload = {"auto_managed": sorted(set(safe_versions))} - tmp_path = metadata_path.with_suffix(metadata_path.suffix + ".tmp") + def _mark_auto_managed(cls, repo_owner: str, repo_name: str, version: str) -> None: + marker_dir = cls._auto_managed_marker_dir(repo_owner, repo_name) try: - with open(tmp_path, "w", encoding="utf-8") as fh: - json.dump(payload, fh, indent=2, sort_keys=True) - os.replace(tmp_path, metadata_path) + marker_dir.mkdir(parents=True, exist_ok=True) + (marker_dir / version).touch() except OSError as exc: logging.warning( - "Could not write frontend auto-managed metadata at %s: %s", - metadata_path, + "Could not record auto-managed frontend version %s: %s", + version, exc, ) - if tmp_path.exists(): - try: - tmp_path.unlink() - except OSError: - pass @classmethod def _prune_auto_managed_versions( cls, repo_owner: str, repo_name: str, keep_version: str ) -> None: - tracked = cls._read_auto_managed_versions(repo_owner, repo_name) - if not tracked and keep_version is None: - return - + """Remove previously auto-downloaded versions other than ``keep_version``.""" provider_dir = cls._provider_dir(repo_owner, repo_name) - try: - provider_dir_resolved = provider_dir.resolve() - except OSError as exc: - logging.warning( - "Could not resolve provider directory %s for cleanup: %s", - provider_dir, - exc, - ) - return - - for stale_version in tracked: + for stale_version in cls._read_auto_managed_versions(repo_owner, repo_name): if stale_version == keep_version: continue - # Re-check even though read already filters: keeps rmtree provably - # bounded under provider_dir regardless of caller. - if not cls._is_safe_version_dirname(stale_version): - logging.warning( - "Refusing to clean up suspicious frontend version name: %r", - stale_version, - ) - continue + # stale_version is a single-component marker name, so this is always a + # direct child of provider_dir. stale_path = provider_dir / stale_version - if not stale_path.exists(): - continue - try: - stale_resolved = stale_path.resolve() - except OSError as exc: - logging.warning( - "Could not resolve stale frontend path %s: %s", - stale_path, - exc, - ) - continue - if ( - stale_resolved == provider_dir_resolved - or provider_dir_resolved not in stale_resolved.parents - ): - logging.warning( - "Refusing to remove path outside provider dir: %s (provider=%s)", - stale_resolved, - provider_dir_resolved, - ) - continue - try: - shutil.rmtree(stale_path) - logging.info( - "Removed stale auto-managed frontend version: %s", - stale_path, - ) - except OSError as exc: - logging.warning( - "Failed to remove stale frontend version at %s: %s", - stale_path, - exc, - ) - - new_tracked = [keep_version] if keep_version else [] - cls._write_auto_managed_versions(repo_owner, repo_name, new_tracked) + if stale_path.exists(): + try: + shutil.rmtree(stale_path) + logging.info( + "Removed stale auto-managed frontend version: %s", stale_path + ) + except OSError as exc: + logging.warning( + "Failed to remove stale frontend version at %s: %s", + stale_path, + exc, + ) + continue + cls._untrack_auto_managed_version(repo_owner, repo_name, stale_version) + cls._mark_auto_managed(repo_owner, repo_name, keep_version) @classmethod def _untrack_auto_managed_version( cls, repo_owner: str, repo_name: str, version: str ) -> None: - tracked = cls._read_auto_managed_versions(repo_owner, repo_name) - if version not in tracked: + """Stop auto-managing ``version`` (e.g. the user pinned it); keeps its files.""" + marker = cls._auto_managed_marker_dir(repo_owner, repo_name) / version + try: + marker.unlink() + except FileNotFoundError: return - tracked = [v for v in tracked if v != version] - cls._write_auto_managed_versions(repo_owner, repo_name, tracked) + except OSError as exc: + logging.warning( + "Could not untrack auto-managed frontend version %s: %s", + version, + exc, + ) @classmethod def get_required_frontend_version(cls) -> str: @@ -526,18 +467,13 @@ def init_frontend_unsafe( is_auto_managed = version in cls.AUTO_MANAGED_VERSION_SPECIFIERS if version.startswith("v"): - expected_path = str( - Path(cls.CUSTOM_FRONTENDS_ROOT) - / f"{repo_owner}_{repo_name}" - / version.lstrip("v") - ) + pinned_version = version.lstrip("v") + expected_path = str(cls._provider_dir(repo_owner, repo_name) / pinned_version) if os.path.exists(expected_path): logging.info( f"Using existing copy of specific frontend version tag: {repo_owner}/{repo_name}@{version}" ) - cls._untrack_auto_managed_version( - repo_owner, repo_name, version.lstrip("v") - ) + cls._untrack_auto_managed_version(repo_owner, repo_name, pinned_version) return expected_path logging.info( @@ -548,40 +484,47 @@ def init_frontend_unsafe( release = provider.get_release(version) semantic_version = release["tag_name"].lstrip("v") - web_root = str( - Path(cls.CUSTOM_FRONTENDS_ROOT) / provider.folder_name / semantic_version - ) - download_succeeded = os.path.exists(web_root) - if not download_succeeded: - try: - os.makedirs(web_root, exist_ok=True) - logging.info( - "Downloading frontend(%s) version(%s) to (%s)", - provider.folder_name, - semantic_version, - web_root, - ) - logging.debug(release) - download_release_asset_zip(release, destination_path=web_root) - download_succeeded = True - finally: - # Clean up the directory if it is empty, i.e. the download failed - if not os.listdir(web_root): - os.rmdir(web_root) - download_succeeded = False - - if download_succeeded: + web_root = str(cls._provider_dir(repo_owner, repo_name) / semantic_version) + + if cls._ensure_release_downloaded(provider, semantic_version, web_root, release): if is_auto_managed: - cls._prune_auto_managed_versions( - repo_owner, repo_name, semantic_version - ) + cls._prune_auto_managed_versions(repo_owner, repo_name, semantic_version) else: - cls._untrack_auto_managed_version( - repo_owner, repo_name, semantic_version - ) + cls._untrack_auto_managed_version(repo_owner, repo_name, semantic_version) return web_root + @classmethod + def _ensure_release_downloaded( + cls, + provider: "FrontEndProvider", + semantic_version: str, + web_root: str, + release: Release, + ) -> bool: + """Ensure ``release`` is present at ``web_root``. + + Returns True if the version is available on disk afterwards. A failed + download leaves no empty directory behind. + """ + if os.path.exists(web_root): + return True + try: + os.makedirs(web_root, exist_ok=True) + logging.info( + "Downloading frontend(%s) version(%s) to (%s)", + provider.folder_name, + semantic_version, + web_root, + ) + logging.debug(release) + download_release_asset_zip(release, destination_path=web_root) + finally: + # Clean up the directory if it is empty, i.e. the download failed + if not os.listdir(web_root): + os.rmdir(web_root) + return os.path.isdir(web_root) + @classmethod def init_frontend(cls, version_string: str) -> str: """ diff --git a/tests-unit/app_test/frontend_manager_test.py b/tests-unit/app_test/frontend_manager_test.py index aa2e35c3f45e..e681af976377 100644 --- a/tests-unit/app_test/frontend_manager_test.py +++ b/tests-unit/app_test/frontend_manager_test.py @@ -309,8 +309,10 @@ def _make_version_dir(root, owner, repo, version): return folder -def test_auto_managed_metadata_roundtrip(custom_frontends_root): - FrontendManager._write_auto_managed_versions("o", "r", ["1.0.0", "1.1.0", "1.0.0"]) +def test_auto_managed_markers_roundtrip(custom_frontends_root): + FrontendManager._mark_auto_managed("o", "r", "1.0.0") + FrontendManager._mark_auto_managed("o", "r", "1.1.0") + FrontendManager._mark_auto_managed("o", "r", "1.0.0") # idempotent assert FrontendManager._read_auto_managed_versions("o", "r") == ["1.0.0", "1.1.0"] @@ -318,61 +320,26 @@ def test_read_auto_managed_versions_missing(custom_frontends_root): assert FrontendManager._read_auto_managed_versions("o", "r") == [] -def test_read_auto_managed_versions_corrupt(custom_frontends_root): - provider_dir = custom_frontends_root / "o_r" - provider_dir.mkdir() - (provider_dir / FrontendManager.AUTO_MANAGED_METADATA_FILENAME).write_text( - "not json" - ) - assert FrontendManager._read_auto_managed_versions("o", "r") == [] - +def test_mark_auto_managed_does_not_create_version_dir(custom_frontends_root): + FrontendManager._mark_auto_managed("o", "r", "1.0.0") -@pytest.mark.parametrize("payload", ["[]", "null", '"oops"', "42"]) -def test_read_auto_managed_versions_non_dict_root(custom_frontends_root, payload): provider_dir = custom_frontends_root / "o_r" - provider_dir.mkdir() - (provider_dir / FrontendManager.AUTO_MANAGED_METADATA_FILENAME).write_text(payload) - assert FrontendManager._read_auto_managed_versions("o", "r") == [] - - -def test_read_auto_managed_versions_drops_unsafe_names(custom_frontends_root): - provider_dir = custom_frontends_root / "o_r" - provider_dir.mkdir() - (provider_dir / FrontendManager.AUTO_MANAGED_METADATA_FILENAME).write_text( - '{"auto_managed": ["1.0.0", "../escape", "/etc", "..", ".", "ok-1.2"]}' - ) - assert FrontendManager._read_auto_managed_versions("o", "r") == ["1.0.0", "ok-1.2"] - - -def test_write_auto_managed_versions_drops_unsafe_names(custom_frontends_root): - FrontendManager._write_auto_managed_versions( - "o", "r", ["1.0.0", "../escape", "/etc/passwd", "..", "."] - ) + # The marker is recorded, but no version directory is created as a side effect. + assert not (provider_dir / "1.0.0").exists() assert FrontendManager._read_auto_managed_versions("o", "r") == ["1.0.0"] -def test_prune_refuses_to_delete_outside_provider_dir( - custom_frontends_root, monkeypatch -): - sibling = custom_frontends_root / "do-not-touch" - sibling.mkdir() - (sibling / "marker").write_text("keep me") +def test_prune_only_touches_marked_siblings(custom_frontends_root): + # A sibling provider directory must never be affected by pruning. + sibling = _make_version_dir(custom_frontends_root, "other", "repo", "9.9.9") - provider_dir = custom_frontends_root / "o_r" - provider_dir.mkdir() _make_version_dir(custom_frontends_root, "o", "r", "1.0.0") - - monkeypatch.setattr( - FrontendManager, - "_read_auto_managed_versions", - classmethod(lambda cls, owner, repo: ["1.0.0", "../do-not-touch"]), - ) + FrontendManager._mark_auto_managed("o", "r", "1.0.0") FrontendManager._prune_auto_managed_versions("o", "r", keep_version="2.0.0") assert sibling.exists() - assert (sibling / "marker").read_text() == "keep me" - assert not (provider_dir / "1.0.0").exists() + assert not (custom_frontends_root / "o_r" / "1.0.0").exists() def test_prune_auto_managed_versions_removes_stale_and_keeps_pinned( @@ -381,7 +348,8 @@ def test_prune_auto_managed_versions_removes_stale_and_keeps_pinned( _make_version_dir(custom_frontends_root, "o", "r", "1.0.0") _make_version_dir(custom_frontends_root, "o", "r", "1.1.0") pinned = _make_version_dir(custom_frontends_root, "o", "r", "1.2.0") - FrontendManager._write_auto_managed_versions("o", "r", ["1.0.0", "1.1.0"]) + FrontendManager._mark_auto_managed("o", "r", "1.0.0") + FrontendManager._mark_auto_managed("o", "r", "1.1.0") FrontendManager._prune_auto_managed_versions("o", "r", keep_version="1.1.0") @@ -394,7 +362,8 @@ def test_prune_auto_managed_versions_removes_stale_and_keeps_pinned( def test_untrack_auto_managed_version_does_not_delete_folder(custom_frontends_root): version_dir = _make_version_dir(custom_frontends_root, "o", "r", "1.0.0") - FrontendManager._write_auto_managed_versions("o", "r", ["1.0.0", "1.1.0"]) + FrontendManager._mark_auto_managed("o", "r", "1.0.0") + FrontendManager._mark_auto_managed("o", "r", "1.1.0") FrontendManager._untrack_auto_managed_version("o", "r", "1.0.0") @@ -402,6 +371,11 @@ def test_untrack_auto_managed_version_does_not_delete_folder(custom_frontends_ro assert FrontendManager._read_auto_managed_versions("o", "r") == ["1.1.0"] +def test_untrack_auto_managed_version_missing_is_noop(custom_frontends_root): + FrontendManager._untrack_auto_managed_version("o", "r", "1.0.0") + assert FrontendManager._read_auto_managed_versions("o", "r") == [] + + def test_init_frontend_latest_prunes_previous_auto_managed_versions( custom_frontends_root, mock_provider, mock_releases ): @@ -409,9 +383,7 @@ def test_init_frontend_latest_prunes_previous_auto_managed_versions( pinned = _make_version_dir( custom_frontends_root, "test-owner", "test-repo", "1.1.5" ) - FrontendManager._write_auto_managed_versions( - "test-owner", "test-repo", ["1.0.0"] - ) + FrontendManager._mark_auto_managed("test-owner", "test-repo", "1.0.0") def fake_download(release, destination_path): Path(destination_path).mkdir(parents=True, exist_ok=True) @@ -439,9 +411,7 @@ def test_init_frontend_explicit_version_promotes_out_of_auto_managed( custom_frontends_root, mock_provider ): _make_version_dir(custom_frontends_root, "test-owner", "test-repo", "1.0.0") - FrontendManager._write_auto_managed_versions( - "test-owner", "test-repo", ["1.0.0"] - ) + FrontendManager._mark_auto_managed("test-owner", "test-repo", "1.0.0") result = FrontendManager.init_frontend_unsafe( "test-owner/test-repo@v1.0.0", mock_provider @@ -459,9 +429,7 @@ def test_init_frontend_explicit_no_v_prefix_promotes_out_of_auto_managed( custom_frontends_root, mock_provider ): _make_version_dir(custom_frontends_root, "test-owner", "test-repo", "1.0.0") - FrontendManager._write_auto_managed_versions( - "test-owner", "test-repo", ["1.0.0"] - ) + FrontendManager._mark_auto_managed("test-owner", "test-repo", "1.0.0") with patch( "app.frontend_management.download_release_asset_zip"