Vito
4ef236aaf7
* feat(history): define bounded output reference contracts * feat(database): migrate execution output evidence to schema v3 * feat(history): persist immutable run output references * test(history): isolate mocked node gallery monitor * feat(history): capture bounded output descriptors with shared leases * fix(history): authenticate payload owners and release failed captures * feat(history): attach output evidence to exact finalized attempts * test(history): verify run output evidence on real ComfyUI * fix(history): release idle worker payload references * docs(history): clarify defensive lease cleanup * test(history): align output fixtures with native path styles * test(history): separate restore completion from writer drain * test(history): bound batch drains by progress and total time
372 lines
14 KiB
Python
372 lines
14 KiB
Python
"""Exact, bounded output locators carry no authority to access files."""
|
|
|
|
import hashlib
|
|
import unittest
|
|
from dataclasses import FrozenInstanceError, replace
|
|
from unittest.mock import patch
|
|
|
|
from test_execution_contexts import load_module
|
|
|
|
|
|
class OutputContractTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.m = load_module("execution_outputs")
|
|
|
|
def reference(self, **changes) -> object:
|
|
reference = self.m.OutputReference(
|
|
"7", 0, "output", "", "same.png", "posix", "a" * 64
|
|
)
|
|
return replace(reference, **changes)
|
|
|
|
def test_exact_unicode_and_duplicate_locators_survive(self) -> None:
|
|
m = self.m
|
|
a = self.reference()
|
|
b = replace(a, image_ordinal=1)
|
|
bundle = m.OutputBundle((a, b), "complete", "reported")
|
|
self.assertIsNone(m.validate_output_bundle(bundle))
|
|
self.assertEqual(m.output_payload_size(bundle), 552)
|
|
self.assertEqual(
|
|
m.output_payload_size(bundle),
|
|
128
|
|
+ sum(
|
|
128
|
|
+ sum(
|
|
load_module("execution_contexts").text_size(v, 4096)
|
|
for v in (
|
|
r.output_node_id,
|
|
r.folder_type,
|
|
r.subfolder,
|
|
r.filename,
|
|
r.path_style,
|
|
r.root_key or "",
|
|
)
|
|
)
|
|
for r in bundle.references
|
|
),
|
|
)
|
|
variants = ("é.png", "e\u0301.png", "É.png", " é.png", "é.png ")
|
|
references = tuple(
|
|
self.reference(image_ordinal=i, filename=name, subfolder=" 雪 / Café ")
|
|
for i, name in enumerate(variants)
|
|
)
|
|
exact = m.OutputBundle(references, "complete", "reported")
|
|
self.assertIsNone(m.validate_output_bundle(exact))
|
|
self.assertEqual(tuple(r.filename for r in exact.references), variants)
|
|
self.assertEqual(exact.references[0].subfolder, " 雪 / Café ")
|
|
self.assertEqual(
|
|
m.output_payload_size(m.OutputBundle((), "complete", "none_reported")), 128
|
|
)
|
|
with self.assertRaises(FrozenInstanceError):
|
|
a.filename = "changed.png"
|
|
|
|
def test_posix_backslash_is_literal(self) -> None:
|
|
m = self.m
|
|
reference = self.reference(subfolder=r"a\b/雪", filename=r"a\b:c.png")
|
|
self.assertIsNone(m.validate_output_reference(reference))
|
|
self.assertEqual(reference.filename, r"a\b:c.png")
|
|
for subfolder in (
|
|
"/absolute",
|
|
"../escape",
|
|
"a/../b",
|
|
"./a",
|
|
"a//b",
|
|
"a/",
|
|
".",
|
|
"..",
|
|
):
|
|
with self.subTest(subfolder=subfolder), self.assertRaises(ValueError):
|
|
m.validate_output_reference(replace(reference, subfolder=subfolder))
|
|
for filename in ("", "/absolute", "a/b", ".", "..", "a\x00b", "\ud800"):
|
|
with self.subTest(filename=repr(filename)), self.assertRaises(ValueError):
|
|
m.validate_output_reference(replace(reference, filename=filename))
|
|
|
|
def test_windows_drive_unc_traversal_ads_rejected(self) -> None:
|
|
m = self.m
|
|
reference = self.reference(
|
|
path_style="windows", subfolder=r"a\雪/b", filename=" image.png "
|
|
)
|
|
self.assertIsNone(m.validate_output_reference(reference))
|
|
for subfolder in (
|
|
r"C:\absolute",
|
|
"C:relative",
|
|
r"\\server\share",
|
|
r"\rooted",
|
|
"/rooted",
|
|
"../escape",
|
|
r"a\..\b",
|
|
r"a\.\b",
|
|
r"a\\b",
|
|
"a//b",
|
|
r"a:stream",
|
|
"a/",
|
|
"a\\",
|
|
".",
|
|
"..",
|
|
):
|
|
with self.subTest(subfolder=subfolder), self.assertRaises(ValueError):
|
|
m.validate_output_reference(replace(reference, subfolder=subfolder))
|
|
for filename in (r"a\b.png", "a/b.png", "C:foo.png", "a.png:stream", ".", ".."):
|
|
with self.subTest(filename=filename), self.assertRaises(ValueError):
|
|
m.validate_output_reference(replace(reference, filename=filename))
|
|
|
|
def test_root_namespace_is_versioned_style_and_folder_specific(self) -> None:
|
|
m = self.m
|
|
roots = m.snapshot_output_roots("posix", "/a/../雪/root/", "/雪/root")
|
|
self.assertEqual(roots.path_style, "posix")
|
|
self.assertEqual(
|
|
roots.output_key,
|
|
hashlib.sha256('[1,"posix","output","/雪/root"]'.encode()).hexdigest(),
|
|
)
|
|
self.assertEqual(
|
|
roots.temp_key,
|
|
hashlib.sha256('[1,"posix","temp","/雪/root"]'.encode()).hexdigest(),
|
|
)
|
|
self.assertNotEqual(roots.output_key, roots.temp_key)
|
|
self.assertNotEqual(
|
|
roots.output_key,
|
|
m.snapshot_output_roots("posix", "/雪/Root", None).output_key,
|
|
)
|
|
windows = m.snapshot_output_roots(
|
|
"windows", r"C:\a\..\雪\root", r"\\server\share\a\.."
|
|
)
|
|
self.assertEqual(
|
|
windows.output_key,
|
|
hashlib.sha256(
|
|
r'[1,"windows","output","C:\\雪\\root"]'.encode()
|
|
).hexdigest(),
|
|
)
|
|
self.assertIsNotNone(windows.temp_key)
|
|
invalid = (
|
|
None,
|
|
True,
|
|
b"/tmp",
|
|
"relative",
|
|
"",
|
|
"/a\x00b",
|
|
"/\ud800",
|
|
"/" + "x" * 8192,
|
|
"/" + "é" * 4096,
|
|
)
|
|
for root in invalid:
|
|
with self.subTest(root_type=type(root).__name__):
|
|
snapshot = m.snapshot_output_roots("posix", root, "/tmp")
|
|
self.assertIsNone(snapshot.output_key)
|
|
self.assertIsNotNone(snapshot.temp_key)
|
|
self.assertIsNotNone(
|
|
m.snapshot_output_roots("posix", "/" + "x" * 8191, None).output_key
|
|
)
|
|
self.assertIsNotNone(
|
|
m.snapshot_output_roots("posix", "/" + "é" * 4095 + "x", None).output_key
|
|
)
|
|
for root in (
|
|
r"\\server",
|
|
"\\\\server\\",
|
|
r"\\server\.",
|
|
r"1:\root",
|
|
r"C:relative",
|
|
r"\rooted",
|
|
"/rooted",
|
|
r"\\?\C:\root",
|
|
r"\\.\C:\root",
|
|
):
|
|
with self.subTest(root=root):
|
|
self.assertIsNone(
|
|
m.snapshot_output_roots("windows", root, None).output_key
|
|
)
|
|
for style in ("linux", "POSIX", None, True):
|
|
with self.assertRaises(ValueError):
|
|
m.snapshot_output_roots(style, "/tmp", "/tmp")
|
|
|
|
def test_bounds_and_state_row_consistency(self) -> None:
|
|
m = self.m
|
|
a = self.reference()
|
|
valid = {
|
|
"complete": ("reported", "none_reported"),
|
|
"partial": (
|
|
"invalid_evidence",
|
|
"unsupported",
|
|
"limit_reached",
|
|
"root_unavailable",
|
|
"budget_exhausted",
|
|
),
|
|
"unavailable": ("unavailable", "invalid_evidence", "budget_exhausted"),
|
|
"conflicting": ("conflicting",),
|
|
}
|
|
reasons = (
|
|
"reported",
|
|
"none_reported",
|
|
"invalid_evidence",
|
|
"unsupported",
|
|
"limit_reached",
|
|
"root_unavailable",
|
|
"budget_exhausted",
|
|
"unavailable",
|
|
"conflicting",
|
|
"unknown",
|
|
"",
|
|
)
|
|
for state in (*valid, "not_captured", "unknown"):
|
|
for reason in reasons:
|
|
for references in ((), (a,)):
|
|
accepted = reason in valid.get(state, ())
|
|
if state == "complete":
|
|
accepted = accepted and bool(references) == (
|
|
reason == "reported"
|
|
)
|
|
if state == "unavailable":
|
|
accepted = accepted and not references
|
|
if reason == "root_unavailable":
|
|
accepted = accepted and bool(references)
|
|
bundle = m.OutputBundle(references, state, reason)
|
|
with self.subTest(state=state, reason=reason, rows=len(references)):
|
|
if accepted:
|
|
self.assertIsNone(m.validate_output_bundle(bundle))
|
|
else:
|
|
with self.assertRaises(ValueError):
|
|
m.validate_output_bundle(bundle)
|
|
null_root = replace(a, root_key=None)
|
|
for state, reason in (
|
|
("partial", "root_unavailable"),
|
|
("conflicting", "conflicting"),
|
|
):
|
|
m.validate_output_bundle(m.OutputBundle((null_root,), state, reason))
|
|
with self.assertRaises(ValueError):
|
|
m.validate_output_bundle(
|
|
m.OutputBundle((null_root,), "complete", "reported")
|
|
)
|
|
for ordinal in (0, 255):
|
|
m.validate_output_reference(replace(a, image_ordinal=ordinal))
|
|
for changes in (
|
|
{"image_ordinal": -1},
|
|
{"image_ordinal": 256},
|
|
{"image_ordinal": True},
|
|
{"image_ordinal": 1.0},
|
|
{"output_node_id": ""},
|
|
{"output_node_id": "x" * 129},
|
|
{"output_node_id": "\ud800"},
|
|
{"output_node_id": "a\x00b"},
|
|
{"filename": "é" * 513},
|
|
{"subfolder": "雪" * 683},
|
|
{"root_key": "A" * 64},
|
|
{"root_key": "a" * 63},
|
|
{"root_key": "z" * 64},
|
|
{"folder_type": "input"},
|
|
{"path_style": "native"},
|
|
):
|
|
with self.subTest(changes=changes), self.assertRaises(ValueError):
|
|
m.validate_output_reference(replace(a, **changes))
|
|
ordered = (
|
|
replace(a, output_node_id="10"),
|
|
replace(a, output_node_id="2"),
|
|
a,
|
|
replace(a, image_ordinal=1),
|
|
)
|
|
m.validate_output_bundle(m.OutputBundle(ordered, "complete", "reported"))
|
|
for references in (
|
|
(a, a),
|
|
tuple(reversed(ordered)),
|
|
tuple(replace(a, output_node_id=str(i).zfill(3)) for i in range(257)),
|
|
):
|
|
with self.assertRaises(ValueError):
|
|
m.validate_output_bundle(
|
|
m.OutputBundle(references, "complete", "reported")
|
|
)
|
|
maximum = replace(
|
|
a,
|
|
output_node_id="😀" * 128,
|
|
filename="x" * 1024,
|
|
subfolder="x" * 2048,
|
|
path_style="windows",
|
|
)
|
|
self.assertIsNone(m.validate_output_reference(maximum))
|
|
self.assertEqual(
|
|
m.output_payload_size(m.OutputBundle((maximum,), "complete", "reported")),
|
|
3917,
|
|
)
|
|
prefix = tuple(replace(maximum, image_ordinal=i) for i in range(34))
|
|
tail = replace(maximum, image_ordinal=34, filename="x", subfolder="x" * 1400)
|
|
exact = m.OutputBundle((*prefix, tail), "complete", "reported")
|
|
self.assertEqual(m.output_payload_size(exact), 131072)
|
|
with self.assertRaises(ValueError):
|
|
m.output_payload_size(
|
|
replace(
|
|
exact, references=(*prefix, replace(tail, subfolder="x" * 1401))
|
|
)
|
|
)
|
|
for changes in (
|
|
{"format_version": True},
|
|
{"format_version": 2},
|
|
{"source": "host_cache"},
|
|
{"references": [a]},
|
|
):
|
|
with self.subTest(changes=changes), self.assertRaises(ValueError):
|
|
m.validate_output_bundle(
|
|
replace(m.OutputBundle((a,), "complete", "reported"), **changes)
|
|
)
|
|
|
|
def test_no_coercion_or_filesystem_calls(self) -> None:
|
|
m = self.m
|
|
|
|
class Hostile:
|
|
def __str__(self):
|
|
raise AssertionError("coerced arbitrary object")
|
|
|
|
def __fspath__(self):
|
|
raise AssertionError("read arbitrary path protocol")
|
|
|
|
def __iter__(self):
|
|
raise AssertionError("iterated arbitrary object")
|
|
|
|
class Text(str):
|
|
pass
|
|
|
|
class Sequence(tuple):
|
|
pass
|
|
|
|
hostile = Hostile()
|
|
for field in (
|
|
"output_node_id",
|
|
"folder_type",
|
|
"subfolder",
|
|
"filename",
|
|
"path_style",
|
|
"root_key",
|
|
):
|
|
for value in (hostile, Text("output")):
|
|
with self.subTest(field=field), self.assertRaises(ValueError):
|
|
m.validate_output_reference(self.reference(**{field: value}))
|
|
for value in (hostile, Sequence((self.reference(),))):
|
|
with self.assertRaises(ValueError):
|
|
m.validate_output_bundle(m.OutputBundle(value, "complete", "reported"))
|
|
for field in ("state", "reason", "source"):
|
|
with self.subTest(field=field), self.assertRaises(ValueError):
|
|
m.validate_output_bundle(
|
|
replace(
|
|
m.OutputBundle((), "complete", "none_reported"),
|
|
**{field: hostile},
|
|
)
|
|
)
|
|
for value in (hostile, Text("/tmp")):
|
|
self.assertIsNone(m.snapshot_output_roots("posix", value, None).output_key)
|
|
with (
|
|
patch("builtins.open", side_effect=AssertionError("filesystem open")),
|
|
patch("os.stat", side_effect=AssertionError("filesystem stat")),
|
|
patch("os.lstat", side_effect=AssertionError("filesystem lstat")),
|
|
patch(
|
|
"os.path.realpath", side_effect=AssertionError("filesystem resolution")
|
|
),
|
|
):
|
|
roots = m.snapshot_output_roots(
|
|
"posix", "/nonexistent/output", "/nonexistent/temp"
|
|
)
|
|
reference = self.reference(root_key=roots.output_key)
|
|
self.assertIsNone(
|
|
m.validate_output_bundle(
|
|
m.OutputBundle((reference,), "complete", "reported")
|
|
)
|
|
)
|
|
self.assertEqual(
|
|
m.output_payload_size(m.OutputBundle((), "unavailable", "unavailable")),
|
|
128,
|
|
)
|