refactor(storage): gate clone() behind an opt-in allowlist; expand duplicate tests
Follow-up polish from the duplicate-subgraph review: - BaseSqliteStore.clone() now requires an opt-in `_cloneable` flag (default off), so a new or secret-bearing store can never be cloned by accident — defence-in-depth on top of the endpoint's `_DUPLICABLE_KINDS` restriction. Only value-source and colour-strip stores are flagged. - Fix `_check_name_unique` annotation (`str | None = None`). - Add tests: `layers[].brightness_source_id` remap, the warnings safety net firing, the clone allowlist invariant, and clone() refusing a non-cloneable store.
This commit is contained in:
@@ -29,6 +29,10 @@ class BaseSqliteStore(Generic[T]):
|
|||||||
|
|
||||||
_table_name: str
|
_table_name: str
|
||||||
_entity_name: str
|
_entity_name: str
|
||||||
|
# Opt-in allowlist for clone(): defaults off so a new (possibly
|
||||||
|
# secret-bearing) store is never cloneable by accident. Subclasses that hold
|
||||||
|
# no inline secrets and are safe to duplicate set this True.
|
||||||
|
_cloneable: bool = False
|
||||||
|
|
||||||
def __init__(self, db: Database, deserializer: Callable[[dict], T]):
|
def __init__(self, db: Database, deserializer: Callable[[dict], T]):
|
||||||
self._db = db
|
self._db = db
|
||||||
@@ -153,6 +157,8 @@ class BaseSqliteStore(Generic[T]):
|
|||||||
model might hold. Callers must restrict cloning to non-secret-bearing
|
model might hold. Callers must restrict cloning to non-secret-bearing
|
||||||
kinds (see ``_DUPLICABLE_KINDS`` in ``api/routes/graph.py``).
|
kinds (see ``_DUPLICABLE_KINDS`` in ``api/routes/graph.py``).
|
||||||
"""
|
"""
|
||||||
|
if not self._cloneable:
|
||||||
|
raise NotImplementedError(f"{self._entity_name} store does not support cloning")
|
||||||
with self._lock:
|
with self._lock:
|
||||||
if item_id not in self._items:
|
if item_id not in self._items:
|
||||||
from ledgrab.storage.base_store import EntityNotFoundError
|
from ledgrab.storage.base_store import EntityNotFoundError
|
||||||
@@ -180,7 +186,7 @@ class BaseSqliteStore(Generic[T]):
|
|||||||
|
|
||||||
# -- Helpers -------------------------------------------------------------
|
# -- Helpers -------------------------------------------------------------
|
||||||
|
|
||||||
def _check_name_unique(self, name: str, exclude_id: str = None) -> None:
|
def _check_name_unique(self, name: str, exclude_id: str | None = None) -> None:
|
||||||
"""Raise ValueError if *name* is empty or already taken.
|
"""Raise ValueError if *name* is empty or already taken.
|
||||||
|
|
||||||
Must be called while holding ``self._lock``.
|
Must be called while holding ``self._lock``.
|
||||||
|
|||||||
@@ -33,6 +33,7 @@ class ColorStripStore(BaseSqliteStore[ColorStripSource]):
|
|||||||
|
|
||||||
_table_name = "color_strip_sources"
|
_table_name = "color_strip_sources"
|
||||||
_entity_name = "Color strip source"
|
_entity_name = "Color strip source"
|
||||||
|
_cloneable = True # no inline secrets — only references shared entities by id
|
||||||
|
|
||||||
def __init__(self, db: Database):
|
def __init__(self, db: Database):
|
||||||
super().__init__(db, ColorStripSource.from_dict)
|
super().__init__(db, ColorStripSource.from_dict)
|
||||||
|
|||||||
@@ -27,6 +27,7 @@ class ValueSourceStore(BaseSqliteStore[ValueSource]):
|
|||||||
|
|
||||||
_table_name = "value_sources"
|
_table_name = "value_sources"
|
||||||
_entity_name = "Value source"
|
_entity_name = "Value source"
|
||||||
|
_cloneable = True # no inline secrets — only references shared entities by id
|
||||||
|
|
||||||
def __init__(self, db: Database):
|
def __init__(self, db: Database):
|
||||||
super().__init__(db, ValueSource.from_dict)
|
super().__init__(db, ValueSource.from_dict)
|
||||||
|
|||||||
@@ -94,6 +94,35 @@ def test_duplicate_remaps_bindable_slot_to_cloned_value_source(stores):
|
|||||||
assert c_new["intensity"]["source_id"] == res["id_map"][vs.id]
|
assert c_new["intensity"]["source_id"] == res["id_map"][vs.id]
|
||||||
|
|
||||||
|
|
||||||
|
def test_duplicate_remaps_layer_brightness_source(stores):
|
||||||
|
"""A composite layer's value-source brightness binding (list + value ref) is
|
||||||
|
remapped when that value source is also in the selection."""
|
||||||
|
css, vss = stores
|
||||||
|
vs = vss.create_source(name="Dim", source_type="static", value=0.3)
|
||||||
|
leaf = css.create_source(name="Leaf", source_type="single_color", colors=[[9, 9, 9]])
|
||||||
|
comp = css.create_source(
|
||||||
|
name="Comp",
|
||||||
|
source_type="composite",
|
||||||
|
layers=[{**_layer(leaf.id), "brightness_source_id": vs.id}],
|
||||||
|
)
|
||||||
|
res = _duplicate_subgraph([comp.id, leaf.id, vs.id], " (copy)")
|
||||||
|
layer = serialize_entity(css.get(res["id_map"][comp.id]))["layers"][0]
|
||||||
|
assert layer["source_id"] == res["id_map"][leaf.id]
|
||||||
|
assert layer["brightness_source_id"] == res["id_map"][vs.id]
|
||||||
|
|
||||||
|
|
||||||
|
def test_duplicate_safety_net_flags_unremapped_ref(stores, monkeypatch):
|
||||||
|
"""If a reference somehow isn't remapped, the post-clone safety net reports
|
||||||
|
it in `warnings` rather than silently leaving two pipelines sharing a node."""
|
||||||
|
css, _vss = stores
|
||||||
|
b = css.create_source(name="B", source_type="single_color", colors=[[1, 2, 3]])
|
||||||
|
a = css.create_source(name="A", source_type="composite", layers=[_layer(b.id)])
|
||||||
|
# Force remap to a no-op so the clone keeps pointing at the original in-set id.
|
||||||
|
monkeypatch.setattr("ledgrab.api.routes.graph.remap_refs", lambda *a, **k: 0)
|
||||||
|
res = _duplicate_subgraph([a.id, b.id], " (copy)")
|
||||||
|
assert any(w["id"] == res["id_map"][a.id] for w in res["warnings"])
|
||||||
|
|
||||||
|
|
||||||
def test_duplicate_value_source(stores):
|
def test_duplicate_value_source(stores):
|
||||||
_css, vss = stores
|
_css, vss = stores
|
||||||
v = vss.create_source(name="V", source_type="static", value=0.5)
|
v = vss.create_source(name="V", source_type="static", value=0.5)
|
||||||
@@ -120,3 +149,23 @@ def test_duplicate_name_collision_is_suffixed(stores):
|
|||||||
res = _duplicate_subgraph([v.id], " (copy)")
|
res = _duplicate_subgraph([v.id], " (copy)")
|
||||||
new = vss.get(res["id_map"][v.id])
|
new = vss.get(res["id_map"][v.id])
|
||||||
assert new.name == "V (copy) 2"
|
assert new.name == "V (copy) 2"
|
||||||
|
|
||||||
|
|
||||||
|
def test_clone_allowlist_invariant():
|
||||||
|
"""Only explicitly-flagged (secret-free) stores are cloneable; the base
|
||||||
|
default is off so a new store is never cloneable by accident."""
|
||||||
|
from ledgrab.storage.base_sqlite_store import BaseSqliteStore
|
||||||
|
|
||||||
|
assert BaseSqliteStore._cloneable is False
|
||||||
|
assert ColorStripStore._cloneable is True
|
||||||
|
assert ValueSourceStore._cloneable is True
|
||||||
|
|
||||||
|
|
||||||
|
def test_clone_refuses_non_cloneable_store(stores, monkeypatch):
|
||||||
|
"""clone() refuses stores not on the allowlist (defence-in-depth even though
|
||||||
|
the duplicate endpoint already restricts to the safe kinds)."""
|
||||||
|
css, _vss = stores
|
||||||
|
src = css.create_source(name="X", source_type="single_color", colors=[[1, 1, 1]])
|
||||||
|
monkeypatch.setattr(type(css), "_cloneable", False)
|
||||||
|
with pytest.raises(NotImplementedError):
|
||||||
|
css.clone(src.id, "X (copy)")
|
||||||
|
|||||||
Reference in New Issue
Block a user