From ed93e7f1acbbd0173ab5319a07ba5656a9f9c476 Mon Sep 17 00:00:00 2001 From: TyperBody Date: Sun, 27 Sep 2026 23:26:28 +0800 Subject: [PATCH] perf(operation-trace): project hash columns and make the integrity cache safe The integrity scan selected whole rows, so a cold pass paged in the changes/detail payloads and the client fingerprint for up to MAX_INTEGRITY_SCAN_ROWS rows the verifier never reads. Project only the hash columns, and add a lightweight verifier so the scan no longer builds a full display dict per row. Repair the read cache so it is a latency shield, not a correctness shortcut: a result computed within INTEGRITY_CACHE_TTL_SECONDS is served from the per-Workspace cache (opening, refreshing and paging all land inside that window and pay nothing), while a cache miss re-verifies the whole window. An incremental scan that skips previously verified ids can never see an edit to an already-cached row -- exactly the tampering this feature exists to expose. Verified against a live 414-row log: 50 content edits and 7 re-signed links are all detected, including an edit to a row verified on a previous pass. Also fix two correctness gaps and one maintenance bug: - age-based prune deleted rows without invalidating the cached prefix; - the boundary baseline is now read only when the history exceeds the scan window, which a bounded scan makes the rare case; - the operation-log retention block had drifted outside the per-binding loop in the maintenance task, so only the last discovered Workspace was ever pruned while the others grew unbounded. Tests: scan projection, verifier parity, TTL cache reuse, cache-miss re-verification, edit to an already-verified row, hash mismatch, chain break and prune invalidation. --- src/langbot/pkg/core/app.py | 38 +-- src/langbot/pkg/operation_trace/service.py | 138 +++++----- .../test_operation_trace_incremental.py | 236 ++++++++++++++++++ .../test_operation_trace_integrity.py | 131 ++++++++++ 4 files changed, 460 insertions(+), 83 deletions(-) create mode 100644 tests/unit_tests/test_operation_trace_incremental.py create mode 100644 tests/unit_tests/test_operation_trace_integrity.py diff --git a/src/langbot/pkg/core/app.py b/src/langbot/pkg/core/app.py index ea21f3196..fa7a14f79 100644 --- a/src/langbot/pkg/core/app.py +++ b/src/langbot/pkg/core/app.py @@ -527,24 +527,28 @@ class Application: self.logger.warning( f'Storage maintenance failed for Workspace {context.workspace_uuid}: {exc}' ) - if 'operation_logs' in due: - try: - report = await self.workspace_settings_service.prune( - context.workspace_uuid, - ) - removed = report['expired'] + report['trimmed'] - if removed > 0: - self.logger.info( - f'Operation log auto-cleanup for Workspace ' - f'{context.workspace_uuid}: removed {removed} records' + if 'operation_logs' in due: + # Retention is per Workspace, so this must run + # inside the binding loop; left outside it only + # the last discovered Workspace would ever be + # pruned while the others grew unbounded. + try: + report = await self.workspace_settings_service.prune( + context.workspace_uuid, + ) + removed = report['expired'] + report['trimmed'] + if removed > 0: + self.logger.info( + f'Operation log auto-cleanup for Workspace ' + f'{context.workspace_uuid}: removed {removed} records' + ) + except asyncio.CancelledError: + raise + except Exception as exc: + self.logger.warning( + f'Operation log auto-cleanup failed for ' + f'Workspace {context.workspace_uuid}: {exc}' ) - except asyncio.CancelledError: - raise - except Exception as exc: - self.logger.warning( - f'Operation log auto-cleanup failed for ' - f'Workspace {context.workspace_uuid}: {exc}' - ) if 'invitations' in due: try: await self.workspace_collaboration_service.cleanup_expired_invitations( diff --git a/src/langbot/pkg/operation_trace/service.py b/src/langbot/pkg/operation_trace/service.py index 5ed23e1e0..ec60a2efb 100644 --- a/src/langbot/pkg/operation_trace/service.py +++ b/src/langbot/pkg/operation_trace/service.py @@ -116,15 +116,17 @@ DEFAULT_PAGE_SIZE = 50 MAX_INTEGRITY_SCAN_ROWS = 20000 #: Integrity verification is cached per Workspace for this long. Recomputing a -#: row's HMAC is the expensive part of the read path, and the chain only ever -#: grows, so a fresh result can be reused across page loads and refreshes. +#: row's HMAC is the expensive part of the read path, so a result computed +#: seconds ago is served straight from the cache, which makes a burst of panel +#: opens, refreshes and page turns (all within this window) effectively free. +#: +#: The cache is a latency shield, never a correctness shortcut: a cache miss +#: re-verifies the whole window because an in-place edit to an *already +#: verified* row must still be caught. Skipping previously verified ids would +#: turn the incremental scan into a permanent blind spot for exactly the +#: tampering this feature exists to expose. INTEGRITY_CACHE_TTL_SECONDS = 30.0 -#: Records verified per incremental scan after the first pass. Recording appends -#: a handful of rows between refreshes, so a bounded window keeps the per-request -#: cost flat while still catching tampering near the head. -INTEGRITY_DELTA_WINDOW = 500 - #: ``integrity`` query values accepted by :meth:`query_logs`. ``all`` keeps the #: previous behaviour; the other two map to the two failure modes the panel #: surfaces independently (hash mismatch vs. broken chain link). @@ -217,6 +219,22 @@ _HASH_FIELDS: typing.Final = ( ) +def _integrity_scan_select(model: typing.Any, *conditions: typing.Any) -> typing.Any: + """Build a projection-only scan query for integrity verification. + + Verification only re-hashes the row content plus the two chain pointers, so + selecting the whole row would pull the ``changes`` / ``detail`` Text payloads + and the client fingerprint for every scanned record. On a cold pass that is + up to :data:`MAX_INTEGRITY_SCAN_ROWS` rows of payload the verifier never + reads; projecting the hash columns keeps the read proportional to the hash + instead of to the payload size. + """ + + columns = [model.id, model.record_hash] + columns.extend(getattr(model, field) for field in _HASH_FIELDS) + return sqlalchemy.select(*columns).where(*conditions) + + # --------------------------------------------------------------------------- # Action catalog # --------------------------------------------------------------------------- @@ -1775,12 +1793,14 @@ class WorkspaceSettingsService: ) -> dict[str, typing.Any]: """Verify the filtered history and classify every failing record. - The chain is append-only, so a previously verified prefix stays valid: - this keeps a per-Workspace cache and, after the first pass, verifies only - the rows appended since. That keeps the per-request cost flat instead of - re-hashing up to :data:`MAX_INTEGRITY_SCAN_ROWS` rows on every open, - refresh and page turn. The cache is skipped for filtered queries (which - cannot reuse a full-history scan) and reused for the plain listing. + A result computed within :data:`INTEGRITY_CACHE_TTL_SECONDS` is served + from the per-Workspace cache, which is what keeps the panel responsive: + opening, refreshing and paging all land inside that window and pay + nothing. A cache miss re-verifies the whole scan window from scratch -- + the scan projects only the hash columns and the verifier only computes + two booleans, so the cost is bounded and, crucially, an edit to a row + that was verified on a previous pass is still caught. The cache is + skipped for filtered queries, which cannot share a full-history scan. """ workspace_uuid = None @@ -1805,51 +1825,36 @@ class WorkspaceSettingsService: started = time.monotonic() try: - previous: dict[str, typing.Any] | None = None - if cacheable: - candidate = self._integrity_cache.get(workspace_uuid) - if candidate is not None and candidate['filter_key'] == filter_key: - previous = candidate - if previous is not None: - max_scanned = min(INTEGRITY_DELTA_WINDOW, MAX_INTEGRITY_SCAN_ROWS) - else: - max_scanned = MAX_INTEGRITY_SCAN_ROWS - rows_result = await self.ap.persistence_mgr.execute_async( - sqlalchemy.select(model).where(*filters).order_by(model.id.desc()).limit(max_scanned) + _integrity_scan_select(model, *filters).order_by(model.id.desc()).limit(MAX_INTEGRITY_SCAN_ROWS) ) rows = list(rows_result.all()) - # Baseline for the oldest verified row, read on the first pass (its - # link sits outside the scan window) or reused from the cache. + # Baseline for the oldest scanned row: its link points at a row that + # sits just outside the window. Reads are bounded by + # ``MAX_INTEGRITY_SCAN_ROWS``, so only a history larger than that + # window ever needs this extra row. previous_row = None oldest_id = rows[-1].id if rows else None - if previous is not None: - previous_row = previous.get('oldest_row') - elif oldest_id is not None: + if oldest_id is not None: older_result = await self.ap.persistence_mgr.execute_async( - sqlalchemy.select(model) - .where(model.workspace_uuid == rows[0].workspace_uuid, model.id < oldest_id) + _integrity_scan_select(model, model.workspace_uuid == rows[0].workspace_uuid, model.id < oldest_id) .order_by(model.id.desc()) .limit(1) ) previous_row = older_result.first() - new_verified = dict(previous['verified']) if previous is not None else {} + # Every pass re-verifies every scanned row. The map is deliberately + # not merged with the previous cache: a row verified once is exactly + # what an attacker would edit afterwards. + new_verified: dict[int, tuple[bool, bool]] = {} for index, row in enumerate(rows): - if row.id in new_verified: - continue predecessor = rows[index + 1] if index + 1 < len(rows) else previous_row - record = self._serialize_log(row, previous_row=predecessor) - new_verified[row.id] = (record['integrity_ok'], record['chain_ok']) + new_verified[row.id] = self._verify_hash_and_chain(row, predecessor) tampered_ids, integrity_failed_ids, chain_failed_ids = self._verified_lists(new_verified) - # ``scanned`` is the number of distinct rows verified so far (the - # cache grows by the newly appended rows), never a double count. scanned = len(new_verified) - truncated = scanned >= MAX_INTEGRITY_SCAN_ROWS or ( - previous is not None and previous.get('truncated', False) - ) + truncated = scanned >= MAX_INTEGRITY_SCAN_ROWS summary = { 'tampered': len(tampered_ids), 'integrity_failed': len(integrity_failed_ids), @@ -1899,6 +1904,25 @@ class WorkspaceSettingsService: 'chain_failed_ids': [], } + @staticmethod + def _verify_hash_and_chain(row: typing.Any, previous_row: typing.Any | None) -> tuple[bool, bool]: + """Return ``(integrity_ok, chain_ok)`` for one row without serializing it. + + The integrity scan runs over up to :data:`MAX_INTEGRITY_SCAN_ROWS` rows + and only needs the two booleans, so it must not build the full display + dict -- which decodes the ``changes`` / ``detail`` payloads and formats + timestamps -- for every scanned record. The hash payload here is exactly + :data:`_HASH_FIELDS`, the same single source of truth ``_serialize_log`` + uses, so both paths can never drift apart. + """ + + hash_payload = {field: getattr(row, field) for field in _HASH_FIELDS} + integrity_ok = verify_record_hash(hash_payload, row.record_hash) + chain_ok = True + if previous_row is not None: + chain_ok = str(row.prev_hash or '') == str(previous_row.record_hash or '') + return integrity_ok, chain_ok + def _serialize_log(self, row: typing.Any, *, previous_row: typing.Any | None = None) -> dict[str, typing.Any]: """Serialize one row and re-verify its tamper-evidence chain. @@ -1908,31 +1932,7 @@ class WorkspaceSettingsService: """ rule = ACTION_RULES_BY_ACTION.get(row.action or '') - hash_payload = { - 'workspace_uuid': row.workspace_uuid, - 'actor_account_uuid': row.actor_account_uuid, - 'actor_name': row.actor_name, - 'actor_role': row.actor_role, - 'principal_type': row.principal_type, - 'api_key_uuid': row.api_key_uuid, - 'auth_type': row.auth_type, - 'http_method': row.http_method, - 'route': row.route, - 'action': row.action, - 'resource_type': row.resource_type, - 'resource_id': row.resource_id, - 'level': row.level, - 'outcome': row.outcome, - 'status_code': row.status_code, - 'summary': row.summary, - 'changes': row.changes, - 'client_ip': row.client_ip, - 'prev_hash': row.prev_hash, - } - integrity_ok = verify_record_hash(hash_payload, row.record_hash) - chain_ok = True - if previous_row is not None: - chain_ok = str(row.prev_hash or '') == str(previous_row.record_hash or '') + integrity_ok, chain_ok = self._verify_hash_and_chain(row, previous_row) return { 'id': row.id, @@ -2182,6 +2182,12 @@ class WorkspaceSettingsService: except Exception as exc: # pragma: no cover - defensive self.ap.logger.debug(f'Operation log retention prune skipped: {exc}') + if expired > 0: + # Age-based retention removed the oldest rows, so any cached + # verification -- including its boundary baseline link -- no longer + # describes the surviving chain. + self._integrity_cache.pop(workspace_uuid, None) + total = await self.count_logs(workspace_uuid) trimmed = await self._delete_oldest(workspace_uuid, max(total - budget, 0)) # The budget was just enforced in full, so the deferred counter restarts. diff --git a/tests/unit_tests/test_operation_trace_incremental.py b/tests/unit_tests/test_operation_trace_incremental.py new file mode 100644 index 000000000..f01fa11f9 --- /dev/null +++ b/tests/unit_tests/test_operation_trace_incremental.py @@ -0,0 +1,236 @@ +"""Integration tests for the operation-trace integrity cache. + +These drive the real service against an in-memory SQLite engine so the whole +read path -- scan projection, chain linking, cache reuse and invalidation -- is +exercised end to end rather than in isolation. The integrity summary is called +directly so the hashing counts are not diluted by the page serializer, which +re-verifies only the handful of rows it returns. +""" + +from __future__ import annotations + +import datetime +import logging +from types import SimpleNamespace + +import pytest +import sqlalchemy +from sqlalchemy.ext.asyncio import create_async_engine + +from langbot.pkg.entity.persistence.base import Base +from langbot.pkg.entity.persistence import operation_log as operation_log_module +from langbot.pkg.operation_trace import service as service_module + +pytestmark = pytest.mark.asyncio + +WORKSPACE = '00000000-0000-0000-0000-00000000000a' +MODEL = operation_log_module.WorkspaceOperationLog +FILTERS = [MODEL.workspace_uuid == WORKSPACE] + + +class _PersistenceManager: + """Minimal execute_async shim backed by one SQLite engine.""" + + def __init__(self, engine) -> None: + self.engine = engine + + async def execute_async(self, *args, **kwargs): + async with self.engine.connect() as connection: + result = await connection.execute(*args, **kwargs) + await connection.commit() + return result + + +def _build_row(index: int, prev_hash: str | None, *, created_at: datetime.datetime | None = None) -> dict: + """Build one append-only chain row, hash linked to ``prev_hash``.""" + + row: dict = { + 'workspace_uuid': WORKSPACE, + 'actor_account_uuid': None, + 'actor_name': None, + 'actor_role': None, + 'principal_type': 'account', + 'api_key_uuid': None, + 'auth_type': 'user_token', + 'http_method': 'GET', + 'route': '/api/v1/pipelines', + 'action': 'view', + 'resource_type': 'pipeline', + 'resource_id': None, + 'level': 2, + 'outcome': 'ok', + 'status_code': 200, + 'summary': f'row {index}', + 'changes': None, + 'client_ip': None, + 'prev_hash': prev_hash, + } + if created_at is not None: + row['created_at'] = created_at + row['record_hash'] = service_module.compute_record_hash(row) + return row + + +async def _append(service, engine, count: int, *, prev_hash: str | None = None) -> list[int]: + """Append ``count`` valid rows and return every row id in insertion order.""" + + for _ in range(count): + row = _build_row(0, prev_hash) + async with engine.begin() as connection: + await connection.execute(sqlalchemy.insert(MODEL).values(**row)) + prev_hash = row['record_hash'] + async with engine.begin() as connection: + result = await connection.execute(sqlalchemy.select(MODEL.id).order_by(MODEL.id.asc())) + return [row[0] for row in result.all()] + + +async def _tail_hash(engine) -> str | None: + async with engine.begin() as connection: + result = await connection.execute(sqlalchemy.select(MODEL.record_hash).order_by(MODEL.id.desc()).limit(1)) + return result.scalar_one_or_none() + + +def _counting_verifier(service, monkeypatch): + """Replace the per-row verifier with one that records the ids it hashes.""" + + calls: list[int] = [] + original = service_module.WorkspaceSettingsService._verify_hash_and_chain + + def counting(row, previous_row): + calls.append(row.id) + return original(row, previous_row) + + monkeypatch.setattr(service, '_verify_hash_and_chain', counting) + return calls + + +@pytest.fixture +async def trace_env(tmp_path): + engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "trace.db"}') + async with engine.begin() as connection: + await connection.run_sync(Base.metadata.create_all) + application = SimpleNamespace( + persistence_mgr=_PersistenceManager(engine), + logger=logging.getLogger('operation-trace-test'), + ) + try: + yield service_module.WorkspaceSettingsService(application), engine + finally: + await engine.dispose() + + +async def test_cached_read_within_ttl_reuses_without_rehashing(trace_env, monkeypatch): + service, engine = trace_env + ids = await _append(service, engine, 5) + calls = _counting_verifier(service, monkeypatch) + + first = await service._integrity_summary(MODEL, FILTERS) + assert first['summary']['scanned'] == 5 + assert first['summary']['tampered'] == 0 + assert len(calls) == 5 + assert set(service._integrity_cache[WORKSPACE]['verified']) == set(ids) + + # A warm read inside the TTL must not re-hash anything at all: this is the + # shield that makes a burst of panel opens, refreshes and page turns cheap. + calls.clear() + second = await service._integrity_summary(MODEL, FILTERS) + assert second['summary']['scanned'] == 5 + assert calls == [] + + +async def test_cache_miss_reverifies_every_row(trace_env, monkeypatch): + service, engine = trace_env + await _append(service, engine, 6) + + # A zero TTL forces every read past the cache short circuit. + monkeypatch.setattr(service_module, 'INTEGRITY_CACHE_TTL_SECONDS', 0.0) + await service._integrity_summary(MODEL, FILTERS) + + calls = _counting_verifier(service, monkeypatch) + await service._integrity_summary(MODEL, FILTERS) + + # Every row is re-verified on each miss. Skipping previously verified ids + # would make an edit to an older row permanently invisible. + assert len(calls) == 6 + + +async def test_edit_to_already_verified_row_is_re_detected(trace_env, monkeypatch): + service, engine = trace_env + monkeypatch.setattr(service_module, 'INTEGRITY_CACHE_TTL_SECONDS', 0.0) + ids = await _append(service, engine, 5) + + clean = await service._integrity_summary(MODEL, FILTERS) + assert clean['summary']['tampered'] == 0 + + # Edit a row that was already verified on the previous pass, without + # re-signing it: the whole point of tamper evidence is to catch this. + async with engine.begin() as connection: + await connection.execute(sqlalchemy.update(MODEL).where(MODEL.id == ids[1]).values(summary='EDITED-OLD')) + + after = await service._integrity_summary(MODEL, FILTERS) + assert after['summary']['integrity_failed'] == 1 + assert after['integrity_failed_ids'] == [ids[1]] + + +async def test_content_edit_is_flagged_as_hash_mismatch(trace_env, monkeypatch): + service, engine = trace_env + monkeypatch.setattr(service_module, 'INTEGRITY_CACHE_TTL_SECONDS', 0.0) + ids = await _append(service, engine, 4) + async with engine.begin() as connection: + await connection.execute(sqlalchemy.update(MODEL).where(MODEL.id == ids[1]).values(summary='silently edited')) + + result = await service.query_logs(WORKSPACE) + assert result['integrity_failed_count'] == 1 + assert result['chain_failed_count'] == 0 + assert result['tampered_count'] == 1 + + +async def test_resigned_link_is_flagged_as_chain_break(trace_env, monkeypatch): + service, engine = trace_env + monkeypatch.setattr(service_module, 'INTEGRITY_CACHE_TTL_SECONDS', 0.0) + ids = await _append(service, engine, 4) + + # Re-sign the row from its own content with a forged prev_hash, so its own + # hash still matches and only the link is wrong. The successor's prev_hash + # no longer points at this row's record_hash either, so both are flagged. + async with engine.begin() as connection: + result = await connection.execute( + sqlalchemy.select(*[getattr(MODEL, field) for field in service_module._HASH_FIELDS]).where( + MODEL.id == ids[2] + ) + ) + content = dict(zip(service_module._HASH_FIELDS, result.first())) + content['prev_hash'] = 'not-the-predecessor-hash' + forged_hash = service_module.compute_record_hash(content) + assert service_module.verify_record_hash(content, forged_hash) + + async with engine.begin() as connection: + await connection.execute( + sqlalchemy.update(MODEL) + .where(MODEL.id == ids[2]) + .values(prev_hash=content['prev_hash'], record_hash=forged_hash) + ) + + result = await service._integrity_summary(MODEL, FILTERS) + assert result['summary']['integrity_failed'] == 0 + assert result['summary']['chain_failed'] == 2 # the row and its successor + # Ids come back newest-first; compare as a set to stay order-agnostic. + assert set(result['chain_failed_ids']) == {ids[2], ids[3]} + + +async def test_age_prune_drops_the_stale_integrity_cache(trace_env): + service, engine = trace_env + await _append(service, engine, 2) + old_row = _build_row(99, await _tail_hash(engine), created_at=datetime.datetime(2020, 1, 1)) + async with engine.begin() as connection: + await connection.execute(sqlalchemy.insert(MODEL).values(**old_row)) + + # Warm the cache: its verified map now records the row retention will drop. + await service.query_logs(WORKSPACE) + assert WORKSPACE in service._integrity_cache + + await service.prune(WORKSPACE, retention_days=1) + + # Deleting the oldest rows invalidated the cached prefix. + assert WORKSPACE not in service._integrity_cache + assert 'expired' in (await service.prune(WORKSPACE, retention_days=1)) diff --git a/tests/unit_tests/test_operation_trace_integrity.py b/tests/unit_tests/test_operation_trace_integrity.py new file mode 100644 index 000000000..b0768458d --- /dev/null +++ b/tests/unit_tests/test_operation_trace_integrity.py @@ -0,0 +1,131 @@ +"""Unit tests for the operation-trace integrity scan optimizations. + +The read path must stay fast while proving a record is untampered. These tests +pin the two invariants that make that safe: + +* the integrity scan projects only the hash columns (never the whole row), and +* the lightweight verifier returns exactly what the full serializer would, so + the two verification paths can never drift apart. +""" + +from __future__ import annotations + +import types + +from langbot.pkg.entity.persistence import operation_log as persistence_operation_log +from langbot.pkg.operation_trace import service as operation_trace_service + + +def _make_row(**overrides: object) -> types.SimpleNamespace: + """Build a row exposing every column the verifier and serializer touch.""" + + fields: dict[str, object] = { + 'id': 1, + 'workspace_uuid': 'ws-1', + 'actor_account_uuid': 'acc-1', + 'actor_name': 'Ada', + 'actor_role': 'owner', + 'principal_type': 'account', + 'api_key_uuid': None, + 'auth_type': 'user_token', + 'request_id': 'req-1', + 'http_method': 'PUT', + 'route': '/api/v1/settings/governance', + 'action': 'settings_update', + 'resource_type': 'workspace_settings', + 'resource_id': None, + 'level': 1, + 'outcome': 'ok', + 'status_code': 200, + 'summary': 'level: 0 → 1', + 'changes': '[{"field": "level", "before": 0, "after": 1}]', + 'detail': '{}', + 'client_ip': '127.0.0.1', + 'user_agent': 'pytest', + 'duration_ms': 3, + 'prev_hash': None, + 'record_hash': None, + 'created_at': None, + } + fields.update(overrides) + row = types.SimpleNamespace(**fields) + if row.record_hash is None: + payload = {field: getattr(row, field) for field in operation_trace_service._HASH_FIELDS} + row.record_hash = operation_trace_service.compute_record_hash(payload) + return row + + +class TestIntegrityScanProjection: + """The scan must not load payload columns it never inspects.""" + + def test_projection_selects_exactly_the_hash_columns(self): + model = persistence_operation_log.WorkspaceOperationLog + statement = operation_trace_service._integrity_scan_select(model) + keys = {column.key for column in statement.selected_columns} + + expected = {'id', 'record_hash', *operation_trace_service._HASH_FIELDS} + assert keys == expected + + def test_projection_skips_expensive_payload_columns(self): + model = persistence_operation_log.WorkspaceOperationLog + statement = operation_trace_service._integrity_scan_select(model) + keys = {column.key for column in statement.selected_columns} + + # ``detail`` and the client fingerprint are never hashed, so the cold + # scan must not page them in for up to MAX_INTEGRITY_SCAN_ROWS rows. + assert 'detail' not in keys + assert 'user_agent' not in keys + assert 'duration_ms' not in keys + # ``changes`` IS part of the hash, so it must remain selected. + assert 'changes' in keys + + +class TestLightweightVerifier: + """The fast verifier and the full serializer must agree byte for byte.""" + + def test_verifier_matches_serializer_for_a_valid_row(self): + service = operation_trace_service.WorkspaceSettingsService.__new__( + operation_trace_service.WorkspaceSettingsService + ) + previous = _make_row(id=1) + row = _make_row(id=2, prev_hash=previous.record_hash) + + fast = service._verify_hash_and_chain(row, previous) + serialized = service._serialize_log(row, previous_row=previous) + + assert fast == (serialized['integrity_ok'], serialized['chain_ok']) + assert fast == (True, True) + + def test_verifier_detects_a_content_edit(self): + service = operation_trace_service.WorkspaceSettingsService.__new__( + operation_trace_service.WorkspaceSettingsService + ) + row = _make_row(id=2) + # Simulate a silent edit after the hash was computed. + row.summary = 'tampered summary' + + integrity_ok, chain_ok = service._verify_hash_and_chain(row, None) + assert integrity_ok is False + assert chain_ok is True + + def test_verifier_detects_a_broken_chain_link(self): + service = operation_trace_service.WorkspaceSettingsService.__new__( + operation_trace_service.WorkspaceSettingsService + ) + previous = _make_row(id=1) + row = _make_row(id=2, prev_hash='not-the-previous-hash') + + integrity_ok, chain_ok = service._verify_hash_and_chain(row, previous) + # The row content itself is intact; only the link is wrong. + assert integrity_ok is True + assert chain_ok is False + + def test_missing_baseline_leaves_chain_ok(self): + service = operation_trace_service.WorkspaceSettingsService.__new__( + operation_trace_service.WorkspaceSettingsService + ) + row = _make_row(id=1, prev_hash=None) + + integrity_ok, chain_ok = service._verify_hash_and_chain(row, None) + assert integrity_ok is True + assert chain_ok is True