From 0cc8c3dcfd12c2fb0494a86d67af37d8204ce7c9 Mon Sep 17 00:00:00 2001 From: Chin Rock Date: Mon, 2 Nov 2026 04:19:48 +0000 Subject: [PATCH] fix(audit): verify filtered logs against workspace predecessors Co-authored-by: multica-agent --- src/langbot/pkg/operation_trace/service.py | 74 ++++++------- .../test_operation_trace_incremental.py | 100 ++++++++++++++++++ 2 files changed, 137 insertions(+), 37 deletions(-) diff --git a/src/langbot/pkg/operation_trace/service.py b/src/langbot/pkg/operation_trace/service.py index dbd281e8f..e5431a503 100644 --- a/src/langbot/pkg/operation_trace/service.py +++ b/src/langbot/pkg/operation_trace/service.py @@ -1219,7 +1219,7 @@ class WorkspaceSettingsService: # later read only has to verify the rows appended since. Without this the # panel re-hashed up to MAX_INTEGRITY_SCAN_ROWS rows on every open, # refresh and page turn, which is what made reading the log feel slow. - self._integrity_cache: dict[str, dict[str, typing.Any]] = {} + self._integrity_cache: dict[tuple, dict[str, typing.Any]] = {} self._dropped_count = 0 # Failed persistence attempts, so a silently broken writer (a missing # tenant scope, an RLS rejection) is reported instead of only counted. @@ -2040,25 +2040,13 @@ class WorkspaceSettingsService: ) rows = list(rows_result.all()) - # Fetch one extra older row so the oldest visible record can still - # have its chain link verified instead of being reported as broken. - chain_rows: list[typing.Any] = list(rows) - if rows: - oldest_id = rows[-1].id - older_result = await self.ap.persistence_mgr.execute_async( - sqlalchemy.select(model) - .where(model.workspace_uuid == workspace_uuid, model.id < oldest_id) - .order_by(model.id.desc()) - .limit(1) - ) - chain_rows = list(rows) + list(older_result.all()) - + predecessors = await self._log_predecessors(model, rows) records = [ self._serialize_log( row, - previous_row=chain_rows[index + 1] if index + 1 < len(chain_rows) else None, + previous_row=predecessors.get(row.id), ) - for index, row in enumerate(rows) + for row in rows ] integrity_summary = summary['summary'] except Exception as exc: # pragma: no cover - defensive @@ -2094,6 +2082,27 @@ class WorkspaceSettingsService: 'truncated': False, } + async def _log_predecessors(self, model: typing.Any, rows: list[typing.Any]) -> dict[int, typing.Any]: + """Fetch each row's immediate predecessor in its unfiltered Workspace history.""" + + if not rows: + return {} + previous = model.__table__.alias('previous_log') + previous_id = ( + sqlalchemy.select(previous.c.id) + .where(previous.c.workspace_uuid == model.workspace_uuid, previous.c.id < model.id) + .order_by(previous.c.id.desc()) + .limit(1) + .correlate(model) + .scalar_subquery() + ) + result = await self.ap.persistence_mgr.execute_async( + sqlalchemy.select(model.id, previous.c.record_hash) + .join(previous, previous.c.id == previous_id) + .where(model.id.in_([row.id for row in rows])) + ) + return {row.id: row for row in result.all()} + def _invalidate_integrity_cache(self, workspace_uuid: str) -> None: """Drop every cached verification for one Workspace. @@ -2144,13 +2153,14 @@ class WorkspaceSettingsService: A result computed within :data:`INTEGRITY_CACHE_TTL_SECONDS` is served from the cache, which is what keeps the panel responsive: opening, refreshing and paging all land inside that window and pay nothing. The - cache key is ``(Workspace, listing filters)`` -- an integrity drill-down - reuses the very result it is drilling into, and both "everything" and + cache key includes the Workspace, listing filters and bound values. An + integrity drill-down reuses its source result, and both "everything" and "mutations only" are cached independently -- so a cache miss is rare. A miss re-verifies the *whole* window from scratch: the scan projects only the hash columns and the verifier only computes two booleans, so the cost is bounded, and an edit to a row verified on a previous pass is - still caught. + still caught. Each link is checked against its immediate predecessor in + the unfiltered Workspace history, independently of the scan window. """ workspace_uuid = None @@ -2160,7 +2170,11 @@ class WorkspaceSettingsService: except AttributeError: continue break - cache_key = (workspace_uuid, tuple(str(condition) for condition in filters)) + compiled_filters = [condition.compile() for condition in filters] + cache_key = ( + workspace_uuid, + tuple((str(condition), tuple(sorted(condition.params.items()))) for condition in compiled_filters), + ) cacheable = workspace_uuid is not None now = time.monotonic() @@ -2176,27 +2190,13 @@ class WorkspaceSettingsService: ) rows = list(rows_result.all()) - # 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 oldest_id is not None: - older_result = await self.ap.persistence_mgr.execute_async( - _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() - + predecessors = await self._log_predecessors(model, rows) # 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): - predecessor = rows[index + 1] if index + 1 < len(rows) else previous_row - new_verified[row.id] = self._verify_hash_and_chain(row, predecessor) + for row in rows: + new_verified[row.id] = self._verify_hash_and_chain(row, predecessors.get(row.id)) tampered_ids, integrity_failed_ids, chain_failed_ids = self._verified_lists(new_verified) scanned = len(new_verified) diff --git a/tests/unit_tests/test_operation_trace_incremental.py b/tests/unit_tests/test_operation_trace_incremental.py index 2f2f29bdd..0374ef566 100644 --- a/tests/unit_tests/test_operation_trace_incremental.py +++ b/tests/unit_tests/test_operation_trace_incremental.py @@ -275,3 +275,103 @@ async def test_invalidation_clears_every_cached_view(trace_env): # A deletion invalidates *all* views for the Workspace, not just one. assert _cached_views(service) == [] + + +@pytest.fixture +async def mixed_history(trace_env): + service, engine = trace_env + previous_hash = None + ids = [] + for index, resource_type in enumerate(['bot', 'pipeline', 'bot', 'model', 'bot', 'pipeline']): + row = _build_row(index, previous_hash, created_at=datetime.datetime(2026, 1, index + 1)) + row.update( + resource_type=resource_type, + action='update' if resource_type == 'bot' else 'view', + level=1 if resource_type == 'bot' else 2, + actor_account_uuid='bot-operator' if resource_type == 'bot' else 'reader', + ) + row['record_hash'] = service_module.compute_record_hash(row) + previous_hash = row['record_hash'] + foreign_row = _build_row(index, None) + foreign_row['workspace_uuid'] = 'other-workspace' + foreign_row['record_hash'] = service_module.compute_record_hash(foreign_row) + async with engine.begin() as connection: + result = await connection.execute(sqlalchemy.insert(MODEL).values(**row)) + ids.append(result.inserted_primary_key[0]) + await connection.execute(sqlalchemy.insert(MODEL).values(**foreign_row)) + return service, engine, ids + + +@pytest.mark.parametrize( + 'filters, expected_count', + [ + ({}, 6), + ({'resource_type': 'bot'}, 3), + ({'action': 'update'}, 3), + ({'level': 1}, 3), + ({'actor_account_uuid': 'bot-operator'}, 3), + ({'search': 'bot'}, 3), + ({'since': datetime.datetime(2026, 1, 2), 'until': datetime.datetime(2026, 1, 5)}, 4), + ({'resource_type': 'missing'}, 0), + ], +) +async def test_filtered_history_uses_workspace_predecessors(mixed_history, filters, expected_count): + service, _engine, _ids = mixed_history + for offset in (0, 1): + result = await service.query_logs(WORKSPACE, limit=2, offset=offset, **filters) + assert result['total'] == expected_count + assert result['scanned_count'] == expected_count + assert result['tampered_count'] == result['chain_failed_count'] == result['integrity_failed_count'] == 0 + assert all(row['chain_ok'] and row['integrity_ok'] for row in result['records']) + + +@pytest.mark.parametrize('field', ['summary', 'prev_hash']) +async def test_filtered_history_detects_tampering(mixed_history, monkeypatch, field): + service, engine, ids = mixed_history + monkeypatch.setattr(service_module, 'INTEGRITY_CACHE_TTL_SECONDS', 0.0) + await service.query_logs(WORKSPACE, resource_type='bot') + async with engine.begin() as connection: + await connection.execute(sqlalchemy.update(MODEL).where(MODEL.id == ids[2]).values(**{field: 'tampered'})) + for filters in ({}, {'resource_type': 'bot'}, {'resource_type': 'bot', 'integrity': 'issues'}): + result = await service.query_logs(WORKSPACE, **filters) + assert result['tampered_count'] == result['integrity_failed_count'] == 1 + assert result['chain_failed_count'] == (1 if field == 'prev_hash' else 0) + damaged = next(row for row in result['records'] if row['id'] == ids[2]) + assert damaged['integrity_ok'] is False + assert damaged['chain_ok'] is (field != 'prev_hash') + assert all(row['chain_ok'] for row in result['records'] if row['id'] != ids[2]) + + +async def test_filter_values_have_distinct_cache_entries(mixed_history): + service, engine, ids = mixed_history + async with engine.begin() as connection: + await connection.execute(sqlalchemy.update(MODEL).where(MODEL.id == ids[2]).values(summary='tampered')) + for resource_type, scanned, tampered in [('bot', 3, 1), ('pipeline', 2, 0), ('model', 1, 0), ('bot', 3, 1)]: + result = await service.query_logs(WORKSPACE, resource_type=resource_type) + assert result['scanned_count'] == scanned + assert result['tampered_count'] == tampered + assert result['chain_failed_count'] == 0 + assert len(_cached_views(service)) == 3 + + +async def test_integrity_drilldown_keeps_real_predecessors(mixed_history): + service, engine, ids = mixed_history + damaged_ids = {ids[0], ids[4]} + async with engine.begin() as connection: + await connection.execute(sqlalchemy.update(MODEL).where(MODEL.id.in_(damaged_ids)).values(summary='tampered')) + result = await service.query_logs(WORKSPACE, resource_type='bot', integrity='hash_mismatch') + assert {row['id'] for row in result['records']} == damaged_ids + assert result['tampered_count'] == result['integrity_failed_count'] == 2 + assert result['chain_failed_count'] == 0 + assert all(row['chain_ok'] and not row['integrity_ok'] for row in result['records']) + + +async def test_filtered_scan_limit_keeps_real_predecessors(mixed_history, monkeypatch): + service, _engine, _ids = mixed_history + monkeypatch.setattr(service_module, 'MAX_INTEGRITY_SCAN_ROWS', 2) + result = await service.query_logs(WORKSPACE, resource_type='bot') + assert result['total'] == 3 + assert result['scanned_count'] == 2 + assert result['scan_truncated'] is True + assert result['tampered_count'] == 0 + assert all(row['chain_ok'] for row in result['records'])