diff --git a/app/assets/api/routes.py b/app/assets/api/routes.py index d43485861..9f1fc9340 100644 --- a/app/assets/api/routes.py +++ b/app/assets/api/routes.py @@ -31,6 +31,7 @@ from app.assets.services import ( asset_exists, create_from_hash, delete_asset_reference, + delete_asset_reference_with_file, get_asset_detail, list_assets_page, list_tags, @@ -40,6 +41,10 @@ from app.assets.services import ( upload_from_temp_path, ) from app.assets.services.cursor import InvalidCursorError +from app.assets.services.file_location import ( + is_loopback_address, + reveal_file_in_file_manager, +) from app.assets.services.path_utils import compute_display_name from app.assets.services.tagging import list_tag_histogram @@ -636,20 +641,61 @@ async def update_asset_route(request: web.Request) -> web.Response: @_require_assets_feature_enabled async def delete_asset_route(request: web.Request) -> web.Response: reference_id = str(uuid.UUID(request.match_info["id"])) + owner_id = USER_MANAGER.get_request_user_id(request) + delete_content = request.query.get("delete_content") == "true" try: - # Deleting an asset is a soft delete of the reference; the underlying - # content is preserved (it may be shared with other references). - deleted = delete_asset_reference( - reference_id=reference_id, - owner_id=USER_MANAGER.get_request_user_id(request), - delete_content_if_orphan=False, + if delete_content: + detail = get_asset_detail(reference_id=reference_id, owner_id=owner_id) + if detail is None: + return _build_error_response( + 404, + "ASSET_NOT_FOUND", + f"AssetReference {reference_id} not found.", + ) + + allowed_roots = [] + if "input" in detail.tags: + allowed_roots.append(folder_paths.get_input_directory()) + if "output" in detail.tags: + allowed_roots.append(folder_paths.get_output_directory()) + + if not detail.ref.file_path or not any( + folder_paths.is_within_directory(root, detail.ref.file_path) + for root in allowed_roots + ): + return _build_error_response( + 403, + "ASSET_DELETE_FORBIDDEN", + "Only tagged files inside the input or output directory can be deleted with their content.", + ) + + deleted = delete_asset_reference_with_file( + reference_id=reference_id, + owner_id=owner_id, + staging_directory=folder_paths.get_temp_directory(), + expected_file_path=detail.ref.file_path, + allowed_directories=allowed_roots, + allow_ownerless=not user_manager.args.multi_user, + ) + else: + deleted = delete_asset_reference( + reference_id=reference_id, + owner_id=owner_id, + delete_content_if_orphan=False, + ) + except PermissionError as error: + return _build_error_response( + 403, + "ASSET_DELETE_FORBIDDEN", + str(error), + {"id": reference_id}, ) except Exception: logging.exception( "delete_asset_reference failed for reference_id=%s, owner_id=%s", reference_id, - USER_MANAGER.get_request_user_id(request), + owner_id, ) return _build_error_response(500, "INTERNAL", "Unexpected server error.") @@ -660,6 +706,69 @@ async def delete_asset_route(request: web.Request) -> web.Response: return web.Response(status=204) +@ROUTES.post(f"/api/assets/{{id:{UUID_RE}}}/open-location") +@_require_assets_feature_enabled +async def open_asset_location_route(request: web.Request) -> web.Response: + reference_id = str(uuid.UUID(request.match_info["id"])) + if not is_loopback_address(request.remote): + return _build_error_response( + 403, + "LOCAL_ACCESS_REQUIRED", + "Opening a file location is only available from the ComfyUI host machine.", + ) + + if request.headers.get("Sec-Fetch-Site") not in (None, "same-origin"): + return _build_error_response( + 403, + "CROSS_SITE_REQUEST_FORBIDDEN", + "Cross-site requests cannot open local file locations.", + ) + + detail = get_asset_detail( + reference_id=reference_id, + owner_id=USER_MANAGER.get_request_user_id(request), + ) + if detail is None: + return _build_error_response( + 404, + "ASSET_NOT_FOUND", + f"AssetReference {reference_id} not found.", + ) + + file_path = detail.ref.file_path + output_root = folder_paths.get_output_directory() + if ( + "output" not in detail.tags + or not file_path + or not folder_paths.is_within_directory(output_root, file_path) + ): + return _build_error_response( + 403, + "ASSET_LOCATION_FORBIDDEN", + "Only generated files inside the output directory can be opened.", + ) + + try: + reveal_file_in_file_manager(file_path) + except FileNotFoundError: + return _build_error_response( + 404, + "ASSET_FILE_NOT_FOUND", + "The generated source file no longer exists on disk.", + ) + except OSError: + logging.exception( + "Failed to reveal asset file for reference_id=%s", reference_id + ) + return _build_error_response( + 500, + "FILE_MANAGER_ERROR", + "Unable to open the system file manager.", + ) + + return web.Response(status=204) + + @ROUTES.get("/api/tags") @_require_assets_feature_enabled async def get_tags(request: web.Request) -> web.Response: diff --git a/app/assets/database/queries/__init__.py b/app/assets/database/queries/__init__.py index 9949e84e1..4e8c54137 100644 --- a/app/assets/database/queries/__init__.py +++ b/app/assets/database/queries/__init__.py @@ -11,6 +11,7 @@ from app.assets.database.queries.asset import ( from app.assets.database.queries.asset_reference import ( CacheStateRow, UnenrichedReferenceRow, + any_reference_exists_for_asset_id, bulk_insert_references_ignore_conflicts, bulk_update_enrichment_level, count_active_siblings, @@ -75,6 +76,7 @@ __all__ = [ "RemoveTagsResult", "SetTagsResult", "UnenrichedReferenceRow", + "any_reference_exists_for_asset_id", "add_missing_tag_for_asset_id", "add_tags_to_reference", "asset_exists_by_hash", diff --git a/app/assets/database/queries/asset_reference.py b/app/assets/database/queries/asset_reference.py index 126a0c9a4..37853a0c6 100644 --- a/app/assets/database/queries/asset_reference.py +++ b/app/assets/database/queries/asset_reference.py @@ -145,6 +145,19 @@ def reference_exists_for_asset_id( return session.execute(q).first() is not None +def any_reference_exists_for_asset_id( + session: Session, + asset_id: str, +) -> bool: + q = ( + select(sa.literal(True)) + .select_from(AssetReference) + .where(AssetReference.asset_id == asset_id) + .limit(1) + ) + return session.execute(q).first() is not None + + def reference_exists( session: Session, reference_id: str, diff --git a/app/assets/services/__init__.py b/app/assets/services/__init__.py index 03990966b..a7bdbc25b 100644 --- a/app/assets/services/__init__.py +++ b/app/assets/services/__init__.py @@ -1,6 +1,7 @@ from app.assets.services.asset_management import ( asset_exists, delete_asset_reference, + delete_asset_reference_with_file, get_asset_by_hash, get_asset_detail, list_assets_page, @@ -72,6 +73,7 @@ __all__ = [ "batch_insert_seed_assets", "create_from_hash", "delete_asset_reference", + "delete_asset_reference_with_file", "get_asset_by_hash", "get_asset_detail", "ingest_existing_file", diff --git a/app/assets/services/asset_management.py b/app/assets/services/asset_management.py index efdfd31a8..d508a4448 100644 --- a/app/assets/services/asset_management.py +++ b/app/assets/services/asset_management.py @@ -1,9 +1,13 @@ import contextlib +import logging import mimetypes import os +import uuid from datetime import timezone from typing import Sequence +import folder_paths + from app.assets.services.cursor import ( CursorPayload, InvalidCursorError, @@ -17,6 +21,7 @@ from app.assets.services.cursor import ( from app.assets.database.models import Asset from app.assets.database.queries import ( + any_reference_exists_for_asset_id, asset_exists_by_hash, reference_exists_for_asset_id, delete_reference_by_id, @@ -221,6 +226,114 @@ def delete_asset_reference( return True +def delete_asset_reference_with_file( + reference_id: str, + owner_id: str, + staging_directory: str, + expected_file_path: str, + allowed_directories: Sequence[str], + allow_ownerless: bool = False, +) -> bool: + with create_session() as session: + ref = get_reference_by_id(session, reference_id=reference_id) + if ref is None or ref.deleted_at is not None: + return False + + caller_owner_id = (owner_id or "").strip() + owns_reference = ref.owner_id == caller_owner_id + may_delete_ownerless = allow_ownerless and ref.owner_id == "" + if not (owns_reference or may_delete_ownerless): + raise PermissionError( + "Only the owning user can delete an asset's source file." + ) + + asset_id = ref.asset_id + file_path = ref.file_path + if not file_path or os.path.realpath(file_path) != os.path.realpath( + expected_file_path + ): + raise PermissionError( + "The asset source path changed before it could be deleted." + ) + if not any( + folder_paths.is_within_directory(directory, file_path) + for directory in allowed_directories + ): + raise PermissionError( + "The asset source file is outside the directories allowed for deletion." + ) + + deleted = delete_reference_by_id( + session, reference_id=reference_id, owner_id=owner_id + ) + if not deleted: + session.rollback() + return False + + staged_file_path: str | None = None + if file_path and os.path.isfile(file_path): + source_directory = os.path.dirname(file_path) + os.makedirs(staging_directory, exist_ok=True) + staged_file_path = os.path.join( + staging_directory, f".comfy-delete-{uuid.uuid4().hex}.tmp" + ) + if not any( + folder_paths.is_within_directory(directory, file_path) + for directory in allowed_directories + ): + raise PermissionError( + "The asset source file moved outside the directories allowed for deletion." + ) + try: + os.replace(file_path, staged_file_path) + except OSError: + staged_file_path = os.path.join( + source_directory, f".comfy-delete-{uuid.uuid4().hex}.tmp" + ) + if not any( + folder_paths.is_within_directory(directory, file_path) + for directory in allowed_directories + ): + raise PermissionError( + "The asset source file moved outside the directories allowed for deletion." + ) + os.replace(file_path, staged_file_path) + + try: + session.flush() + if not any_reference_exists_for_asset_id(session, asset_id=asset_id): + asset = session.get(Asset, asset_id) + if asset is not None: + session.delete(asset) + session.commit() + except Exception: + session.rollback() + if ( + staged_file_path + and file_path + and os.path.isfile(staged_file_path) + and not os.path.exists(file_path) + ): + try: + os.replace(staged_file_path, file_path) + except OSError: + logging.exception( + "Failed to restore staged asset file after rollback: %s", + file_path, + ) + raise + + if staged_file_path: + try: + os.remove(staged_file_path) + except OSError: + logging.exception( + "Failed to remove staged asset file after commit: %s", + staged_file_path, + ) + return True + + def set_asset_preview( reference_id: str, preview_reference_id: str | None = None, diff --git a/app/assets/services/file_location.py b/app/assets/services/file_location.py new file mode 100644 index 000000000..3b17e4e72 --- /dev/null +++ b/app/assets/services/file_location.py @@ -0,0 +1,46 @@ +import ipaddress +import os +import subprocess +import sys + + +def is_loopback_address(address: str | None) -> bool: + if not address: + return False + + try: + parsed = ipaddress.ip_address(address.split("%", 1)[0]) + except ValueError: + return False + + if parsed.is_loopback: + return True + return bool( + parsed.version == 6 and parsed.ipv4_mapped and parsed.ipv4_mapped.is_loopback + ) + + +def build_file_manager_command( + file_path: str, platform: str | None = None +) -> list[str]: + normalized_path = os.path.abspath(file_path) + current_platform = platform or sys.platform + + if current_platform == "win32": + return ["explorer.exe", f"/select,{normalized_path}"] + if current_platform == "darwin": + return ["open", "-R", normalized_path] + return ["xdg-open", os.path.dirname(normalized_path)] + + +def reveal_file_in_file_manager(file_path: str) -> None: + normalized_path = os.path.abspath(file_path) + if not os.path.isfile(normalized_path): + raise FileNotFoundError(normalized_path) + + subprocess.Popen( + build_file_manager_command(normalized_path), + close_fds=True, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) diff --git a/tests-unit/assets_test/services/test_asset_management.py b/tests-unit/assets_test/services/test_asset_management.py index e8ff989e9..50dca936b 100644 --- a/tests-unit/assets_test/services/test_asset_management.py +++ b/tests-unit/assets_test/services/test_asset_management.py @@ -1,4 +1,5 @@ """Tests for asset_management services.""" +import folder_paths import pytest from sqlalchemy.orm import Session @@ -6,6 +7,7 @@ from app.assets.database.models import Asset, AssetReference from app.assets.database.queries import ensure_tags_exist, add_tags_to_reference from app.assets.helpers import get_utc_now from app.assets.services import ( + delete_asset_reference_with_file, get_asset_detail, update_asset_metadata, delete_asset_reference, @@ -215,6 +217,209 @@ class TestDeleteAssetReference: assert session.get(Asset, asset_id) is None +class TestDeleteAssetReferenceWithFile: + def test_deletes_only_selected_reference_file( + self, mock_create_session, session: Session, temp_dir + ): + selected_file = temp_dir / "selected.bin" + retained_file = temp_dir / "retained.bin" + selected_file.write_bytes(b"same-content") + retained_file.write_bytes(b"same-content") + + asset = _make_asset(session) + selected_ref = _make_reference(session, asset, name=selected_file.name) + selected_ref.file_path = str(selected_file) + retained_ref = _make_reference(session, asset, name=retained_file.name) + retained_ref.file_path = str(retained_file) + asset_id = asset.id + session.commit() + + result = delete_asset_reference_with_file( + reference_id=selected_ref.id, + owner_id="", + staging_directory=str(temp_dir / "staging"), + expected_file_path=str(selected_file), + allowed_directories=[str(temp_dir)], + ) + + assert result is True + assert not selected_file.exists() + assert retained_file.exists() + assert session.get(Asset, asset_id) is not None + assert session.get(AssetReference, retained_ref.id) is not None + + def test_rejects_ownerless_reference_without_explicit_permission( + self, mock_create_session, session: Session, temp_dir + ): + selected_file = temp_dir / "selected.bin" + selected_file.write_bytes(b"content") + + asset = _make_asset(session) + selected_ref = _make_reference(session, asset, owner_id="") + selected_ref.file_path = str(selected_file) + selected_ref_id = selected_ref.id + session.commit() + + with pytest.raises(PermissionError, match="owning user"): + delete_asset_reference_with_file( + reference_id=selected_ref_id, + owner_id="another-user", + staging_directory=str(temp_dir / "staging"), + expected_file_path=str(selected_file), + allowed_directories=[str(temp_dir)], + ) + + assert selected_file.exists() + assert session.get(AssetReference, selected_ref_id) is not None + + def test_allows_ownerless_reference_in_single_user_mode( + self, mock_create_session, session: Session, temp_dir + ): + selected_file = temp_dir / "selected.bin" + selected_file.write_bytes(b"content") + + asset = _make_asset(session) + selected_ref = _make_reference(session, asset, owner_id="") + selected_ref.file_path = str(selected_file) + selected_ref_id = selected_ref.id + session.commit() + + result = delete_asset_reference_with_file( + reference_id=selected_ref_id, + owner_id="default", + staging_directory=str(temp_dir / "staging"), + expected_file_path=str(selected_file), + allowed_directories=[str(temp_dir)], + allow_ownerless=True, + ) + + assert result is True + assert not selected_file.exists() + assert session.get(AssetReference, selected_ref_id) is None + + def test_preserves_soft_deleted_shared_reference( + self, mock_create_session, session: Session, temp_dir + ): + selected_file = temp_dir / "selected.bin" + retained_file = temp_dir / "retained.bin" + selected_file.write_bytes(b"same-content") + retained_file.write_bytes(b"same-content") + + asset = _make_asset(session) + selected_ref = _make_reference(session, asset, name=selected_file.name) + selected_ref.file_path = str(selected_file) + retained_ref = _make_reference(session, asset, name=retained_file.name) + retained_ref.file_path = str(retained_file) + retained_ref.deleted_at = get_utc_now() + asset_id = asset.id + retained_ref_id = retained_ref.id + session.commit() + + result = delete_asset_reference_with_file( + reference_id=selected_ref.id, + owner_id="", + staging_directory=str(temp_dir / "staging"), + expected_file_path=str(selected_file), + allowed_directories=[str(temp_dir)], + ) + + assert result is True + assert not selected_file.exists() + assert retained_file.exists() + assert session.get(Asset, asset_id) is not None + assert session.get(AssetReference, retained_ref_id) is not None + + def test_restores_file_when_commit_fails( + self, mock_create_session, session: Session, temp_dir, monkeypatch + ): + selected_file = temp_dir / "selected.bin" + selected_file.write_bytes(b"content") + + asset = _make_asset(session) + selected_ref = _make_reference(session, asset, name=selected_file.name) + selected_ref.file_path = str(selected_file) + selected_ref_id = selected_ref.id + session.commit() + + def fail_commit(_session): + raise RuntimeError("commit failed") + + monkeypatch.setattr(Session, "commit", fail_commit) + + with pytest.raises(RuntimeError, match="commit failed"): + delete_asset_reference_with_file( + reference_id=selected_ref_id, + owner_id="", + staging_directory=str(temp_dir / "staging"), + expected_file_path=str(selected_file), + allowed_directories=[str(temp_dir)], + ) + + session.expire_all() + assert selected_file.read_bytes() == b"content" + assert session.get(AssetReference, selected_ref_id) is not None + + def test_rejects_a_source_path_changed_after_authorization( + self, mock_create_session, session: Session, temp_dir + ): + selected_file = temp_dir / "selected.bin" + replacement_file = temp_dir / "replacement.bin" + selected_file.write_bytes(b"selected") + replacement_file.write_bytes(b"replacement") + + asset = _make_asset(session) + selected_ref = _make_reference(session, asset, name=selected_file.name) + selected_ref.file_path = str(replacement_file) + selected_ref_id = selected_ref.id + session.commit() + + with pytest.raises(PermissionError, match="source path changed"): + delete_asset_reference_with_file( + reference_id=selected_ref_id, + owner_id="", + staging_directory=str(temp_dir / "staging"), + expected_file_path=str(selected_file), + allowed_directories=[str(temp_dir)], + ) + + assert selected_file.exists() + assert replacement_file.exists() + assert session.get(AssetReference, selected_ref_id) is not None + + def test_rechecks_containment_immediately_before_staging( + self, mock_create_session, session: Session, temp_dir, monkeypatch + ): + selected_file = temp_dir / "managed" / "selected.bin" + selected_file.parent.mkdir() + selected_file.write_bytes(b"content") + + asset = _make_asset(session) + selected_ref = _make_reference(session, asset, name=selected_file.name) + selected_ref.file_path = str(selected_file) + selected_ref_id = selected_ref.id + session.commit() + + containment_results = iter([True, False]) + monkeypatch.setattr( + folder_paths, + "is_within_directory", + lambda *_args: next(containment_results), + ) + + with pytest.raises(PermissionError, match="moved outside"): + delete_asset_reference_with_file( + reference_id=selected_ref_id, + owner_id="", + staging_directory=str(temp_dir / "staging"), + expected_file_path=str(selected_file), + allowed_directories=[str(temp_dir / "managed")], + ) + + session.expire_all() + assert selected_file.read_bytes() == b"content" + assert session.get(AssetReference, selected_ref_id) is not None + + class TestSetAssetPreview: def test_sets_preview(self, mock_create_session, session: Session): asset = _make_asset(session, hash_val="blake3:main") diff --git a/tests-unit/assets_test/services/test_file_location.py b/tests-unit/assets_test/services/test_file_location.py new file mode 100644 index 000000000..5380df36e --- /dev/null +++ b/tests-unit/assets_test/services/test_file_location.py @@ -0,0 +1,69 @@ +import subprocess + +import pytest + +from app.assets.services.file_location import ( + build_file_manager_command, + is_loopback_address, + reveal_file_in_file_manager, +) + + +@pytest.mark.parametrize( + "address", + ["127.0.0.1", "127.42.0.9", "::1", "::ffff:127.0.0.1"], +) +def test_accepts_loopback_addresses(address): + assert is_loopback_address(address) + + +@pytest.mark.parametrize( + "address", + [None, "", "localhost", "0.0.0.0", "192.168.1.20", "::", "fe80::1"], +) +def test_rejects_non_loopback_addresses(address): + assert not is_loopback_address(address) + + +def test_builds_platform_commands(tmp_path): + source = tmp_path / "folder with spaces" / "render.mp4" + source.parent.mkdir() + source.write_bytes(b"video") + + assert build_file_manager_command(str(source), "win32") == [ + "explorer.exe", + f"/select,{source}", + ] + assert build_file_manager_command(str(source), "darwin") == [ + "open", + "-R", + str(source), + ] + assert build_file_manager_command(str(source), "linux") == [ + "xdg-open", + str(source.parent), + ] + + +def test_reveal_launches_without_a_shell(tmp_path, monkeypatch): + source = tmp_path / "render.png" + source.write_bytes(b"image") + calls = [] + + def capture_popen(*args, **kwargs): + calls.append((args, kwargs)) + + monkeypatch.setattr(subprocess, "Popen", capture_popen) + + reveal_file_in_file_manager(str(source)) + + assert len(calls) == 1 + args, kwargs = calls[0] + assert args[0] + assert kwargs["close_fds"] is True + assert "shell" not in kwargs + + +def test_reveal_rejects_missing_files(tmp_path): + with pytest.raises(FileNotFoundError): + reveal_file_in_file_manager(str(tmp_path / "missing.mp4")) diff --git a/tests-unit/assets_test/test_asset_location_route.py b/tests-unit/assets_test/test_asset_location_route.py new file mode 100644 index 000000000..9ea715593 --- /dev/null +++ b/tests-unit/assets_test/test_asset_location_route.py @@ -0,0 +1,183 @@ +import asyncio +import json +import uuid +from types import SimpleNamespace +from unittest.mock import Mock + +from app.assets.api import routes + + +def make_request(remote="127.0.0.1", headers=None, query=None): + return SimpleNamespace( + match_info={"id": str(uuid.uuid4())}, + remote=remote, + headers=headers or {}, + query=query or {}, + ) + + +def run_route(request): + return asyncio.run(routes.open_asset_location_route.__wrapped__(request)) + + +def run_delete_route(request): + return asyncio.run(routes.delete_asset_route.__wrapped__(request)) + + +def test_open_location_rejects_lan_clients(monkeypatch): + get_detail = Mock() + monkeypatch.setattr(routes, "get_asset_detail", get_detail) + + response = run_route(make_request(remote="192.168.1.50")) + + assert response.status == 403 + assert json.loads(response.text)["error"]["code"] == "LOCAL_ACCESS_REQUIRED" + get_detail.assert_not_called() + + +def test_open_location_rejects_cross_site_requests(monkeypatch): + get_detail = Mock() + monkeypatch.setattr(routes, "get_asset_detail", get_detail) + + response = run_route(make_request(headers={"Sec-Fetch-Site": "cross-site"})) + + assert response.status == 403 + assert json.loads(response.text)["error"]["code"] == "CROSS_SITE_REQUEST_FORBIDDEN" + get_detail.assert_not_called() + + +def test_open_location_reveals_managed_generated_file(tmp_path, monkeypatch): + source = tmp_path / "output" / "video" / "render.mp4" + source.parent.mkdir(parents=True) + source.write_bytes(b"video") + reveal = Mock() + + monkeypatch.setattr( + routes, + "USER_MANAGER", + SimpleNamespace(get_request_user_id=lambda _request: "default"), + ) + monkeypatch.setattr( + routes, + "get_asset_detail", + lambda **_kwargs: SimpleNamespace( + tags=["output"], ref=SimpleNamespace(file_path=str(source)) + ), + ) + monkeypatch.setattr( + routes.folder_paths, + "get_output_directory", + lambda: str(tmp_path / "output"), + ) + monkeypatch.setattr(routes, "reveal_file_in_file_manager", reveal) + + response = run_route(make_request()) + + assert response.status == 204 + reveal.assert_called_once_with(str(source)) + + +def test_open_location_rejects_non_output_assets(tmp_path, monkeypatch): + source = tmp_path / "input" / "image.png" + source.parent.mkdir() + source.write_bytes(b"image") + + monkeypatch.setattr( + routes, + "USER_MANAGER", + SimpleNamespace(get_request_user_id=lambda _request: "default"), + ) + monkeypatch.setattr( + routes, + "get_asset_detail", + lambda **_kwargs: SimpleNamespace( + tags=["input"], ref=SimpleNamespace(file_path=str(source)) + ), + ) + monkeypatch.setattr( + routes.folder_paths, + "get_output_directory", + lambda: str(tmp_path / "output"), + ) + + response = run_route(make_request()) + + assert response.status == 403 + assert json.loads(response.text)["error"]["code"] == "ASSET_LOCATION_FORBIDDEN" + + +def test_delete_content_rejects_paths_outside_managed_root(tmp_path, monkeypatch): + source = tmp_path / "outside" / "render.png" + source.parent.mkdir() + source.write_bytes(b"image") + delete_with_file = Mock() + + monkeypatch.setattr( + routes, + "USER_MANAGER", + SimpleNamespace(get_request_user_id=lambda _request: "default"), + ) + monkeypatch.setattr( + routes, + "get_asset_detail", + lambda **_kwargs: SimpleNamespace( + tags=["output"], ref=SimpleNamespace(file_path=str(source)) + ), + ) + monkeypatch.setattr( + routes.folder_paths, + "get_output_directory", + lambda: str(tmp_path / "output"), + ) + monkeypatch.setattr(routes, "delete_asset_reference_with_file", delete_with_file) + + response = run_delete_route(make_request(query={"delete_content": "true"})) + + assert response.status == 403 + assert json.loads(response.text)["error"]["code"] == "ASSET_DELETE_FORBIDDEN" + delete_with_file.assert_not_called() + + +def test_delete_content_uses_guarded_file_service(tmp_path, monkeypatch): + source = tmp_path / "output" / "render.png" + source.parent.mkdir() + source.write_bytes(b"image") + delete_with_file = Mock(return_value=True) + + monkeypatch.setattr( + routes, + "USER_MANAGER", + SimpleNamespace(get_request_user_id=lambda _request: "default"), + ) + monkeypatch.setattr( + routes, + "get_asset_detail", + lambda **_kwargs: SimpleNamespace( + tags=["output"], ref=SimpleNamespace(file_path=str(source)) + ), + ) + monkeypatch.setattr( + routes.folder_paths, + "get_output_directory", + lambda: str(tmp_path / "output"), + ) + monkeypatch.setattr( + routes.folder_paths, + "get_temp_directory", + lambda: str(tmp_path / "temp"), + ) + monkeypatch.setattr(routes.user_manager.args, "multi_user", False) + monkeypatch.setattr(routes, "delete_asset_reference_with_file", delete_with_file) + + request = make_request(query={"delete_content": "true"}) + response = run_delete_route(request) + + assert response.status == 204 + delete_with_file.assert_called_once_with( + reference_id=request.match_info["id"], + owner_id="default", + staging_directory=str(tmp_path / "temp"), + expected_file_path=str(source), + allowed_directories=[str(tmp_path / "output")], + allow_ownerless=True, + )