diff --git a/src/langbot/pkg/api/http/controller/group.py b/src/langbot/pkg/api/http/controller/group.py index 1b102ff09..079020893 100644 --- a/src/langbot/pkg/api/http/controller/group.py +++ b/src/langbot/pkg/api/http/controller/group.py @@ -387,6 +387,28 @@ class RouterGroup(abc.ABC): if candidate is not None: resource_id = str(candidate) + # Routes without an explicit resource_param still name the resource + # they act on, either as a URL parameter or inside the payload of an + # install-style endpoint. Resolving that identity here gives every + # resource family a "which plugin/skill/knowledge base" trace without + # touching each handler. The body is read defensively: an audit path + # must never fail (or block) because a request carries no JSON. + try: + # Imported lazily so the base controller never participates in an + # import cycle with the service layer at module load time. + from ..service import settings as settings_service + + request_body: dict[str, typing.Any] | None = None + if str(quart.request.method or '').upper() not in ('GET', 'HEAD', 'OPTIONS'): + raw_body = await quart.request.get_json(silent=True) + if isinstance(raw_body, dict): + request_body = raw_body + identity = settings_service.resolve_resource_identity(kwargs, request_body) + if identity: + resource_id = identity + except Exception: # pragma: no cover - identity resolution is best effort + pass + # Handlers may publish request-local traceability facts when a # static route declaration cannot describe the runtime diff, e.g. # the previous and the new role of a member. ``quart.g`` is diff --git a/src/langbot/pkg/api/http/controller/groups/settings.py b/src/langbot/pkg/api/http/controller/groups/settings.py index e69e3d704..ee906859b 100644 --- a/src/langbot/pkg/api/http/controller/groups/settings.py +++ b/src/langbot/pkg/api/http/controller/groups/settings.py @@ -169,6 +169,9 @@ class SettingsRouterGroup(group.RouterGroup): level=_parse_int(level_raw, 0) if level_raw not in (None, '') else None, since=_parse_timestamp(args.get('since')), until=_parse_timestamp(args.get('until')), + # Optional verification filter: lets the panel turn the three + # tamper counters into a drill-down instead of a dead badge. + integrity=args.get('integrity') or None, ) return self.success(data=result) diff --git a/src/langbot/pkg/api/http/service/settings.py b/src/langbot/pkg/api/http/service/settings.py index 3b0907851..a9c541b47 100644 --- a/src/langbot/pkg/api/http/service/settings.py +++ b/src/langbot/pkg/api/http/service/settings.py @@ -28,6 +28,8 @@ import enum import hashlib import hmac import json +import logging +import time import typing import sqlalchemy @@ -44,6 +46,8 @@ from ....utils import constants from ..authz import WorkspaceRole from ..context import PrincipalType, RequestContext +logger = logging.getLogger(__name__) + # --------------------------------------------------------------------------- # Storage keys (WorkspaceMetadata.value is a short string column) @@ -77,6 +81,30 @@ MAX_DEDUPE_WINDOW_SECONDS = 3600 MAX_PAGE_SIZE = 200 DEFAULT_PAGE_SIZE = 50 +#: How many newest records the integrity summary reads before the numbers +#: beside the list are served from an approximate scan. Verifying a record means +#: recomputing its HMAC, so an unbounded scan on every page/refresh would make the +#: panel latency grow with the history size. ``MAX_MAX_ROWS`` (500k) is the +#: configured ceiling for a Workspace, so a full-table verification at that scale +#: is exactly what this bound avoids. +MAX_INTEGRITY_SCAN_ROWS = 20000 + +#: ``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). +INTEGRITY_FILTER_ALL = 'all' +INTEGRITY_FILTER_ISSUES = 'issues' +INTEGRITY_FILTER_HASH_MISMATCH = 'hash_mismatch' +INTEGRITY_FILTER_CHAIN_BROKEN = 'chain_broken' +_INTEGRITY_FILTERS = frozenset( + { + INTEGRITY_FILTER_ALL, + INTEGRITY_FILTER_ISSUES, + INTEGRITY_FILTER_HASH_MISMATCH, + INTEGRITY_FILTER_CHAIN_BROKEN, + } +) + #: Upper bound for a single export so a download cannot load the whole table. MAX_EXPORT_ROWS = 10000 @@ -559,6 +587,59 @@ def changed_fields( return changes +#: Ordered key groups that identify the resource a request acts on. Each group +#: is joined with ``/`` so ``author`` + ``plugin_name`` renders as +#: ``author/plugin_name`` — the same identity the UI already shows. Only these +#: keys are consulted: reading arbitrary payload fields would let a caller +#: inject unbounded, unattributed text into the audit trail. +_IDENTITY_KEY_GROUPS: typing.Final[tuple[tuple[str, ...], ...]] = ( + ('plugin_author', 'plugin_name'), + ('author', 'plugin_name'), + ('author', 'name'), + ('skill_uuid',), + ('knowledge_base_uuid',), + ('server_uuid',), + ('pipeline_uuid',), + ('provider_uuid',), + ('model_uuid',), + ('account_uuid',), + ('email',), + ('name',), + ('uuid',), +) + + +def resolve_resource_identity( + path_params: typing.Mapping[str, typing.Any] | None, + body: typing.Mapping[str, typing.Any] | None, +) -> str | None: + """Best-effort identity of the resource one request acts on. + + Path parameters describe the resource a route was registered for and are + therefore trusted; the request body is only consulted for install-style + endpoints that carry the identity in their payload (a marketplace install + names the plugin in the body, not in the URL). The result is bounded and + sensitive-looking keys are skipped so a secret can never be echoed back + through the trace. + """ + + params = path_params or {} + payload = body or {} + for group in _IDENTITY_KEY_GROUPS: + parts: list[str] = [] + for key in group: + value = params.get(key) + if value is None: + value = payload.get(key) + if value is None or value == '' or is_sensitive_field(key): + parts = [] + break + parts.append(str(value)) + if parts: + return typing.cast(str, truncate('/'.join(parts), _MAX_RESOURCE_ID_CHARS)) + return None + + def build_summary(rule: ActionRule, changes: list[dict[str, typing.Any]]) -> str: """Build a short change digest stored alongside the record. @@ -1231,11 +1312,25 @@ class WorkspaceSettingsService: level: int | None = None, since: datetime.datetime | None = None, until: datetime.datetime | None = None, + integrity: str | None = None, ) -> dict[str, typing.Any]: - """Return one page of operation records plus a coarse summary.""" + """Return one page of operation records plus a Workspace-wide summary. + + The three verification counters describe the whole filtered history, not + just the returned page: an operator opening page 2 must still see that + 39 records are tampered. Only ``total`` and ``records`` follow the + pagination window. + + ``integrity`` optionally narrows the listing to the records that failed + verification, which lets the panel make its counters actionable instead + of decorative. + """ resolved_limit = max(1, min(int(limit or DEFAULT_PAGE_SIZE), MAX_PAGE_SIZE)) resolved_offset = max(int(offset or 0), 0) + resolved_integrity = (integrity or INTEGRITY_FILTER_ALL).strip().lower() + if resolved_integrity not in _INTEGRITY_FILTERS: + resolved_integrity = INTEGRITY_FILTER_ALL try: model = persistence_operation_log.WorkspaceOperationLog @@ -1253,14 +1348,32 @@ class WorkspaceSettingsService: if until is not None: filters.append(model.created_at <= until) + # Verify the filtered history once per request. This replaces the + # previous per-page sum, which made the counters silently restart at + # zero on every page after the first. + summary = await self._integrity_summary(model, filters) + + # ``total`` follows the active listing filter so the pagination badge + # and the pager stay consistent with what the operator asked to see. + visible_filters = list(filters) + if resolved_integrity == INTEGRITY_FILTER_ISSUES: + ids = summary['tampered_ids'] + visible_filters.append(model.id.in_(ids) if ids else sqlalchemy.false()) + elif resolved_integrity == INTEGRITY_FILTER_HASH_MISMATCH: + ids = summary['integrity_failed_ids'] + visible_filters.append(model.id.in_(ids) if ids else sqlalchemy.false()) + elif resolved_integrity == INTEGRITY_FILTER_CHAIN_BROKEN: + ids = summary['chain_failed_ids'] + visible_filters.append(model.id.in_(ids) if ids else sqlalchemy.false()) + total_result = await self.ap.persistence_mgr.execute_async( - sqlalchemy.select(sqlalchemy.func.count()).select_from(model).where(*filters) + sqlalchemy.select(sqlalchemy.func.count()).select_from(model).where(*visible_filters) ) total = int(total_result.scalar_one_or_none() or 0) rows_result = await self.ap.persistence_mgr.execute_async( sqlalchemy.select(model) - .where(*filters) + .where(*visible_filters) .order_by(model.id.desc()) .limit(resolved_limit) .offset(resolved_offset) @@ -1287,10 +1400,12 @@ class WorkspaceSettingsService: ) for index, row in enumerate(rows) ] + integrity_summary = summary['summary'] except Exception as exc: # pragma: no cover - defensive self.ap.logger.debug(f'Operation log query skipped: {exc}') total = 0 records = [] + integrity_summary = self._empty_integrity_summary() return { 'records': records, @@ -1299,12 +1414,123 @@ class WorkspaceSettingsService: 'offset': resolved_offset, # Report the two failure modes separately so the panel can tell a # content edit (integrity) apart from a dropped link (chain) instead - # of collapsing both into a single "tampered" signal. - 'tampered_count': sum(1 for record in records if record.get('tampered')), - 'integrity_failed_count': sum(1 for record in records if not record.get('integrity_ok')), - 'chain_failed_count': sum(1 for record in records if not record.get('chain_ok')), + # of collapsing both into a single "tampered" signal. The counters + # cover the whole filtered history, not only this page. + 'tampered_count': integrity_summary['tampered'], + 'integrity_failed_count': integrity_summary['integrity_failed'], + 'chain_failed_count': integrity_summary['chain_failed'], + 'scanned_count': integrity_summary['scanned'], + 'scan_truncated': integrity_summary['truncated'], + 'integrity_filter': resolved_integrity, } + @staticmethod + def _empty_integrity_summary() -> dict[str, typing.Any]: + return { + 'tampered': 0, + 'integrity_failed': 0, + 'chain_failed': 0, + 'scanned': 0, + 'truncated': False, + } + + async def _integrity_summary( + self, + model: typing.Any, + filters: list[typing.Any], + ) -> dict[str, typing.Any]: + """Verify the filtered history and classify every failing record. + + Returns the three counters, the number of rows actually verified and the + record id lists needed to narrow the listing to one failure mode. The + scan walks the records newest-first and stops at + :data:`MAX_INTEGRITY_SCAN_ROWS`, so both the counters and the id lists + always describe the same, well-defined slice of the history. + """ + + started = time.monotonic() + try: + rows_result = await self.ap.persistence_mgr.execute_async( + sqlalchemy.select(model) + .where(*filters) + .order_by(model.id.desc()) + .limit(MAX_INTEGRITY_SCAN_ROWS) + ) + rows = list(rows_result.all()) + + oldest_id = rows[-1].id if rows else None + previous_row = None + verification_failed = False + if oldest_id is not None: + # Check the link of the oldest verified row against the row that + # precedes it, even when it sits outside the scan window: the + # baseline must still be read from the table, not guessed, or the + # boundary record would be reported as broken forever. + older_result = await self.ap.persistence_mgr.execute_async( + sqlalchemy.select(model) + .where(model.workspace_uuid == rows[0].workspace_uuid, model.id < oldest_id) + .order_by(model.id.desc()) + .limit(1) + ) + previous_row = older_result.first() + + tampered_ids: list[int] = [] + integrity_failed_ids: list[int] = [] + chain_failed_ids: list[int] = [] + for index, row in enumerate(rows): + predecessor = rows[index + 1] if index + 1 < len(rows) else previous_row + record = self._serialize_log(row, previous_row=predecessor) + if not record['integrity_ok']: + integrity_failed_ids.append(row.id) + if not record['chain_ok']: + chain_failed_ids.append(row.id) + if record['tampered']: + tampered_ids.append(row.id) + verification_failed = True + + summary = { + 'tampered': len(tampered_ids), + 'integrity_failed': len(integrity_failed_ids), + 'chain_failed': len(chain_failed_ids), + 'scanned': len(rows), + 'truncated': len(rows) >= MAX_INTEGRITY_SCAN_ROWS, + } + if verification_failed or summary['truncated']: + # Worth logging: either the store was edited underneath us, or the + # history outgrew the verification window and the counters became + # approximate. + logger.warning( + 'Operation log integrity scan: %s tampered / %s hash / %s chain over %s rows (truncated=%s)', + summary['tampered'], + summary['integrity_failed'], + summary['chain_failed'], + summary['scanned'], + summary['truncated'], + ) + elapsed_ms = int((time.monotonic() - started) * 1000) + if elapsed_ms >= 250: + logger.debug( + 'Operation log integrity scan took %sms over %s rows', + elapsed_ms, + summary['scanned'], + ) + return { + 'summary': summary, + 'tampered_ids': tampered_ids, + 'integrity_failed_ids': integrity_failed_ids, + 'chain_failed_ids': chain_failed_ids, + } + except Exception as exc: # pragma: no cover - defensive + # Never take the whole log panel down because verification failed: + # fall back to zeroed counters and an unfiltered listing. + logger.debug(f'Operation log integrity summary skipped: {exc}') + return { + 'summary': self._empty_integrity_summary(), + 'tampered_ids': [], + 'integrity_failed_ids': [], + 'chain_failed_ids': [], + } + 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. diff --git a/web/src/app/home/components/workspace-settings/OperationTracePanel.tsx b/web/src/app/home/components/workspace-settings/OperationTracePanel.tsx index 62fc7bf96..429affa67 100644 --- a/web/src/app/home/components/workspace-settings/OperationTracePanel.tsx +++ b/web/src/app/home/components/workspace-settings/OperationTracePanel.tsx @@ -34,6 +34,7 @@ import { import type { OperationChangeField, OperationGovernance, + OperationIntegrityFilter, OperationLevel, OperationLogFilters, OperationLogPage, @@ -206,6 +207,19 @@ export default function OperationTracePanel({ // total badge disagree with the list) whenever a page is re-fetched. const visibleRecords = page?.records ?? []; + const integrityFilter: OperationIntegrityFilter = query.integrity ?? 'all'; + + // Clicking a counter turns it into a drill-down: the panel is the only place + // the operator can learn *which* records failed verification, so a badge that + // cannot be acted on would be a dead end. + function toggleIntegrityFilter(next: OperationIntegrityFilter) { + setQuery((prev) => ({ + ...prev, + integrity: prev.integrity === next ? undefined : next, + offset: 0, + })); + } + async function changeLevel(level: OperationLevel) { if (!canConfigure) return; setSaving(true); @@ -235,6 +249,7 @@ export default function OperationTracePanel({ level: query.level, since: query.since, until: query.until, + integrity: query.integrity, }); const response = await backendClient.downloadFile(url); const disposition = response.headers['content-disposition'] as @@ -450,22 +465,44 @@ export default function OperationTracePanel({ {t('operationTrace.records')} {total} - {(page?.tampered_count ?? 0) > 0 && - (page?.integrity_failed_count ?? 0) > 0 && ( + {/* The two failure modes are rendered independently: a record can + drop its chain link without corrupting its own hash, so gating + one badge on the other counter would hide a real mismatch. + The counters cover the whole filtered history, so they stay + stable while the operator pages through the records. */} + {(page?.integrity_failed_count ?? 0) > 0 && ( + + )} + {(page?.chain_failed_count ?? 0) > 0 && ( + + )} + {page?.scan_truncated && ( + + {t('operationTrace.scanTruncated', { + count: page?.scanned_count ?? 0, + })} + + )}