From aa7741000aa21a3c44ce854eccc0717288c30f4f Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Sun, 18 Oct 2026 11:27:24 -0700 Subject: [PATCH] feat(history): resolve reported outputs through current authorization (#209) --- database/history_gallery.py | 127 ++++ py/api/history.py | 411 ++++++++++- py/api/history_current.py | 399 ++++++++++ py/api/images.py | 71 +- py/api/inspection_service.py | 72 ++ tests/test_api_history_outputs.py | 1090 ++++++++++++++++++++++++++++ tests/test_api_images.py | 2 +- tests/test_api_prompt_inspector.py | 4 +- tests/test_history_current.py | 469 ++++++++++++ tests/test_inspection_response.py | 42 ++ utils/inspection_response.py | 53 +- 11 files changed, 2671 insertions(+), 69 deletions(-) create mode 100644 database/history_gallery.py create mode 100644 py/api/history_current.py create mode 100644 py/api/inspection_service.py create mode 100644 tests/test_api_history_outputs.py create mode 100644 tests/test_history_current.py diff --git a/database/history_gallery.py b/database/history_gallery.py new file mode 100644 index 00000000..08d83240 --- /dev/null +++ b/database/history_gallery.py @@ -0,0 +1,127 @@ +"""Bounded exact-path saved display within the caller's history snapshot.""" + +import os +import sqlite3 + +try: + from ..utils.inspection_response import MAX_LIBRARY_BYTES, _library_projection +except ImportError: + from utils.inspection_response import MAX_LIBRARY_BYTES, _library_projection + + +def _display_bytes(value: bytes | None) -> str | None: + if value is None: + return None + return value.decode("utf-8", "replace") + + +def _tags( + conn: sqlite3.Connection, owner: int, *, pair: bool +) -> tuple[list[str], bool]: + table, column = ( + ("prompt_pair_tags", "pair_id") if pair else ("prompt_tags", "prompt_id") + ) + tags = [] + used = 0 + cursor = conn.execute( + f"SELECT substr(CAST(t.name AS BLOB),1,?) FROM {table} pt " + f"JOIN tags t ON t.id=pt.tag_id WHERE pt.{column}=? LIMIT 129", + (MAX_LIBRARY_BYTES + 1, owner), + ) + for row in cursor: + used += len(row[0]) + if used > MAX_LIBRARY_BYTES or len(tags) == 128: + return [], True + tags.append(_display_bytes(row[0])) + return tags, False + + +def _saved_projection(conn: sqlite3.Connection, row: sqlite3.Row) -> tuple[dict, dict]: + saved = {"prompt_id": row["prompt_id"]} + pairs = list( + conn.execute( + "SELECT pp.id,pp.positive_prompt_id," + "substr(CAST(pp.title AS BLOB),1,?) AS title," + "substr(CAST(pp.category AS BLOB),1,?) AS category " + "FROM image_prompt_pairs ip JOIN prompt_pairs pp ON pp.id=ip.pair_id " + "WHERE ip.image_id=? LIMIT 2", + (MAX_LIBRARY_BYTES + 1, MAX_LIBRARY_BYTES + 1, row["id"]), + ) + ) + if len(pairs) > 1: + return saved, _library_projection({"status": "limited"}) + pair = pairs[0] if pairs else None + title = pair["title"] if pair is not None else None + category = pair["category"] if pair is not None else row["category"] + tags, limited = _tags( + conn, + pair["id"] if pair is not None else row["prompt_id"], + pair=pair is not None, + ) + limited = limited or any( + value is not None and len(value) > MAX_LIBRARY_BYTES + for value in (title, category) + ) + if pair is not None: + saved["pair"] = { + "id": pair["id"], + "positive_prompt_id": pair["positive_prompt_id"], + } + library = _library_projection( + { + "status": "linked", + "positive_id": row["prompt_id"], + "pair_id": pair["id"] if pair is not None else None, + "title": _display_bytes(title), + "category": _display_bytes(category), + "tags": tags, + "metadata_scope": "pair" if pair is not None else "positive", + } + ) + if limited: + library = _library_projection({"status": "limited", "tags": []}) + return saved, library + + +def read_history_gallery( + conn: sqlite3.Connection, paths: tuple[str, ...] +) -> dict[str, dict]: + """Read at most two rows/path and two pair associations/unique row; no writes.""" + if not conn.in_transaction: + raise sqlite3.ProgrammingError("Saved history display requires caller snapshot") + if ( + type(paths) is not tuple + or len(paths) > 100 + or any( + type(path) is not str + or not os.path.isabs(path) + or path != os.path.normcase(os.path.normpath(path)) + or "\0" in path + for path in paths + ) + ): + raise ValueError("Invalid history gallery paths") + result = {} + for path in dict.fromkeys(paths): + rows = list( + conn.execute( + "SELECT gi.id,gi.prompt_id," + "substr(CAST(p.category AS BLOB),1,?) AS category " + "FROM generated_images gi JOIN prompts p ON p.id=gi.prompt_id " + "WHERE gi.file_path=? OR gi.image_path=? LIMIT 2", + (MAX_LIBRARY_BYTES + 1, path, path), + ) + ) + match = "none" if not rows else "unique" if len(rows) == 1 else "ambiguous" + saved, library = None, _library_projection( + {"status": "none" if not rows else "ambiguous"} + ) + if match == "unique": + saved, library = _saved_projection(conn, rows[0]) + result[path] = { + "gallery_match": match, + "gallery_image_id": rows[0]["id"] if match == "unique" else None, + "saved": saved, + "library": library, + } + return result diff --git a/py/api/history.py b/py/api/history.py index 63a465a6..32d82703 100644 --- a/py/api/history.py +++ b/py/api/history.py @@ -1,5 +1,6 @@ """Authenticated chronological windows and bounded frozen run details.""" +import asyncio import base64 import binascii from functools import wraps @@ -7,22 +8,51 @@ import json import re import sqlite3 from typing import Callable +from urllib.parse import urlencode from aiohttp import web from .library import _decode, _encode, _scope +from .history_current import ( + CurrentLocationUnavailable, + CurrentOutputScope, + VIDEO_EXTENSIONS, + _OpenedCurrentMedia, + _current_media_response, + _prepare_current_media, + _reauthorize_current_target, + candidate_locator, + resolve_current_candidate, + snapshot_current_output_scope, +) +from .inspection_service import CurrentFileChanged, inspect_current_file try: from ...database.execution_history import HistoryEpochChanged - from ...database.history_reads import RunKey, WindowPosition + from ...database.history_gallery import read_history_gallery + from ...database.history_reads import ( + HistoryCaptureChanged, + OutputPosition, + RunKey, + WindowPosition, + ) from ...utils.history_read_fence import HistoryReadFence + from ...utils.inspection_response import encode_history_inspection_response except ImportError: from database.execution_history import HistoryEpochChanged - from database.history_reads import RunKey, WindowPosition + from database.history_gallery import read_history_gallery + from database.history_reads import ( + HistoryCaptureChanged, + OutputPosition, + RunKey, + WindowPosition, + ) from utils.history_read_fence import HistoryReadFence + from utils.inspection_response import encode_history_inspection_response _RUN_MAX_BYTES = 8 * 1024 * 1024 +_OUTPUTS_MAX_BYTES = 1024 * 1024 _UUID = re.compile(r"[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}") _DIGEST = re.compile(r"[0-9a-f]{64}") _DECIMAL = re.compile(r"0|[1-9][0-9]{0,18}") @@ -40,6 +70,9 @@ _MESSAGES = { "history-changed": "Recorded history changed; reload recorded history", "storage-unavailable": "Recorded history is temporarily unavailable", "response-limit": "Recorded history response exceeds the response limit", + "capture-changed": "Reported output capture changed; reload reported outputs", + "candidate-stale": "Current output locations changed; reload reported outputs", + "not-authorized": "Current output media is unavailable or not authorized", } @@ -197,6 +230,129 @@ def _encode_window_cursor( ) +def _node(value: object) -> str: + if type(value) is not str or not 1 <= len(value) <= 128 or "\0" in value: + raise ValueError("Invalid output node") + if len(value.encode("utf-8")) > 512: + raise ValueError("Invalid output node size") + return value + + +def _decode_output_cursor( + token: str | None, run_id: str, epoch: str +) -> OutputPosition | None: + if token is None: + return None + try: + value = _cursor_payload(token, "history-outputs") + if type(value) is not dict or set(value) != { + "version", + "epoch", + "run_id", + "capture_fingerprint", + "after", + }: + raise ValueError + if ( + type(value["version"]) is not int + or value["version"] != 1 + or _uuid(value["run_id"]) != run_id + or type(value["capture_fingerprint"]) is not str + or _DIGEST.fullmatch(value["capture_fingerprint"]) is None + or type(value["after"]) is not list + or len(value["after"]) != 2 + or type(value["after"][1]) is not int + or not 0 <= value["after"][1] <= 255 + ): + raise ValueError + node = _node(value["after"][0]) + if _uuid(value["epoch"]) != epoch: + raise HistoryReadError( + 409, "history-changed", {"action": "refresh-history"} + ) + return OutputPosition(value["capture_fingerprint"], (node, value["after"][1])) + except (ValueError, UnicodeError): + raise HistoryReadError(400, "invalid-cursor") from None + + +def _encode_output_cursor( + position: OutputPosition | None, run_id: str, epoch: str +) -> str | None: + if position is None: + return None + return _encode( + { + "version": 1, + "epoch": epoch, + "run_id": run_id, + "capture_fingerprint": position.capture_fingerprint, + "after": list(position.after), + }, + _scope({"kind": "history-outputs"}), + ) + + +def _current_stale() -> HistoryReadError: + return HistoryReadError(409, "candidate-stale", {"action": "refresh-outputs"}) + + +def _current_unavailable(reason: str) -> HistoryReadError: + if reason == "temporary_unavailable": + return HistoryReadError(503, "storage-unavailable", {"action": "retry-read"}) + if reason == "missing": + return HistoryReadError(404, "not-found", {"action": "retry-current-file"}) + return HistoryReadError(403, "not-authorized", {"action": "retry-current-file"}) + + +def _check_current( + scope: CurrentOutputScope, + api: object, + fence: HistoryReadFence, + epoch: str, + generation: int, +) -> None: + if snapshot_current_output_scope(api).current_scope != scope.current_scope: + raise _current_stale() + fence.validate(epoch, generation) + + +def _media_descriptor(receipt: dict, candidate: dict, epoch: str) -> dict: + occurrence = receipt["occurrence"] + query = urlencode( + { + "node": occurrence["output_node_id"], + "ordinal": occurrence["image_ordinal"], + "epoch": epoch, + "current_scope": candidate["current_scope"], + } + ) + base = f"/prompt_manager/history/runs/{occurrence['run_id']}/output/current" + return { + "kind": "reported-location", + "occurrence": occurrence, + "history_epoch": epoch, + "current_scope": candidate["current_scope"], + "location_key": candidate["location_key"], + "reference": receipt["reference"], + "filename": receipt["reference"]["filename"], + "is_video": receipt["reference"]["filename"] + .lower() + .endswith(tuple(VIDEO_EXTENSIONS)), + "url": f"{base}/media?{query}", + "thumbnail_url": f"{base}/thumbnail?{query}", + "inspection_url": f"{base}/inspection?{query}", + } + + +def _close_prepared_result(task: asyncio.Task) -> None: + try: + result = task.result() + except (Exception, asyncio.CancelledError): + return + if not isinstance(result, bytes): + result.close() + + def _wire_run(run: dict) -> None: for key in _RUN_DECIMALS: if run[key] is not None: @@ -229,6 +385,18 @@ class HistoryRoutesMixin: self.get_library_history ) routes.get("/prompt_manager/history/runs/{run_id}")(self.get_history_run) + routes.get("/prompt_manager/history/runs/{run_id}/outputs")( + self.get_history_outputs + ) + routes.get("/prompt_manager/history/runs/{run_id}/output/current/media")( + self.get_history_current_media + ) + routes.get("/prompt_manager/history/runs/{run_id}/output/current/thumbnail")( + self.get_history_current_thumbnail + ) + routes.get("/prompt_manager/history/runs/{run_id}/output/current/inspection")( + self.get_history_current_inspection + ) def _history_fence(self) -> HistoryReadFence: # Called synchronously by the event-loop owner before scheduling work. @@ -325,3 +493,242 @@ class HistoryRoutesMixin: return encoded return _success_response(await self._run_in_executor(project)) + + def _history_gallery( + self, conn: sqlite3.Connection, paths: tuple[str, ...] + ) -> tuple[dict, bool]: + try: + return read_history_gallery(conn, paths), False + except Exception as error: + self.logger.warning( + "Saved history display unavailable (%s)", type(error).__name__ + ) + return {}, True + + @_history_errors + async def get_history_outputs(self, request: web.Request) -> web.Response: + _query(request) + run_id = _uuid(request.match_info["run_id"]) + expected_epoch = _uuid(request.query.get("epoch")) + limit = _page_limit(request) + position = _decode_output_cursor( + request.query.get("cursor"), run_id, expected_epoch + ) + fence = self._history_fence() + db = self.db + + def project() -> bytes: + generation = fence.begin() + scope = snapshot_current_output_scope(self) + try: + with db.recorded_history_snapshot(expected_epoch) as epoch: + try: + result = db.list_history_outputs( + run_id, limit=limit, after=position + ) + except HistoryCaptureChanged: + raise HistoryReadError( + 409, "capture-changed", {"action": "refresh-outputs"} + ) from None + if result is not None: + paths = tuple( + dict.fromkeys( + path + for item in result["items"] + if (path := candidate_locator(item["reference"], scope)) + is not None + ) + ) + gallery, saved_unavailable = self._history_gallery( + db.model.get_connection(), paths + ) + finally: + db.model.close() + if result is None: + _check_current(scope, self, fence, epoch, generation) + raise HistoryReadError(404, "not-found") + resolved = {} + for receipt in result["items"]: + reference = receipt["reference"] + locator = tuple(reference.items()) + if locator not in resolved: + try: + if saved_unavailable: + raise sqlite3.OperationalError( + "Optional saved display unavailable" + ) + resolved[locator] = resolve_current_candidate( + reference, scope, gallery + ).public + except Exception: + resolved[locator] = { + "status": "error", + "reason": "temporary_unavailable", + "evidence_class": "locator_candidate", + } + candidate = dict(resolved[locator]) + if candidate["status"] == "available": + candidate["media"] = _media_descriptor(receipt, candidate, epoch) + receipt["current_candidate"] = candidate + result["next_cursor"] = _encode_output_cursor( + result.pop("next_position"), run_id, epoch + ) + encoded = _serialize_history( + { + "success": True, + "history_version": 1, + "history_epoch": epoch, + **result, + } + ) + _check_current(scope, self, fence, epoch, generation) + if len(encoded) > _OUTPUTS_MAX_BYTES: + recovery = ( + {"action": "retry-outputs", "limit": max(1, limit // 2)} + if limit > 1 + else None + ) + raise HistoryReadError(422, "response-limit", recovery) + return encoded + + return _success_response(await self._run_in_executor(project)) + + @_history_errors + async def get_history_current_media( + self, request: web.Request + ) -> web.StreamResponse: + return await self._history_current(request, "media") + + @_history_errors + async def get_history_current_thumbnail( + self, request: web.Request + ) -> web.StreamResponse: + return await self._history_current(request, "thumbnail") + + @_history_errors + async def get_history_current_inspection( + self, request: web.Request + ) -> web.Response: + return await self._history_current(request, "inspection") + + async def _history_current( + self, request: web.Request, kind: str + ) -> web.StreamResponse: + _query(request) + run_id = _uuid(request.match_info["run_id"]) + node = _node(request.query.get("node")) + ordinal = _decimal(request.query.get("ordinal")) + if ordinal > 255: + raise ValueError("Invalid output ordinal") + expected_epoch = _uuid(request.query.get("epoch")) + current_scope = request.query.get("current_scope") + if current_scope is None or _DIGEST.fullmatch(current_scope) is None: + raise ValueError("Invalid current output scope") + fence = self._history_fence() + db = self.db + + def project() -> bytes | _OpenedCurrentMedia: + generation = fence.begin() + scope = snapshot_current_output_scope(self) + try: + with db.recorded_history_snapshot(expected_epoch) as epoch: + if scope.current_scope != current_scope: + raise _current_stale() + result = db.get_history_occurrence(run_id, node, ordinal) + if result is not None: + receipt = result["receipt"] + path = candidate_locator(receipt["reference"], scope) + gallery, saved_unavailable = self._history_gallery( + db.model.get_connection(), + (path,) if path is not None else (), + ) + finally: + db.model.close() + if result is None: + _check_current(scope, self, fence, epoch, generation) + raise HistoryReadError(404, "not-found") + resolved = resolve_current_candidate(receipt["reference"], scope, gallery) + if resolved.public["status"] != "available": + _check_current(scope, self, fence, epoch, generation) + raise _current_unavailable(resolved.public["reason"]) + prepared = None + try: + if kind == "inspection": + saved = gallery.get(path, {}) + _reauthorize_current_target(resolved, scope) + observed = inspect_current_file( + resolved.canonical_target, + source_label=resolved.public["source_label"], + saved=saved.get("saved"), + library=saved.get( + "library", + {"status": "unavailable" if saved_unavailable else "none"}, + ), + ) + _reauthorize_current_target(resolved, scope) + identity = { + **receipt, + "history_epoch": epoch, + "current_scope": scope.current_scope, + "location_key": resolved.public["location_key"], + "evidence_class": "locator_candidate", + "observation": "current-file", + } + try: + prepared = encode_history_inspection_response( + current_output=identity, **observed + ) + except ValueError: + _check_current(scope, self, fence, epoch, generation) + raise HistoryReadError(422, "response-limit") from None + else: + prepared = _prepare_current_media( + resolved, scope, thumbnail=kind == "thumbnail" + ) + _check_current(scope, self, fence, epoch, generation) + return prepared + except CurrentFileChanged: + if prepared is not None and not isinstance(prepared, bytes): + prepared.close() + _check_current(scope, self, fence, epoch, generation) + raise _current_stale() from None + except CurrentLocationUnavailable as error: + if prepared is not None and not isinstance(prepared, bytes): + prepared.close() + _check_current(scope, self, fence, epoch, generation) + raise ( + _current_stale() + if error.stale + else _current_unavailable(error.reason) + ) from None + except BaseException as error: + if prepared is not None and not isinstance(prepared, bytes): + prepared.close() + if isinstance(error, Exception) and not isinstance( + error, (HistoryReadError, HistoryEpochChanged, sqlite3.Error) + ): + _check_current(scope, self, fence, epoch, generation) + raise + + pending = asyncio.create_task(self._run_in_executor(project)) + try: + prepared = await asyncio.shield(pending) + except asyncio.CancelledError: + pending.add_done_callback(_close_prepared_result) + raise + except ( + HistoryReadError, + HistoryEpochChanged, + sqlite3.Error, + OSError, + ValueError, + ): + raise + except Exception as error: + self.logger.warning( + "Current history media unavailable (%s)", type(error).__name__ + ) + raise _current_unavailable("temporary_unavailable") from None + if isinstance(prepared, bytes): + return _success_response(prepared) + return await _current_media_response(request, prepared) diff --git a/py/api/history_current.py b/py/api/history_current.py new file mode 100644 index 00000000..c277e132 --- /dev/null +++ b/py/api/history_current.py @@ -0,0 +1,399 @@ +"""Native reported locations and already-opened current media, without ownership.""" + +import asyncio +from dataclasses import dataclass +import errno +import hashlib +import hmac +import json +import mimetypes +import os +from pathlib import Path +import stat +from typing import BinaryIO + +from aiohttp import web + +from .images import _canonical, _public_path, SERVABLE_EXTENSIONS, VIDEO_EXTENSIONS + +try: + from ...utils.execution_outputs import ( + OutputReference, + OutputRoots, + snapshot_output_roots, + validate_output_reference, + ) +except ImportError: + from utils.execution_outputs import ( + OutputReference, + OutputRoots, + snapshot_output_roots, + validate_output_reference, + ) + +_SECRET = os.urandom(32) +_READ_SIZE = 64 * 1024 + + +def _opaque(value: object) -> str: + raw = json.dumps(value, ensure_ascii=True, separators=(",", ":")).encode("ascii") + return hmac.new(_SECRET, raw, hashlib.sha256).hexdigest() + + +@dataclass(frozen=True) +class CurrentOutputScope: + path_style: str + lexical_output: str | None + lexical_temp: str | None + roots: OutputRoots + canonical_output: Path | None + canonical_temp: Path | None + authorized_roots: tuple[Path, ...] + current_scope: str + + +@dataclass(frozen=True) +class ResolvedCurrentCandidate: + public: dict + canonical_target: Path | None = None + namespace: Path | None = None + gallery_root: Path | None = None + + +class CurrentLocationUnavailable(Exception): + """A bounded current-file failure without the underlying private path.""" + + def __init__(self, reason: str, *, stale: bool = False) -> None: + self.reason = reason + self.stale = stale + super().__init__(reason) + + +def _host_directory(name: str) -> str | None: + try: + import folder_paths + + value = getattr(folder_paths, name)() + except (ImportError, AttributeError, OSError): + return None + return value if type(value) is str else None + + +def snapshot_current_output_scope(api: object) -> CurrentOutputScope: + """Bind host lexical namespaces independently of current gallery authority.""" + from ..config import GalleryConfig + + style = "windows" if os.name == "nt" else "posix" + output = _host_directory("get_output_directory") + temp = _host_directory("get_temp_directory") + roots = snapshot_output_roots(style, output, temp) + canonical_output = _canonical(output) if roots.output_key is not None else None + canonical_temp = _canonical(temp) if roots.temp_key is not None else None + authorized = tuple( + sorted({_canonical(path) for path in api._get_all_output_dirs()}) + ) + policy = [] + for entry in GalleryConfig.MONITORING_DIRECTORIES: + valid, _ = GalleryConfig.validate_gallery_root(entry) + policy.append((entry if type(entry) is str else None, valid)) + current_scope = _opaque( + [ + "prompt-manager-current-output-v1", + style, + output, + temp, + roots.output_key, + roots.temp_key, + str(canonical_output) if canonical_output is not None else None, + str(canonical_temp) if canonical_temp is not None else None, + [str(path) for path in authorized], + sorted( + str(_canonical(path)) + for path in GalleryConfig.allowed_gallery_parents() + ), + sorted(policy, key=lambda item: (item[0] or "", item[1])), + sorted(str(_canonical(path)) for path in GalleryConfig.path_anchors()), + "gallery-validation-v1", + ] + ) + return CurrentOutputScope( + style, + output, + temp, + roots, + canonical_output, + canonical_temp, + authorized, + current_scope, + ) + + +def _descriptor_reason(reference: dict, scope: CurrentOutputScope) -> str | None: + try: + validate_output_reference(OutputReference("current", 0, **reference)) + except (TypeError, ValueError): + return "namespace_mismatch" + if reference["path_style"] != scope.path_style: + return "foreign_path_style" + folder = reference["folder_type"] + expected = scope.roots.output_key if folder == "output" else scope.roots.temp_key + if reference["root_key"] is None or expected is None: + return "root_unavailable" + if reference["root_key"] != expected: + return "namespace_mismatch" + return None + + +def candidate_locator(reference: dict, scope: CurrentOutputScope) -> str | None: + """Validate the native namespace before constructing a lexical path; no IO.""" + if _descriptor_reason(reference, scope) is not None: + return None + root = ( + scope.lexical_output + if reference["folder_type"] == "output" + else scope.lexical_temp + ) + return os.path.normcase( + os.path.normpath( + os.path.join(root, reference["subfolder"], reference["filename"]) + ) + ) + + +def _authorized(target: Path, namespace: Path, scope: CurrentOutputScope) -> bool: + return target.is_relative_to(namespace) and any( + target.is_relative_to(root) for root in scope.authorized_roots + ) + + +def _unavailable(reason: str, *, error: bool = False) -> ResolvedCurrentCandidate: + return ResolvedCurrentCandidate( + { + "status": "error" if error else "unavailable", + "reason": reason, + "evidence_class": "locator_candidate", + } + ) + + +def resolve_current_candidate( + reference: dict, scope: CurrentOutputScope, gallery: dict +) -> ResolvedCurrentCandidate: + """Observe only this exact current locator and its snapshot-owned association.""" + reason = _descriptor_reason(reference, scope) + if reason is not None: + return _unavailable(reason) + lexical = candidate_locator(reference, scope) + namespace = ( + scope.canonical_output + if reference["folder_type"] == "output" + else scope.canonical_temp + ) + try: + if Path(reference["filename"]).suffix.lower() not in SERVABLE_EXTENSIONS: + return _unavailable("unsupported_media") + if namespace is None or not namespace.is_dir(): + return _unavailable("root_unavailable") + target = _canonical(lexical) + if not _authorized(target, namespace, scope): + return _unavailable("not_authorized") + if target.suffix.lower() not in SERVABLE_EXTENSIONS: + return _unavailable("unsupported_media") + observed = target.stat() + if not stat.S_ISREG(observed.st_mode): + return _unavailable("not_regular_file") + if not os.access(target, os.R_OK) or not observed.st_mode & 0o444: + return _unavailable("unreadable") + matching = gallery.get(lexical, {}) + root = max( + (root for root in scope.authorized_roots if target.is_relative_to(root)), + key=lambda root: len(root.parts), + ) + return ResolvedCurrentCandidate( + { + "status": "available", + "reason": None, + "evidence_class": "locator_candidate", + "gallery_match": matching.get("gallery_match", "none"), + "gallery_image_id": matching.get("gallery_image_id"), + "source_label": _public_path(namespace), + "relative_path": Path( + reference["subfolder"], reference["filename"] + ).as_posix(), + "current_scope": scope.current_scope, + "location_key": _opaque( + ["current-location-v1", scope.current_scope, str(target)] + ), + }, + target, + namespace, + root, + ) + except FileNotFoundError: + return _unavailable("missing") + except PermissionError: + return _unavailable("unreadable") + except (OSError, ValueError): + return _unavailable("temporary_unavailable", error=True) + + +def _open_current_target( + target: Path, namespace: Path, scope: CurrentOutputScope +) -> BinaryIO: + """Open only the canonical target and verify authority against the actual handle.""" + if _canonical(target) != target: + raise CurrentLocationUnavailable("not_authorized", stale=True) + if not _authorized(target, namespace, scope): + raise CurrentLocationUnavailable("not_authorized") + if target.suffix.lower() not in SERVABLE_EXTENSIONS: + raise CurrentLocationUnavailable("unsupported_media") + flags = ( + os.O_RDONLY + | getattr(os, "O_BINARY", 0) + | getattr(os, "O_NOFOLLOW", 0) + | getattr(os, "O_NONBLOCK", 0) + ) + try: + fd = os.open(target, flags) + except FileNotFoundError: + raise CurrentLocationUnavailable("missing") from None + except PermissionError: + raise CurrentLocationUnavailable("unreadable") from None + except OSError as error: + if error.errno in (errno.ELOOP, errno.ENOTDIR): + raise CurrentLocationUnavailable("not_authorized", stale=True) from None + raise CurrentLocationUnavailable("temporary_unavailable") from None + try: + observed = os.fstat(fd) + if not stat.S_ISREG(observed.st_mode): + raise CurrentLocationUnavailable("not_regular_file") + if not observed.st_mode & 0o444 or not os.access(target, os.R_OK): + raise CurrentLocationUnavailable("unreadable") + current = os.stat(target, follow_symlinks=False) + if ( + _canonical(target) != target + or (observed.st_dev, observed.st_ino) != (current.st_dev, current.st_ino) + or not _authorized(target, namespace, scope) + ): + raise CurrentLocationUnavailable("not_authorized", stale=True) + return os.fdopen(fd, "rb") + except FileNotFoundError: + os.close(fd) + raise CurrentLocationUnavailable("missing") from None + except PermissionError: + os.close(fd) + raise CurrentLocationUnavailable("unreadable") from None + except BaseException: + os.close(fd) + raise + + +def _reauthorize_current_target( + resolved: ResolvedCurrentCandidate, scope: CurrentOutputScope +) -> None: + if _canonical( + resolved.canonical_target + ) != resolved.canonical_target or not _authorized( + resolved.canonical_target, resolved.namespace, scope + ): + raise CurrentLocationUnavailable("not_authorized", stale=True) + + +@dataclass +class _OpenedCurrentMedia: + handle: BinaryIO + size: int + content_type: str + + def close(self) -> None: + self.handle.close() + + +def _prepare_current_media( + resolved: ResolvedCurrentCandidate, scope: CurrentOutputScope, *, thumbnail: bool +) -> _OpenedCurrentMedia: + target = resolved.canonical_target + if thumbnail: + relative = target.relative_to(resolved.gallery_root) + extension = ( + ".jpg" + if target.suffix.lower() in VIDEO_EXTENSIONS + else target.suffix.lower() + ) + preview = ( + resolved.gallery_root + / "thumbnails" + / f"{relative.with_suffix('').as_posix()}_thumb{extension}" + ) + try: + preview.stat() + except FileNotFoundError: + pass + else: + target = _canonical(preview) + handle = _open_current_target(target, resolved.namespace, scope) + try: + return _OpenedCurrentMedia( + handle, + os.fstat(handle.fileno()).st_size, + mimetypes.guess_type(target.name)[0] or "application/octet-stream", + ) + except BaseException: + handle.close() + raise + + +async def _current_media_response( + request: web.Request, opened: _OpenedCurrentMedia +) -> web.StreamResponse: + """Stream a bounded single range from a previously authorized handle.""" + try: + try: + selected = request.http_range + start, stop = selected.start, selected.stop + partial = request.headers.get("Range") is not None + if ( + partial + and request.headers["Range"].startswith("bytes=-") + and start == 0 + ): + raise ValueError("Empty suffix range") + if start is not None and start < 0: + start = max(0, opened.size + start) + start = start or 0 + stop = min(opened.size, stop) if stop is not None else opened.size + if partial and (start >= stop or start >= opened.size): + raise ValueError("Unsatisfiable current media range") + except ValueError: + return web.Response( + status=416, + headers={ + "Cache-Control": "no-store", + "Content-Range": f"bytes */{opened.size}", + }, + ) + headers = { + "Cache-Control": "no-store", + "Accept-Ranges": "bytes", + "Content-Type": opened.content_type, + "Content-Length": str(stop - start), + } + if partial: + headers["Content-Range"] = f"bytes {start}-{stop - 1}/{opened.size}" + response = web.StreamResponse(status=206 if partial else 200, headers=headers) + await response.prepare(request) + if request.method != "HEAD": + opened.handle.seek(start) + remaining = stop - start + while remaining: + chunk = await asyncio.get_running_loop().run_in_executor( + None, opened.handle.read, min(_READ_SIZE, remaining) + ) + if not chunk: + break + remaining -= len(chunk) + await response.write(chunk) + await response.write_eof() + return response + finally: + opened.close() diff --git a/py/api/images.py b/py/api/images.py index b920ec37..3a46d5fa 100644 --- a/py/api/images.py +++ b/py/api/images.py @@ -20,23 +20,14 @@ from PIL import Image, UnidentifiedImageError from .prompts import bad_request, parse_page_params, publish_image_paths from .prompts import thumbnail_url_for from .prompts import safe_error_message as _safe_error +from .inspection_service import CurrentFileChanged, inspect_current_file try: from ...utils.parallel import map_parallel from ...utils.inspection_response import encode_inspection_response - from ...utils.prompt_inspector import ( - MetadataReadError, - inspect_prompt_evidence, - read_inspection_file, - ) except ImportError: from utils.parallel import map_parallel from utils.inspection_response import encode_inspection_response - from utils.prompt_inspector import ( - MetadataReadError, - inspect_prompt_evidence, - read_inspection_file, - ) # Only these file types are ever served by the image routes, regardless of # what sits inside an allowed directory (defence in depth for the gallery). @@ -1732,18 +1723,6 @@ class ImageRoutesMixin: try: def inspect() -> bytes | None: - before = path.stat() - try: - inspection = read_inspection_file(path) - except MetadataReadError: - inspection = None - try: - after = path.stat() - except FileNotFoundError: - return None - fields = ("st_dev", "st_ino", "st_size", "st_mtime_ns", "st_ctime_ns") - if any(getattr(before, key) != getattr(after, key) for key in fields): - return None saved, unavailable = None, False try: saved = self.db.get_image_prompt_info( @@ -1752,25 +1731,17 @@ class ImageRoutesMixin: except Exception: unavailable = True self.logger.warning("Saved inspection display could not be read") - if inspection is None: - evidence = { - "prompts": { - role: {"status": "unresolved", "text": None, "source": None} - for role in ("positive", "negative") - }, - "resolution": { - "state": "unresolved", - "sampler_id": None, - "reason": "Media metadata could not be read", - }, - } - else: - role_saved = ( - None - if saved and saved.get("association_status") == "ambiguous" - else saved + try: + observed = inspect_current_file( + path, + source_label=_public_path(roots[index]), + saved=saved, + library=self._inspection_library( + saved, unavailable=unavailable + ), ) - evidence = inspect_prompt_evidence(inspection.metadata, role_saved) + except CurrentFileChanged: + return None return encode_inspection_response( image={ "root_index": index, @@ -1778,25 +1749,7 @@ class ImageRoutesMixin: "filename": path.name, }, root_scope=root_scope, - file={ - "observation": "current-file", - "source_label": _public_path(roots[index]), - "media_type": ( - "video" - if path.suffix.lower() in VIDEO_EXTENSIONS - else "image" - ), - "extension": path.suffix.lower(), - "size": before.st_size, - "modified_time": before.st_mtime, - "format": None, - "width": None, - "height": None, - }, - prompts=evidence["prompts"], - resolution=evidence["resolution"], - inspection=inspection, - library=self._inspection_library(saved, unavailable=unavailable), + **observed, ) body = await self._run_in_executor(inspect) diff --git a/py/api/inspection_service.py b/py/api/inspection_service.py new file mode 100644 index 00000000..47ca7362 --- /dev/null +++ b/py/api/inspection_service.py @@ -0,0 +1,72 @@ +"""One bounded current-file inspection worker shared by physical identities.""" + +from pathlib import Path + +try: + from ...utils.prompt_inspector import ( + MetadataReadError, + inspect_prompt_evidence, + read_inspection_file, + ) + from ...utils.video_metadata import is_video_path +except ImportError: + from utils.prompt_inspector import ( + MetadataReadError, + inspect_prompt_evidence, + read_inspection_file, + ) + from utils.video_metadata import is_video_path + + +class CurrentFileChanged(Exception): + """File facts changed across the bounded metadata observation.""" + + +def inspect_current_file( + path: Path, *, source_label: str, saved: dict | None, library: dict +) -> dict: + before = path.stat() + try: + inspection = read_inspection_file(path) + except MetadataReadError: + inspection = None + try: + after = path.stat() + except FileNotFoundError: + raise CurrentFileChanged from None + fields = ("st_dev", "st_ino", "st_size", "st_mtime_ns", "st_ctime_ns") + if any(getattr(before, key) != getattr(after, key) for key in fields): + raise CurrentFileChanged + if inspection is None: + evidence = { + "prompts": { + role: {"status": "unresolved", "text": None, "source": None} + for role in ("positive", "negative") + }, + "resolution": { + "state": "unresolved", + "sampler_id": None, + "reason": "Media metadata could not be read", + }, + } + else: + role_saved = ( + None if saved and saved.get("association_status") == "ambiguous" else saved + ) + evidence = inspect_prompt_evidence(inspection.metadata, role_saved) + return { + "file": { + "observation": "current-file", + "source_label": source_label, + "media_type": "video" if is_video_path(path) else "image", + "extension": path.suffix.lower(), + "size": before.st_size, + "modified_time": before.st_mtime, + "format": None, + "width": None, + "height": None, + }, + **evidence, + "inspection": inspection, + "library": library, + } diff --git a/tests/test_api_history_outputs.py b/tests/test_api_history_outputs.py new file mode 100644 index 00000000..47a6597a --- /dev/null +++ b/tests/test_api_history_outputs.py @@ -0,0 +1,1090 @@ +"""Real occurrence routes preserve receipts and fence current-file publication.""" + +import asyncio +import errno +import importlib +import json +import sqlite3 +import threading +from contextlib import ExitStack +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +from aiohttp import web +from aiohttp.test_utils import make_mocked_request + +from tests.test_api_images import ImageAPITestCase, make_png +from tests.test_api_prompt_inspector import graph, png +from tests.test_execution_context_database import bundle as context_bundle +from tests.test_execution_history_database import JOB, SESSION +from utils.execution_contexts import ContextBundle +from utils.execution_history_events import make_finish, make_start +from utils.execution_outputs import OutputBundle, OutputReference, snapshot_output_roots +from py.config import GalleryConfig + + +class HistoryOutputAPITests(ImageAPITestCase): + async def get_application(self): + app = await super().get_application() + self.temp_dir = self.comfy_dir / "temp" + self.temp_dir.mkdir() + self.host_output = str(self.output_dir) + self.host_temp = str(self.temp_dir) + self.patches = ExitStack() + self.addCleanup(self.patches.close) + self.patches.enter_context( + patch( + "folder_paths.get_output_directory", + lambda: self.host_output, + create=True, + ) + ) + self.patches.enter_context( + patch( + "folder_paths.get_temp_directory", lambda: self.host_temp, create=True + ) + ) + self.patches.enter_context( + patch.object( + GalleryConfig, "allowed_gallery_parents", lambda: [str(self.comfy_dir)] + ) + ) + return app + + async def tearDownAsync(self): + self.api.close_history_reads() + self.patches.close() + await super().tearDownAsync() + + def ref(self, node="1", ordinal=0, filename="image.png", **kwargs): + roots = snapshot_output_roots("posix", self.host_output, self.host_temp) + return OutputReference( + node, + ordinal, + kwargs.get("folder_type", "output"), + kwargs.get("subfolder", ""), + filename, + kwargs.get("path_style", "posix"), + kwargs.get("root_key", roots.output_key), + ) + + def capture(self, references=None, item=0, state="complete", reason=None): + start = make_start( + SESSION, ((0, JOB, {}, {}), item), self.api.db.get_execution_epoch(), item + ) + self.api.db.write_execution_start(start) + finish = make_finish( + SESSION, item, start.storage_epoch, SimpleNamespace(messages=[]), item + ) + self.api.db.write_execution_finish(finish) + if references is not None: + value = OutputBundle( + tuple(references), + state, + reason or ("reported" if references else "none_reported"), + ) + self.assertTrue( + self.api.db.write_execution_outputs_for_finish(finish, value) + ) + return start + + async def outputs(self, start, params=None): + query = {"epoch": start.storage_epoch} + if params is not None: + query = params + response = await self.client.get( + f"/prompt_manager/history/runs/{start.run_id}/outputs", params=query + ) + self.assertEqual( + response.content_type, + "application/json", + "reported-output route is missing", + ) + return response, await response.json() + + async def current(self, url, **kwargs): + response = await self.client.request(kwargs.pop("method", "GET"), url, **kwargs) + return response + + async def media(self, start): + response, body = await self.outputs(start) + self.assertEqual(response.status, 200, body) + return body["items"][0]["current_candidate"]["media"] + + def assert_error(self, response, body, status, code): + self.assertEqual(response.status, status, body) + self.assertFalse(body["success"]) + self.assertEqual(body["history_version"], 1) + self.assertEqual(body["code"], code) + self.assertEqual(response.headers.get("Cache-Control"), "no-store") + self.assertNotIn(str(self.comfy_dir), json.dumps(body)) + + def saved_state(self): + conn = self.api.db.model.get_connection() + return { + table: [ + tuple(row) for row in conn.execute(f"SELECT * FROM {table} ORDER BY id") + ] + for table in ( + "prompts", + "prompt_pairs", + "generated_images", + "image_prompt_pairs", + ) + } + + def restore_fixture(self): + source = str(self.comfy_dir / "restore.db") + self.assertTrue(self.api.db.model.backup_database(source)) + return source + + async def test_ordered_receipts_repeat_locators_without_context_ownership(self): + make_png(self.output_dir / "image.png") + start = self.capture([self.ref("1", 0), self.ref("1", 1), self.ref("2", 0)]) + contexts = ( + context_bundle("frozen A", anchor="1").contexts[0], + context_bundle("frozen B", anchor="2").contexts[0], + ) + self.api.db.write_execution_contexts( + start.run_id, + start.storage_epoch, + ContextBundle(contexts, "complete", "captured"), + ) + repeated = self.capture([self.ref()], item=1) + with patch.object( + self.api.db, + "get_execution_contexts", + side_effect=AssertionError("contexts x outputs"), + ): + response, page = await self.outputs( + start, {"epoch": start.storage_epoch, "limit": "2"} + ) + self.assertEqual(response.status, 200, page) + self.assertEqual(page["retained_occurrence_count"], 3) + self.assertEqual(page["list_status"], "complete") + self.assertEqual( + [ + (r["occurrence"]["output_node_id"], r["occurrence"]["image_ordinal"]) + for r in page["items"] + ], + [("1", 0), ("1", 1)], + ) + self.assertTrue(page["has_more"]) + self.assertLessEqual(len(page["next_cursor"]), 4096) + _, second = await self.outputs( + start, {"epoch": start.storage_epoch, "cursor": page["next_cursor"]} + ) + self.assertEqual(second["items"][0]["occurrence"]["output_node_id"], "2") + self.assertFalse(second["has_more"]) + self.assertIsNone(second["next_cursor"]) + _, other = await self.outputs(repeated) + self.assertNotEqual(other["items"][0]["occurrence"]["run_id"], start.run_id) + for receipt in page["items"] + second["items"] + other["items"]: + self.assertEqual(receipt["file_identity"], "unverified") + self.assertEqual(receipt["newly_generated"], "unknown") + self.assertEqual(receipt["context_membership"], "unknown") + self.assertNotIn("context_id", receipt) + candidate = receipt["current_candidate"] + self.assertEqual(candidate["status"], "available") + self.assertEqual(candidate["evidence_class"], "locator_candidate") + self.assertEqual(candidate["media"]["occurrence"], receipt["occurrence"]) + self.assertEqual(candidate["media"]["reference"], receipt["reference"]) + self.assertEqual(candidate["media"]["kind"], "reported-location") + for field in ("url", "thumbnail_url", "inspection_url"): + self.assertTrue( + candidate["media"][field].startswith( + "/prompt_manager/history/runs/" + f"{receipt['occurrence']['run_id']}/output/current/" + ) + ) + self.assertNotIn("root=", candidate["media"][field]) + self.assertNotIn(str(self.comfy_dir), candidate["media"][field]) + + async def test_complete_zero_incomplete_unavailable_and_not_captured_are_distinct( + self, + ): + cases = [ + ((), "complete", "none_reported", 0, "complete"), + ((self.ref(),), "partial", "limit_reached", 1, "incomplete"), + ((), "unavailable", "unavailable", 0, "incomplete"), + (None, "not_captured", None, None, "not_captured"), + ] + for item, (refs, state, reason, count, status) in enumerate(cases): + start = self.capture(refs, item, state, reason) + response, page = await self.outputs(start) + self.assertEqual(response.status, 200, page) + self.assertEqual(page["capture"]["state"], state) + self.assertEqual(page["retained_occurrence_count"], count) + self.assertEqual(page["list_status"], status) + self.assertFalse(page["has_more"]) + + async def test_epoch_limits_cursor_scope_and_header_conflict(self): + start = self.capture([self.ref("1"), self.ref("2")]) + allowed, page = await self.outputs( + start, {"epoch": start.storage_epoch, "limit": "1"} + ) + self.assertEqual(allowed.status, 200, page) + for params in ( + {}, + {"epoch": "bad"}, + {"epoch": start.storage_epoch, "limit": "0"}, + {"epoch": start.storage_epoch, "limit": "101"}, + {"epoch": start.storage_epoch, "cursor": "bad"}, + ): + response, error = await self.outputs(start, params) + self.assertEqual(response.status, 400, error) + token = page["next_cursor"] + other = self.capture([self.ref()], item=1) + response, error = await self.outputs( + other, {"epoch": start.storage_epoch, "cursor": token} + ) + self.assert_error(response, error, 400, "invalid-cursor") + finish = make_finish( + SESSION, 0, start.storage_epoch, SimpleNamespace(messages=[]), 0 + ) + self.api.db.write_execution_outputs_for_finish( + finish, + OutputBundle((self.ref(filename="other.png"),), "complete", "reported"), + ) + response, error = await self.outputs( + start, {"epoch": start.storage_epoch, "cursor": token} + ) + self.assert_error(response, error, 409, "capture-changed") + _, current = await self.outputs(start) + self.assertEqual(current["capture"]["state"], "conflicting") + self.assertEqual(len(current["items"]), 2) + source = self.restore_fixture() + self.api.db.model.restore_from_file(source) + response, error = await self.outputs(start) + self.assert_error(response, error, 409, "history-changed") + + async def test_metadata_free_malformed_and_replacement_remain_read_only(self): + path = make_png(self.output_dir / "image.png") + start = self.capture([self.ref()]) + self.api.db.write_execution_contexts( + start.run_id, + start.storage_epoch, + context_bundle("Frozen-only negative", "Frozen-only positive"), + ) + frozen_outputs = self.api.db.get_execution_outputs(start.run_id) + before = self.saved_state() + for metadata in ( + {}, + {"prompt": "{malformed"}, + {"prompt": graph("Current file prompt")}, + ): + png(path, **metadata) + descriptor = await self.media(start) + response = await self.current(descriptor["url"]) + self.assertEqual(response.status, 200) + self.assertEqual(await response.read(), path.read_bytes()) + response = await self.current(descriptor["inspection_url"]) + body = await response.json() + self.assertEqual(response.status, 200, body) + identity = body["current_output"] + self.assertEqual(body.get("history_version"), 1) + self.assertEqual(body.get("history_epoch"), start.storage_epoch) + self.assertEqual(identity["occurrence"], descriptor["occurrence"]) + self.assertEqual(identity["reference"], descriptor["reference"]) + self.assertEqual(identity["history_epoch"], start.storage_epoch) + self.assertEqual(identity["current_scope"], descriptor["current_scope"]) + self.assertEqual(identity["location_key"], descriptor["location_key"]) + self.assertEqual(identity["file_identity"], "unverified") + self.assertEqual(body["file"]["observation"], "current-file") + self.assertNotIn("Frozen-only negative", json.dumps(body)) + self.assertNotIn("Frozen-only positive", json.dumps(body)) + self.assertEqual(self.saved_state(), before) + self.assertEqual(body["prompts"]["positive"]["text"], "Current file prompt") + self.assertEqual( + self.api.db.get_execution_outputs(start.run_id), frozen_outputs + ) + + async def test_exact_gallery_matches_do_not_choose_duplicate_owners_or_basename( + self, + ): + path = make_png(self.output_dir / "image.png") + other = self.temp_dir / path.name + make_png(other) + self._link_image(other, "Other root") + start = self.capture([self.ref()]) + _, page = await self.outputs(start) + self.assertEqual(page["items"][0]["current_candidate"]["gallery_match"], "none") + first = self._link_image(path, "first owner") + _, page = await self.outputs(start) + self.assertEqual( + page["items"][0]["current_candidate"]["gallery_image_id"], first + ) + self._link_image(path, "second owner") + descriptor = await self.media(start) + _, page = await self.outputs(start) + candidate = page["items"][0]["current_candidate"] + self.assertEqual(candidate["gallery_match"], "ambiguous") + self.assertIsNone(candidate["gallery_image_id"]) + response = await self.current(descriptor["inspection_url"]) + body = await response.json() + self.assertEqual(response.status, 200, body) + self.assertEqual(body["library"]["status"], "ambiguous") + self.assertIsNone(body["library"].get("positive_id")) + + async def test_scope_reconfiguration_invalidates_urls_without_root_index_fallback( + self, + ): + make_png(self.output_dir / "image.png") + make_png(self.temp_dir / "image.png", color=(0, 255, 0)) + start = self.capture([self.ref()]) + descriptor = await self.media(start) + self.api._get_all_output_dirs = lambda: [self.temp_dir, self.output_dir] + for field in ("url", "thumbnail_url", "inspection_url"): + response = await self.current(descriptor[field]) + self.assert_error(response, await response.json(), 409, "candidate-stale") + current = await self.media(start) + response = await self.current(current["url"]) + self.assertEqual( + await response.read(), (self.output_dir / "image.png").read_bytes() + ) + self.api._get_all_output_dirs = lambda: [self.output_dir, self.temp_dir] + response = await self.current(current["url"]) + self.assertEqual(response.status, 200) + self.assertEqual( + await response.read(), (self.output_dir / "image.png").read_bytes() + ) + self.host_output = str(self.temp_dir) + _, page = await self.outputs(start) + self.assertEqual( + page["items"][0]["current_candidate"]["reason"], "namespace_mismatch" + ) + + async def test_occurrence_query_is_strict_and_viewer_cachebuster_is_compatible( + self, + ): + make_png(self.output_dir / "image.png") + start = self.capture([self.ref()]) + descriptor = await self.media(start) + allowed = await self.current(descriptor["url"] + "&_pm_viewer=123-2") + self.assertEqual(allowed.status, 200) + await allowed.read() + path = descriptor["url"].split("?")[0] + base = { + "node": "1", + "ordinal": "0", + "epoch": start.storage_epoch, + "current_scope": descriptor["current_scope"], + } + for key in base: + params = dict(base) + del params[key] + response = await self.current(path, params=params) + self.assertEqual(response.status, 400) + await response.read() + for values in ( + {"node": ""}, + {"node": "x" * 129}, + {"node": "\0"}, + {"ordinal": "256"}, + {"ordinal": "01"}, + {"current_scope": "A" * 64}, + {"current_scope": "a" * 63}, + ): + response = await self.current(path, params={**base, **values}) + self.assertEqual(response.status, 400, values) + await response.read() + response = await self.current(descriptor["url"] + "&node=1") + self.assertEqual(response.status, 400) + await response.read() + + async def test_media_ranges_head_and_existing_thumbnail_without_generation(self): + path = self.output_dir / "clip.mp4" + payload = bytes(range(256)) * 600 + path.write_bytes(payload) + start = self.capture([self.ref(filename=path.name)]) + descriptor = await self.media(start) + self.assertTrue(descriptor["is_video"]) + for header, status, expected in ( + (None, 200, payload), + ("bytes=3-7", 206, payload[3:8]), + ("bytes=-4", 206, payload[-4:]), + ("bytes=153599-", 206, payload[-1:]), + ): + response = await self.current( + descriptor["url"], headers={} if header is None else {"Range": header} + ) + self.assertEqual(response.status, status) + self.assertEqual(await response.read(), expected) + self.assertEqual(response.headers.get("Cache-Control"), "no-store") + self.assertEqual(response.headers.get("Accept-Ranges"), "bytes") + for header in ("bytes=999999-", "bytes=0-2,5-9", "bytes=bad", "bytes=-0"): + response = await self.current(descriptor["url"], headers={"Range": header}) + self.assertEqual(response.status, 416, header) + self.assertEqual(response.headers.get("Cache-Control"), "no-store") + await response.read() + response = await self.current(descriptor["url"], method="HEAD") + self.assertEqual(response.status, 200) + self.assertEqual(int(response.headers["Content-Length"]), len(payload)) + self.assertEqual(await response.read(), b"") + response = await self.current(descriptor["thumbnail_url"]) + self.assertEqual(await response.read(), payload) + preview = make_png(self.output_dir / "thumbnails" / "clip_thumb.jpg") + response = await self.current(descriptor["thumbnail_url"]) + self.assertEqual(await response.read(), preview.read_bytes()) + + async def test_saved_lookup_failure_preserves_receipts_and_file_evidence( + self, + ): + png(self.output_dir / "image.png", prompt=graph("File evidence")) + start = self.capture([self.ref()]) + descriptor = await self.media(start) + history = importlib.import_module("py.api.history") + with ( + self.assertLogs("prompt_manager.api", "WARNING"), + patch.object( + history, + "read_history_gallery", + side_effect=sqlite3.OperationalError("private/path"), + ), + ): + response, page = await self.outputs(start) + self.assertEqual(response.status, 200, page) + self.assertEqual(len(page["items"]), 1) + self.assertEqual(page["items"][0]["current_candidate"]["status"], "error") + self.assertNotIn("media", page["items"][0]["current_candidate"]) + response = await self.current(descriptor["inspection_url"]) + body = await response.json() + self.assertEqual(response.status, 200, body) + self.assertEqual(body["prompts"]["positive"]["text"], "File evidence") + self.assertEqual(body["library"]["status"], "unavailable") + + async def test_restore_during_candidate_stat_rejects_page(self): + path = make_png(self.output_dir / "image.png") + start = self.capture([self.ref()]) + source = self.restore_fixture() + original = Path.stat + armed = True + + def stat(candidate, *args, **kwargs): + nonlocal armed + result = original(candidate, *args, **kwargs) + if candidate == path and armed: + armed = False + self.api.db.model.restore_from_file(source) + return result + + with patch.object(Path, "stat", stat): + response, body = await self.outputs(start) + self.assert_error(response, body, 409, "history-changed") + + async def test_scope_change_during_candidate_stat_rejects_page(self): + path = make_png(self.output_dir / "image.png") + start = self.capture([self.ref()]) + original = Path.stat + armed = True + + def stat(candidate, *args, **kwargs): + nonlocal armed + result = original(candidate, *args, **kwargs) + if candidate == path and armed: + armed = False + self.host_output = str(self.temp_dir) + return result + + with patch.object(Path, "stat", stat): + response, body = await self.outputs(start) + self.assert_error(response, body, 409, "candidate-stale") + + async def test_restore_during_metadata_read_rejects_inspection(self): + path = png(self.output_dir / "image.png", prompt=graph()) + start = self.capture([self.ref()]) + descriptor = await self.media(start) + source = self.restore_fixture() + original = Path.open + armed = True + handles = [] + + def opened(candidate, *args, **kwargs): + nonlocal armed + handle = original(candidate, *args, **kwargs) + if candidate == path and args and args[0] == "rb": + handles.append(handle) + if armed: + armed = False + self.api.db.model.restore_from_file(source) + return handle + + with patch.object(Path, "open", opened): + response = await self.current(descriptor["inspection_url"]) + body = await response.json() + self.assert_error(response, body, 409, "history-changed") + self.assertTrue(handles) + self.assertTrue(all(handle.closed for handle in handles)) + + async def test_restore_during_open_closes_handle_before_error_headers(self): + make_png(self.output_dir / "image.png") + start = self.capture([self.ref()]) + descriptor = await self.media(start) + source = self.restore_fixture() + mod = importlib.import_module("py.api.history_current") + original = mod._open_current_target + handles = [] + + def opened(*args, **kwargs): + handle = original(*args, **kwargs) + handles.append(handle) + self.api.db.model.restore_from_file(source) + return handle + + with patch.object(mod, "_open_current_target", side_effect=opened): + response = await self.current(descriptor["url"]) + body = await response.json() + self.assert_error(response, body, 409, "history-changed") + self.assertTrue(all(handle.closed for handle in handles)) + + async def test_primary_projection_and_fence_failure_never_return_empty_success( + self, + ): + make_png(self.output_dir / "image.png") + start = self.capture([self.ref()]) + self.assertEqual((await self.outputs(start))[0].status, 200) + with ( + self.assertLogs("prompt_manager.api", "WARNING"), + patch.object( + self.api.db, + "list_history_outputs", + side_effect=sqlite3.OperationalError("private"), + ), + ): + response, body = await self.outputs(start) + self.assert_error(response, body, 503, "storage-unavailable") + with ( + self.assertLogs("prompt_manager.api", "WARNING"), + patch.object( + self.api._history_fence(), + "validate", + side_effect=sqlite3.DatabaseError("private"), + ), + ): + response, body = await self.outputs(start) + self.assert_error(response, body, 503, "storage-unavailable") + + async def test_page_deduplicates_locators_and_closes_connection_before_io( + self, + ): + make_png(self.output_dir / "image.png") + start = self.capture([self.ref("1", i) for i in range(105)]) + self.assertEqual((await self.outputs(start))[0].status, 200) + history = importlib.import_module("py.api.history") + original = history.resolve_current_candidate + locators = [] + + def resolved(reference, scope, gallery): + self.assertIsNone(getattr(self.api.db.model._local, "conn", None)) + locators.append(reference) + return original(reference, scope, gallery) + + with patch.object(history, "resolve_current_candidate", side_effect=resolved): + response, body = await self.outputs( + start, {"epoch": start.storage_epoch, "limit": "100"} + ) + self.assertEqual(response.status, 200, body) + self.assertEqual(len(body["items"]), 100) + self.assertEqual(len(locators), 1) + + async def test_actual_control_heavy_page_uses_serialized_bytes_and_smaller_retry( + self, + ): + name = "\x01\n猫.png" + make_png(self.output_dir / name) + refs = [self.ref("\x01\n猫" * 40, i, filename=name) for i in range(40)] + start = self.capture(refs) + response, body = await self.outputs(start) + self.assertEqual(response.status, 200, body) + encoded = await response.read() + self.assertIn(b"\\u0001", encoded) + self.assertIn(b"%01", encoded) + module = importlib.import_module("py.api.history") + with patch.object(module, "_OUTPUTS_MAX_BYTES", len(encoded)): + response, same = await self.outputs(start) + self.assertEqual(response.status, 200, same) + with patch.object(module, "_OUTPUTS_MAX_BYTES", len(encoded) - 1): + response, error = await self.outputs(start) + self.assert_error(response, error, 422, "response-limit") + self.assertEqual( + error["recovery"], {"action": "retry-outputs", "limit": 20} + ) + response, smaller = await self.outputs( + start, {"epoch": start.storage_epoch, "limit": "20"} + ) + self.assertEqual(response.status, 200, smaller) + self.assertEqual(len(smaller["items"]), 20) + with patch.object(module, "_OUTPUTS_MAX_BYTES", 1): + response, error = await self.outputs( + start, {"epoch": start.storage_epoch, "limit": "1"} + ) + self.assert_error(response, error, 422, "response-limit") + self.assertIsNone(error["recovery"]) + + async def test_legal_capture_can_exceed_one_mib_and_recover_without_lost_receipts( + self, + ): + subfolder = "/".join(["\x01" * 175] * 4) + filename = "\x02" * 20 + ".png" + make_png(self.output_dir / subfolder / filename) + start = self.capture( + [ + self.ref("\x03" * 128, i, filename, subfolder=subfolder) + for i in range(100) + ] + ) + response, error = await self.outputs( + start, {"epoch": start.storage_epoch, "limit": "100"} + ) + self.assert_error(response, error, 422, "response-limit") + self.assertEqual(error["recovery"], {"action": "retry-outputs", "limit": 50}) + response, page = await self.outputs( + start, {"epoch": start.storage_epoch, "limit": "50"} + ) + self.assertEqual(response.status, 200, page) + self.assertLessEqual(len(await response.read()), 1024 * 1024) + self.assertEqual(len(page["items"]), 50) + response, next_page = await self.outputs( + start, + { + "epoch": start.storage_epoch, + "limit": "50", + "cursor": page["next_cursor"], + }, + ) + self.assertEqual(response.status, 200, next_page) + self.assertEqual( + [ + r["occurrence"]["image_ordinal"] + for r in page["items"] + next_page["items"] + ], + list(range(100)), + ) + + async def test_native_alias_allowed_and_foreign_null_temp_denials_keep_receipts( + self, + ): + path = make_png(self.output_dir / "image.png") + alias = self.comfy_dir / "alias" + alias.symlink_to(self.output_dir, target_is_directory=True) + self.host_output = str(alias) + allowed = self.capture([self.ref()]) + descriptor = await self.media(allowed) + response = await self.current(descriptor["url"]) + self.assertEqual(response.status, 200) + self.assertEqual(await response.read(), path.read_bytes()) + make_png(self.temp_dir / "image.png") + temp_key = snapshot_output_roots( + "posix", self.host_output, self.host_temp + ).temp_key + denied = self.capture( + [ + self.ref("1", root_key=None), + self.ref("2", path_style="windows", root_key="a" * 64), + self.ref("3", folder_type="temp", root_key=temp_key), + ], + item=1, + state="partial", + reason="root_unavailable", + ) + response, page = await self.outputs(denied) + self.assertEqual(response.status, 200, page) + self.assertEqual( + [r["current_candidate"]["reason"] for r in page["items"]], + ["root_unavailable", "foreign_path_style", "not_authorized"], + ) + self.assertTrue( + all("media" not in r["current_candidate"] for r in page["items"]) + ) + self.assertTrue(page["items"][2]["transient"]) + + async def test_unavailable_files_keep_receipts_and_current_routes_fail_closed(self): + path = make_png(self.output_dir / "image.png") + start = self.capture([self.ref()]) + descriptor = await self.media(start) + path.unlink() + response, page = await self.outputs(start) + self.assertEqual(page["items"][0]["current_candidate"]["reason"], "missing") + for field in ("url", "thumbnail_url", "inspection_url"): + response = await self.current(descriptor[field]) + self.assert_error(response, await response.json(), 404, "not-found") + path.mkdir() + _, page = await self.outputs(start) + self.assertEqual( + page["items"][0]["current_candidate"]["reason"], "not_regular_file" + ) + path.rmdir() + path.write_bytes(b"media") + with patch("os.access", return_value=False): + _, page = await self.outputs(start) + self.assertEqual( + page["items"][0]["current_candidate"]["reason"], "unreadable" + ) + response = await self.current(descriptor["url"]) + self.assert_error(response, await response.json(), 403, "not-authorized") + path.unlink() + outside = make_png(self.temp_dir / "private.png") + path.symlink_to(outside) + _, page = await self.outputs(start) + self.assertEqual( + page["items"][0]["current_candidate"]["reason"], "not_authorized" + ) + unsupported = self.capture([self.ref(filename="notes.txt")], item=1) + (self.output_dir / "notes.txt").symlink_to(outside) + self.api._get_all_output_dirs = lambda: [self.output_dir, self.temp_dir] + _, page = await self.outputs(unsupported) + self.assertEqual( + page["items"][0]["current_candidate"]["reason"], "unsupported_media" + ) + + async def test_scope_change_during_metadata_read_and_open_rejects_before_headers( + self, + ): + png(self.output_dir / "image.png", prompt=graph()) + start = self.capture([self.ref()]) + descriptor = await self.media(start) + service = importlib.import_module("py.api.inspection_service") + read = service.read_inspection_file + + def changed(path): + result = read(path) + self.host_output = str(self.temp_dir) + return result + + with patch.object(service, "read_inspection_file", side_effect=changed): + response = await self.current(descriptor["inspection_url"]) + self.assert_error(response, await response.json(), 409, "candidate-stale") + self.host_output = str(self.output_dir) + mod = importlib.import_module("py.api.history_current") + opened = mod._open_current_target + handles = [] + + def changed_open(*args, **kwargs): + handle = opened(*args, **kwargs) + handles.append(handle) + self.host_output = str(self.temp_dir) + return handle + + with patch.object(mod, "_open_current_target", side_effect=changed_open): + response = await self.current(descriptor["url"]) + self.assert_error(response, await response.json(), 409, "candidate-stale") + self.assertTrue(all(handle.closed for handle in handles)) + + async def test_canonical_target_symlink_retarget_at_open_never_opens_replacement( + self, + ): + path = make_png(self.output_dir / "image.png") + outside = make_png(self.temp_dir / "private.png") + start = self.capture([self.ref()]) + descriptor = await self.media(start) + mod = importlib.import_module("py.api.history_current") + opened = mod._open_current_target + + def retarget(target, *args, **kwargs): + path.unlink() + path.symlink_to(outside) + return opened(target, *args, **kwargs) + + with patch.object(mod, "_open_current_target", side_effect=retarget): + response = await self.current(descriptor["url"]) + self.assert_error(response, await response.json(), 409, "candidate-stale") + + async def test_late_open_handle_closes_when_request_is_cancelled_during_executor( + self, + ): + make_png(self.output_dir / "image.png") + start = self.capture([self.ref()]) + descriptor = await self.media(start) + request = make_mocked_request( + "GET", descriptor["url"], match_info={"run_id": start.run_id} + ) + mod = importlib.import_module("py.api.history_current") + history = importlib.import_module("py.api.history") + opened = mod._open_current_target + discard = history._close_prepared_result + reached, closed = asyncio.Event(), asyncio.Event() + release = threading.Event() + loop = asyncio.get_running_loop() + handles = [] + + def barrier(*args, **kwargs): + handle = opened(*args, **kwargs) + handles.append(handle) + loop.call_soon_threadsafe(reached.set) + if not release.wait(5): + handle.close() + raise AssertionError("late-open barrier timed out") + return handle + + def discarded(task): + discard(task) + closed.set() + + with ( + patch.object(mod, "_open_current_target", side_effect=barrier), + patch.object(history, "_close_prepared_result", side_effect=discarded), + ): + pending = asyncio.create_task(self.api.get_history_current_media(request)) + try: + await asyncio.wait_for(reached.wait(), 5) + pending.cancel() + with self.assertRaises(asyncio.CancelledError): + await pending + finally: + release.set() + await asyncio.wait_for(closed.wait(), 5) + self.assertEqual(len(handles), 1) + self.assertTrue(handles[0].closed) + + async def test_opened_handle_closes_if_stream_is_cancelled_before_headers(self): + make_png(self.output_dir / "image.png") + start = self.capture([self.ref()]) + descriptor = await self.media(start) + request = make_mocked_request( + "GET", descriptor["url"], match_info={"run_id": start.run_id} + ) + mod = importlib.import_module("py.api.history_current") + opened = mod._open_current_target + handles = [] + + def tracked(*args, **kwargs): + handle = opened(*args, **kwargs) + handles.append(handle) + return handle + + with ( + patch.object(mod, "_open_current_target", side_effect=tracked), + patch.object( + web.StreamResponse, "prepare", side_effect=asyncio.CancelledError + ), + ): + with self.assertRaises(asyncio.CancelledError): + await self.api.get_history_current_media(request) + self.assertEqual(len(handles), 1) + self.assertTrue(handles[0].closed) + + async def test_restore_after_final_fence_finishes_coherent_already_opened_media( + self, + ): + path = make_png(self.output_dir / "image.png") + expected = path.read_bytes() + start = self.capture([self.ref()]) + descriptor = await self.media(start) + source = self.restore_fixture() + fence = self.api._history_fence() + validate = fence.validate + + def after(epoch, generation): + validate(epoch, generation) + self.api.db.model.restore_from_file(source) + path.unlink() + + with patch.object(fence, "validate", side_effect=after): + response = await self.current(descriptor["url"]) + self.assertEqual(response.status, 200) + self.assertEqual(await response.read(), expected) + response = await self.current(descriptor["url"]) + self.assert_error(response, await response.json(), 409, "history-changed") + + async def test_current_inspection_shares_response_limit_and_saved_display_bound( + self, + ): + path = png(self.output_dir / "image.png", prompt=graph("File evidence")) + image = self._link_image(path, "Saved text is never frozen-run text") + with self.api.db.model.get_connection() as conn: + conn.execute( + "UPDATE prompts SET category=? WHERE id=" + "(SELECT prompt_id FROM generated_images WHERE id=?)", + ("猫" * 12000, image), + ) + start = self.capture([self.ref()]) + descriptor = await self.media(start) + response = await self.current(descriptor["inspection_url"]) + body = await response.json() + self.assertEqual(response.status, 200, body) + self.assertEqual(body["library"]["status"], "limited") + self.assertEqual(body["prompts"]["positive"]["text"], "File evidence") + with patch("utils.inspection_response.MAX_RESPONSE_BYTES", 1): + response = await self.current(descriptor["inspection_url"]) + self.assert_error(response, await response.json(), 422, "response-limit") + + async def test_optional_candidate_failure_preserves_every_occurrence_without_urls( + self, + ): + make_png(self.output_dir / "image.png") + start = self.capture([self.ref("1"), self.ref("2")]) + self.assertEqual((await self.outputs(start))[0].status, 200) + history = importlib.import_module("py.api.history") + with patch.object( + history, "resolve_current_candidate", side_effect=OSError("private/path") + ): + response, page = await self.outputs(start) + self.assertEqual(response.status, 200, page) + self.assertEqual(len(page["items"]), 2) + self.assertEqual(page["retained_occurrence_count"], 2) + for receipt in page["items"]: + self.assertEqual( + receipt["current_candidate"], + { + "status": "error", + "reason": "temporary_unavailable", + "evidence_class": "locator_candidate", + }, + ) + self.assertEqual(receipt["file_identity"], "unverified") + self.assertNotIn("private", json.dumps(page)) + + async def test_page_observes_at_most_one_hundred_distinct_targets(self): + for ordinal in range(105): + make_png(self.output_dir / f"image-{ordinal}.png") + start = self.capture([self.ref("1", i, f"image-{i}.png") for i in range(105)]) + observed = [] + stat = Path.stat + + def traced(path, *args, **kwargs): + if path.parent == self.output_dir and path.name.startswith("image-"): + observed.append(path.name) + return stat(path, *args, **kwargs) + + with patch.object(Path, "stat", traced): + response, page = await self.outputs( + start, {"epoch": start.storage_epoch, "limit": "100"} + ) + self.assertEqual(response.status, 200, page) + self.assertEqual(len(page["items"]), 100) + self.assertEqual(observed, [f"image-{i}.png" for i in range(100)]) + + async def test_inspection_retarget_during_read_rejects_metadata(self): + path = png(self.output_dir / "image.png", prompt=graph("Authorized")) + outside = png(self.temp_dir / "private.png", prompt=graph("Private")) + start = self.capture([self.ref()]) + descriptor = await self.media(start) + service = importlib.import_module("py.api.inspection_service") + read = service.read_inspection_file + + def changed(target): + result = read(target) + path.unlink() + path.symlink_to(outside) + return result + + with patch.object(service, "read_inspection_file", side_effect=changed): + response = await self.current(descriptor["inspection_url"]) + body = await response.json() + self.assert_error(response, body, 409, "candidate-stale") + self.assertNotIn("Authorized", json.dumps(body)) + self.assertNotIn("Private", json.dumps(body)) + + async def test_stale_epoch_rejects_urls_without_unbounded_saved_lookup(self): + make_png(self.output_dir / "image.png") + start = self.capture([self.ref()]) + with patch.object( + self.api.db, + "get_image_prompt_info", + side_effect=AssertionError("unbounded convenience lookup"), + ): + descriptor = await self.media(start) + response = await self.current(descriptor["inspection_url"]) + self.assertEqual(response.status, 200) + await response.read() + source = self.restore_fixture() + self.api.db.model.restore_from_file(source) + for field in ("url", "thumbnail_url", "inspection_url"): + response = await self.current(descriptor[field]) + self.assert_error(response, await response.json(), 409, "history-changed") + + async def test_existing_authorized_preview_alias_is_allowed_and_escape_is_denied( + self, + ): + make_png(self.output_dir / "image.png") + preview = make_png( + self.output_dir / "previews" / "image_thumb.png", color=(0, 255, 0) + ) + alias = self.output_dir / "thumbnails" + alias.symlink_to(preview.parent, target_is_directory=True) + start = self.capture([self.ref()]) + descriptor = await self.media(start) + response = await self.current(descriptor["thumbnail_url"]) + self.assertEqual(response.status, 200, await response.read()) + self.assertEqual(await response.read(), preview.read_bytes()) + make_png(self.temp_dir / "image_thumb.png") + alias.unlink() + alias.symlink_to(self.temp_dir, target_is_directory=True) + response = await self.current(descriptor["thumbnail_url"]) + self.assert_error(response, await response.json(), 403, "not-authorized") + + async def test_disappearance_before_open_is_missing_and_closes_handle( + self, + ): + path = make_png(self.output_dir / "image.png") + start = self.capture([self.ref()]) + descriptor = await self.media(start) + mod = importlib.import_module("py.api.history_current") + opened = mod._open_current_target + + def removed(target, *args, **kwargs): + path.unlink() + return opened(target, *args, **kwargs) + + with patch.object(mod, "_open_current_target", side_effect=removed): + response = await self.current(descriptor["url"]) + self.assert_error(response, await response.json(), 404, "not-found") + + async def test_unexpected_inspection_failure_is_bounded_and_does_not_log_paths( + self, + ): + make_png(self.output_dir / "image.png") + start = self.capture([self.ref()]) + descriptor = await self.media(start) + private = "SECRET_PROMPT /private/root/image.png" + with patch( + "py.api.inspection_service.read_inspection_file", + side_effect=RuntimeError(private), + ): + with self.assertLogs("prompt_manager.api", "WARNING") as logs: + response = await self.current(descriptor["inspection_url"]) + self.assertEqual(response.content_type, "application/json") + body = await response.json() + self.assert_error(response, body, 503, "storage-unavailable") + self.assertNotIn(private, json.dumps(body) + " ".join(logs.output)) + + async def test_changed_epoch_or_scope_wins_over_failed_metadata_observation(self): + png(self.output_dir / "image.png", prompt=graph()) + service = importlib.import_module("py.api.inspection_service") + read = service.read_inspection_file + for item, code in enumerate(("history-changed", "candidate-stale")): + with self.subTest(code=code): + start = self.capture([self.ref()], item=item) + descriptor = await self.media(start) + source = self.restore_fixture() + + def failed(path): + read(path) + if code == "history-changed": + self.api.db.model.restore_from_file(source) + else: + self.host_output = str(self.temp_dir) + raise RuntimeError("private observation failure") + + with patch.object(service, "read_inspection_file", side_effect=failed): + response = await self.current(descriptor["inspection_url"]) + self.assert_error(response, await response.json(), 409, code) + + async def test_current_media_open_io_failure_is_retryable_without_private_paths( + self, + ): + make_png(self.output_dir / "image.png") + start = self.capture([self.ref()]) + descriptor = await self.media(start) + response = await self.current(descriptor["url"]) + self.assertEqual(response.status, 200) + await response.read() + with patch( + "py.api.history_current.os.open", + side_effect=OSError(errno.EIO, "private/root"), + ): + response = await self.current(descriptor["url"]) + body = await response.json() + self.assert_error(response, body, 503, "storage-unavailable") + self.assertEqual(body["recovery"], {"action": "retry-read"}) diff --git a/tests/test_api_images.py b/tests/test_api_images.py index 5221c823..2e5d0739 100644 --- a/tests/test_api_images.py +++ b/tests/test_api_images.py @@ -2097,7 +2097,7 @@ class TestRootScopeServing(ImageAPITestCase): side_effect=AssertionError("saved lookup"), ), patch( - "py.api.images.read_inspection_file", + "py.api.inspection_service.read_inspection_file", side_effect=AssertionError("metadata read"), ), patch( diff --git a/tests/test_api_prompt_inspector.py b/tests/test_api_prompt_inspector.py index a297b5aa..ea6743cc 100644 --- a/tests/test_api_prompt_inspector.py +++ b/tests/test_api_prompt_inspector.py @@ -347,7 +347,7 @@ class TestPromptInspector(ImageAPITestCase): self.assertEqual(response.status, 200) self.assertEqual(result["prompts"]["positive"]["status"], "absent") with patch( - "py.api.images.read_inspection_file", + "py.api.inspection_service.read_inspection_file", side_effect=RuntimeError(str(self.comfy_dir)), ): response, result = await self.inspect("image.jpg") @@ -1044,7 +1044,7 @@ class TestPromptInspector(ImageAPITestCase): self.api, "_file_response", side_effect=AssertionError("file response") ), patch( - "py.api.images.read_inspection_file", + "py.api.inspection_service.read_inspection_file", side_effect=AssertionError("metadata read"), ), ): diff --git a/tests/test_history_current.py b/tests/test_history_current.py new file mode 100644 index 00000000..dfc62a89 --- /dev/null +++ b/tests/test_history_current.py @@ -0,0 +1,469 @@ +"""Current locations require a native namespace and independent authorization.""" + +import importlib +import os +import sqlite3 +import subprocess +import sys +import tempfile +import unittest +from contextlib import ExitStack +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +from tests.test_api_images import make_png +from database.operations import PromptDatabase +from py.config import GalleryConfig +from utils.execution_outputs import snapshot_output_roots + + +class CurrentLocationTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.base = Path(self.tmp.name).resolve() + self.output = self.base / "output" + self.temp = self.base / "temp" + self.output.mkdir() + self.temp.mkdir() + self.roots = [self.output] + self.host_output = str(self.output) + self.host_temp = str(self.temp) + self.stack = ExitStack() + self.addCleanup(self.stack.close) + self.stack.enter_context( + patch( + "folder_paths.get_output_directory", + lambda: self.host_output, + create=True, + ) + ) + self.stack.enter_context( + patch( + "folder_paths.get_temp_directory", lambda: self.host_temp, create=True + ) + ) + self.stack.enter_context( + patch.object( + GalleryConfig, "allowed_gallery_parents", lambda: [str(self.base)] + ) + ) + self.stack.enter_context( + patch.object(GalleryConfig, "path_anchors", lambda: [str(self.base)]) + ) + self.api = SimpleNamespace(_get_all_output_dirs=lambda: list(self.roots)) + + def module(self, name="py.api.history_current"): + self.assertIsNotNone( + importlib.util.find_spec(name), f"missing current-location adapter {name}" + ) + return importlib.import_module(name) + + def reference(self, **overrides): + result = { + "folder_type": "output", + "subfolder": "", + "filename": "image.png", + "path_style": "windows" if os.name == "nt" else "posix", + "root_key": snapshot_output_roots( + "windows" if os.name == "nt" else "posix", + self.host_output, + self.host_temp, + ).output_key, + } + result.update(overrides) + return result + + def resolve(self, reference=None, gallery=None): + mod = self.module() + scope = mod.snapshot_current_output_scope(self.api) + return mod.resolve_current_candidate( + reference or self.reference(), scope, gallery or {} + ).public + + def test_lexical_alias_is_not_replaced_by_canonical_root_key(self): + mod = self.module() + path = make_png(self.output / "image.png") + alias = self.base / "alias" + alias.symlink_to(self.output, target_is_directory=True) + self.host_output = str(alias) + reference = self.reference() + scope = mod.snapshot_current_output_scope(self.api) + with patch.object( + Path, "stat", side_effect=AssertionError("locator performed IO") + ): + self.assertEqual( + mod.candidate_locator(reference, scope), str(alias / path.name) + ) + candidate = mod.resolve_current_candidate(reference, scope, {}).public + self.assertEqual(candidate["status"], "available") + self.assertEqual(candidate["gallery_match"], "none") + self.assertRegex(candidate["location_key"], r"^[0-9a-f]{64}$") + wrong = self.reference( + root_key=snapshot_output_roots("posix", str(self.output), None).output_key + ) + self.assertEqual( + mod.resolve_current_candidate(wrong, scope, {}).public["reason"], + "namespace_mismatch", + ) + + def test_null_foreign_relocated_and_unsafe_descriptors_have_allowed_control(self): + make_png(self.output / "image.png") + self.assertEqual(self.resolve()["status"], "available") + for changes, reason in ( + ({"root_key": None}, "root_unavailable"), + ({"root_key": "a" * 64}, "namespace_mismatch"), + ( + {"path_style": "windows", "subfolder": "sub", "root_key": "b" * 64}, + "foreign_path_style", + ), + ({"subfolder": "../outside"}, "namespace_mismatch"), + ({"filename": "/absolute.png"}, "namespace_mismatch"), + ): + with self.subTest(changes=changes): + candidate = self.resolve(self.reference(**changes)) + self.assertEqual( + (candidate["status"], candidate["reason"]), ("unavailable", reason) + ) + self.assertNotIn("media", candidate) + + def test_temp_requires_current_gallery_authorization(self): + make_png(self.temp / "image.png") + reference = self.reference( + folder_type="temp", + root_key=snapshot_output_roots( + "posix", self.host_output, self.host_temp + ).temp_key, + ) + self.assertEqual(self.resolve(reference)["reason"], "not_authorized") + self.roots.append(self.temp) + self.assertEqual(self.resolve(reference)["status"], "available") + + def test_supported_regular_readable_control_and_distinct_failures(self): + path = make_png(self.output / "image.png") + self.assertEqual(self.resolve()["status"], "available") + path.unlink() + self.assertEqual(self.resolve()["reason"], "missing") + path.mkdir() + self.assertEqual(self.resolve()["reason"], "not_regular_file") + path.rmdir() + path.write_bytes(b"media") + with patch("os.access", return_value=False): + self.assertEqual(self.resolve()["reason"], "unreadable") + (self.output / "notes.txt").write_text("private") + self.assertEqual( + self.resolve(self.reference(filename="notes.txt"))["reason"], + "unsupported_media", + ) + + def test_symlink_escape_never_authorizes_another_namespace(self): + allowed = make_png(self.output / "image.png") + outside = make_png(self.temp / "outside.png") + self.roots.append(self.temp) + self.assertEqual(self.resolve()["status"], "available") + allowed.unlink() + allowed.symlink_to(outside) + self.assertEqual(self.resolve()["reason"], "not_authorized") + + def test_scope_tracks_configuration_targets_and_policy_but_not_bytes(self): + mod = self.module() + path = make_png(self.output / "image.png") + first = mod.snapshot_current_output_scope(self.api) + location = mod.resolve_current_candidate(self.reference(), first, {}).public[ + "location_key" + ] + path.write_bytes(b"replacement") + same = mod.snapshot_current_output_scope(self.api) + self.assertEqual(same.current_scope, first.current_scope) + self.assertEqual( + mod.resolve_current_candidate(self.reference(), same, {}).public[ + "location_key" + ], + location, + ) + self.roots.append(self.temp) + changed = mod.snapshot_current_output_scope(self.api) + self.assertNotEqual(changed.current_scope, first.current_scope) + self.roots.reverse() + self.assertEqual( + mod.snapshot_current_output_scope(self.api).current_scope, + changed.current_scope, + ) + with patch.object( + GalleryConfig, "allowed_gallery_parents", lambda: [str(self.output)] + ): + self.assertNotEqual( + mod.snapshot_current_output_scope(self.api).current_scope, + changed.current_scope, + ) + alias = self.base / "alias" + alias.symlink_to(self.output, target_is_directory=True) + self.host_output = str(alias) + before = mod.snapshot_current_output_scope(self.api) + alias.unlink() + alias.symlink_to(self.temp, target_is_directory=True) + self.assertNotEqual( + mod.snapshot_current_output_scope(self.api).current_scope, + before.current_scope, + ) + + def test_gallery_projection_is_snapshot_owned_exact_and_ambiguous(self): + mod = self.module("database.history_gallery") + db = PromptDatabase(str(self.base / "gallery.db")) + self.addCleanup(db.close_all) + path = make_png(self.output / "image.png") + prompt = db.save_prompt("first", prompt_hash="first-hash") + other = db.save_prompt("second", prompt_hash="second-hash") + image = db.link_image_to_prompt(prompt, str(path)) + db.link_image_to_prompt(other, str(self.temp / path.name)) + conn = db.model.get_connection() + with self.assertRaises(Exception): + mod.read_history_gallery(conn, (str(path),)) + with db.recorded_history_snapshot(): + projected = mod.read_history_gallery(conn, (str(path),)) + self.assertTrue(conn.in_transaction) + self.assertEqual(projected[str(path)]["gallery_match"], "unique") + self.assertEqual(projected[str(path)]["gallery_image_id"], image) + db.link_image_to_prompt(other, str(path)) + with db.recorded_history_snapshot(): + projected = mod.read_history_gallery(conn, (str(path),)) + self.assertEqual(projected[str(path)]["gallery_match"], "ambiguous") + self.assertIsNone(projected[str(path)]["gallery_image_id"]) + self.assertIsNone(projected[str(path)]["saved"]) + self.assertEqual(projected[str(path)]["library"]["status"], "ambiguous") + with self.assertRaises(ValueError): + mod.read_history_gallery( + conn, tuple(str(path) + str(i) for i in range(101)) + ) + + def test_gallery_display_is_bounded_without_loading_workflow_or_prompt(self): + mod = self.module("database.history_gallery") + db = PromptDatabase(str(self.base / "gallery.db")) + self.addCleanup(db.close_all) + path = make_png(self.output / "image.png") + prompt = db.save_prompt( + "secret prompt" * 10000, prompt_hash="large-hash", category="猫" * 12000 + ) + image = db.link_image_to_prompt(prompt, str(path)) + with db.model.get_connection() as conn: + conn.execute( + "UPDATE generated_images SET workflow_data=?,prompt_metadata=? " + "WHERE id=?", + ("private" * 100000, "private" * 100000, image), + ) + statements = [] + conn.set_trace_callback(statements.append) + self.addCleanup(conn.set_trace_callback, None) + with db.recorded_history_snapshot(): + result = mod.read_history_gallery(conn, (str(path),)) + display = result[str(path)]["library"] + self.assertEqual(display["status"], "limited") + self.assertEqual(display["tags"], []) + self.assertFalse( + any( + "workflow_data" in sql or "prompt_metadata" in sql or "p.text" in sql + for sql in statements + ) + ) + self.assertFalse(any(sql.startswith("COMMIT") for sql in statements)) + + def test_gallery_queries_stop_after_two_owners_and_two_pair_associations(self): + mod = self.module("database.history_gallery") + db = PromptDatabase(str(self.base / "gallery.db")) + self.addCleanup(db.close_all) + path = make_png(self.output / "image.png") + unique = make_png(self.output / "unique.png") + prompt = db.save_prompt("owner", prompt_hash="owner") + image = db.link_image_to_prompt(prompt, str(unique)) + with db.model.get_connection() as conn: + for i in range(12): + owner = conn.execute( + "INSERT INTO prompts(text,hash) VALUES(?,?)", + (str(i), str(i)), + ).lastrowid + pair = conn.execute( + "INSERT INTO prompt_pairs" + "(positive_prompt_id,negative_prompt_id,title) " + "VALUES(?,?,?)", + (prompt, owner, str(i)), + ).lastrowid + conn.execute( + "INSERT INTO image_prompt_pairs(image_id,pair_id,provenance) " + "VALUES(?,?,'{}')", + (image, pair), + ) + conn.execute( + "INSERT INTO generated_images" + "(prompt_id,image_path,filename,file_path) " + "VALUES(?,?,'image.png',?)", + (owner, str(path), str(path)), + ) + fetched = {"owners": 0, "pairs": 0} + + def row_factory(cursor, values): + names = {column[0] for column in cursor.description} + if names == {"id", "prompt_id", "category"}: + fetched["owners"] += 1 + elif names == {"id", "positive_prompt_id", "title", "category"}: + fetched["pairs"] += 1 + return sqlite3.Row(cursor, values) + + conn.row_factory = row_factory + with db.recorded_history_snapshot(): + result = mod.read_history_gallery(conn, (str(path), str(unique))) + self.assertEqual(fetched, {"owners": 3, "pairs": 2}) + self.assertEqual(result[str(path)]["gallery_match"], "ambiguous") + self.assertEqual(result[str(unique)]["gallery_match"], "unique") + self.assertEqual(result[str(unique)]["library"]["status"], "limited") + self.assertIsNone(result[str(unique)]["library"].get("pair_id")) + + def test_host_root_failure_and_removal_do_not_search_another_root(self): + make_png(self.output / "image.png") + make_png(self.temp / "image.png") + self.roots.append(self.temp) + reference = self.reference() + self.assertEqual(self.resolve(reference)["status"], "available") + with patch("folder_paths.get_output_directory", side_effect=OSError("private")): + self.assertEqual(self.resolve(reference)["reason"], "root_unavailable") + (self.output / "image.png").unlink() + self.output.rmdir() + self.assertEqual(self.resolve(reference)["reason"], "root_unavailable") + + def test_unusable_lexical_namespace_changes_still_change_scope(self): + mod = self.module() + make_png(self.output / "image.png") + reference = self.reference() + self.assertEqual(self.resolve(reference)["status"], "available") + self.host_output = "relative-one" + first = mod.snapshot_current_output_scope(self.api) + self.assertEqual( + mod.resolve_current_candidate(reference, first, {}).public["reason"], + "root_unavailable", + ) + self.host_output = "relative-two" + self.assertNotEqual( + mod.snapshot_current_output_scope(self.api).current_scope, + first.current_scope, + ) + + def test_duplicate_owner_lookup_does_not_sort_the_entire_saved_population(self): + mod = self.module("database.history_gallery") + db = PromptDatabase(str(self.base / "gallery.db")) + self.addCleanup(db.close_all) + path = str(self.output / "image.png") + with db.model.get_connection() as conn: + for item in range(512): + owner = conn.execute( + "INSERT INTO prompts(text) VALUES(?)", (str(item),) + ).lastrowid + conn.execute( + "INSERT INTO generated_images" + "(prompt_id,image_path,filename,file_path) " + "VALUES(?,?,'image.png',?)", + (owner, path, path), + ) + steps = 0 + + def bounded(): + nonlocal steps + steps += 1 + return int(steps > 20) + + conn.set_progress_handler(bounded, 100) + try: + with db.recorded_history_snapshot(): + result = mod.read_history_gallery(conn, (path,)) + except sqlite3.OperationalError as error: + self.fail(f"Two-owner lookup exceeded bounded query work: {error}") + finally: + conn.set_progress_handler(None, 0) + self.assertEqual(result[path]["gallery_match"], "ambiguous") + + def test_two_same_pair_links_and_later_different_pair_withhold_pair_choice(self): + mod = self.module("database.history_gallery") + db = PromptDatabase(str(self.base / "gallery.db")) + self.addCleanup(db.close_all) + path = make_png(self.output / "image.png") + pair = db.save_prompt_pair("owner", "negative one") + other_pair = db.save_prompt_pair("owner", "negative two") + owner = db.get_prompt_pair(pair)["positive_prompt_id"] + image = db.link_image_to_prompt(owner, str(path)) + with db.model.get_connection() as conn: + for pair_id, provenance in ( + (pair, "{}"), + (pair, '{"source":"second"}'), + (other_pair, "{}"), + ): + conn.execute( + "INSERT INTO image_prompt_pairs(image_id,pair_id,provenance) " + "VALUES(?,?,?)", + (image, pair_id, provenance), + ) + with db.recorded_history_snapshot(): + result = mod.read_history_gallery(conn, (str(path),))[str(path)] + self.assertEqual(result["gallery_match"], "unique") + self.assertEqual(result["gallery_image_id"], image) + self.assertEqual(result["library"]["status"], "limited") + self.assertIsNone(result["library"].get("pair_id")) + self.assertIsNone(result["saved"].get("pair")) + + +class HistoryImportTests(unittest.TestCase): + def test_api_import_does_not_require_a_running_host_server(self): + result = subprocess.run( + [ + sys.executable, + "-c", + "import sys, types; " + "sys.modules['folder_paths'] = types.ModuleType('folder_paths'); " + "import py.api", + ], + cwd=Path(__file__).resolve().parents[1], + capture_output=True, + text=True, + check=False, + ) + self.assertEqual(result.returncode, 0, result.stderr) + + def test_minimal_host_fixture_restores_getters_after_real_route_control(self): + program = """ +import io +from pathlib import Path +import sys +import types +import unittest + +host = types.ModuleType("folder_paths") +original = lambda: "unused-minimal-host-output" +host.get_output_directory = original +sys.modules["folder_paths"] = host +from tests.test_api_history_outputs import HistoryOutputAPITests +case = HistoryOutputAPITests( + "test_complete_zero_incomplete_unavailable_and_not_captured_are_distinct" +) +try: + runner = unittest.TextTestRunner(stream=io.StringIO()) + result = runner.run(unittest.TestSuite([case])) + assert host.get_output_directory is original, "fixture leaked host getter" + assert not hasattr(host, "get_temp_directory"), "fixture leaked optional getter" + assert result.wasSuccessful(), result.errors +finally: + if hasattr(case, "api"): + case.api.db.close_all() + if hasattr(case, "_anchor_patch"): + case._anchor_patch.stop() + if hasattr(case, "_temp_db"): + for suffix in ("", "-wal", "-shm"): + Path(case._temp_db.name + suffix).unlink(missing_ok=True) + if hasattr(case, "_output_tmp"): + case._output_tmp.cleanup() +""" + result = subprocess.run( + [sys.executable, "-c", program], + cwd=Path(__file__).resolve().parents[1], + capture_output=True, + text=True, + check=False, + ) + self.assertEqual(result.returncode, 0, result.stderr) diff --git a/tests/test_inspection_response.py b/tests/test_inspection_response.py index 0488abb7..79c81d6a 100644 --- a/tests/test_inspection_response.py +++ b/tests/test_inspection_response.py @@ -463,3 +463,45 @@ class TestInspectionResponse(unittest.TestCase): ) self.assertEqual(result["metadata"]["raw"], read.metadata) self.assertEqual(result["prompts"], PROMPTS) + + def test_history_identity_shares_rooted_evidence_and_library_projection(self): + from utils.inspection_response import encode_history_inspection_response + + read = inspection({"prompt": graph_text()}) + identity = { + "occurrence": { + "run_id": "run", + "output_node_id": "node", + "image_ordinal": 2, + }, + "reference": {"filename": "image.png"}, + "history_epoch": "epoch", + "current_scope": "scope", + "location_key": "location", + "file_identity": "unverified", + } + body = json.loads( + encode_history_inspection_response( + current_output=identity, + file={"size": 7, "observation": "current-file"}, + prompts=PROMPTS, + resolution={"state": "resolved"}, + inspection=read, + library={"status": "linked", "tags": ["猫" * 12000]}, + ) + ) + rooted = json.loads( + encode( + read, + file={"size": 7, "observation": "current-file"}, + library={"status": "linked", "tags": ["猫" * 12000]}, + ) + ) + self.assertEqual(body["current_output"], identity) + self.assertEqual(body.get("history_version"), 1) + self.assertEqual(body.get("history_epoch"), "epoch") + for key in ("file", "prompts", "metadata", "library"): + self.assertEqual(body[key], rooted[key]) + self.assertEqual(body["library"]["status"], "limited") + self.assertNotIn("image", body) + self.assertNotIn("root_scope", body) diff --git a/utils/inspection_response.py b/utils/inspection_response.py index b7f51629..a7cc6411 100644 --- a/utils/inspection_response.py +++ b/utils/inspection_response.py @@ -237,10 +237,9 @@ def _library_projection(library: dict) -> dict: return result -def encode_inspection_response( +def _encode_inspection( *, - image: dict, - root_scope: str, + identity: dict, file: dict, prompts: dict, resolution: dict, @@ -252,8 +251,7 @@ def encode_inspection_response( response = { "success": True, "inspection_version": 1, - "root_scope": root_scope, - "image": image, + **identity, "file": _file_projection(file, inspection), "prompts": prompts, "resolution": resolution, @@ -277,3 +275,48 @@ def encode_inspection_response( if len(body) > MAX_RESPONSE_BYTES: raise ValueError("Inspection response exceeds limits") return body + + +def encode_inspection_response( + *, + image: dict, + root_scope: str, + file: dict, + prompts: dict, + resolution: dict, + inspection: "InspectionRead | None", + library: dict, +) -> bytes: + """Preserve the existing rooted wire and its bounded metadata projection.""" + return _encode_inspection( + identity={"root_scope": root_scope, "image": image}, + file=file, + prompts=prompts, + resolution=resolution, + inspection=inspection, + library=library, + ) + + +def encode_history_inspection_response( + *, + current_output: dict, + file: dict, + prompts: dict, + resolution: dict, + inspection: "InspectionRead | None", + library: dict, +) -> bytes: + """Use occurrence identity with the same current-file evidence projection.""" + return _encode_inspection( + identity={ + "history_version": 1, + "history_epoch": current_output["history_epoch"], + "current_output": current_output, + }, + file=file, + prompts=prompts, + resolution=resolution, + inspection=inspection, + library=library, + )