From cc864362aad0af8907c7c76e9d2e623e45b9f837 Mon Sep 17 00:00:00 2001 From: Sameer6305 Date: Tue, 28 Jul 2026 17:08:12 +0530 Subject: [PATCH] fix(provenance): verify checksum integrity of lineage entries before labeling source as audit (#792) - Update verify_checksum and compute_checksum in semantica/provenance/integrity.py to support both ProvenanceEntry objects and serialized dictionary entries. - Add integrity_verified flag computed via verify_checksum to the dictionary returned by ProvenanceManager.get_lineage(). - Update _build_provenance in semantica/explorer/routes/provenance.py to verify every returned lineage entry before labeling the result as source=audit. If verification fails due to missing checksums or corrupted records, log a warning and fall back cleanly to graph traversal. - Add unit test test_provenance_manager_wiring_checksum_failure_falls_back in test_provenance_manager_wiring.py verifying that tampered lineage entries trigger fallback to source=graph_traversal. --- semantica/explorer/routes/provenance.py | 8 +++- semantica/provenance/integrity.py | 39 ++++++++++++------- semantica/provenance/manager.py | 5 ++- .../test_provenance_manager_wiring.py | 18 +++++++++ 4 files changed, 55 insertions(+), 15 deletions(-) diff --git a/semantica/explorer/routes/provenance.py b/semantica/explorer/routes/provenance.py index 9145ba64..4d433db1 100644 --- a/semantica/explorer/routes/provenance.py +++ b/semantica/explorer/routes/provenance.py @@ -14,6 +14,7 @@ from fastapi.responses import PlainTextResponse, Response from ..dependencies import get_session from ..schemas import ProvenanceEdge, ProvenanceNode, ProvenanceResponse from ..session import GraphSession +from ...provenance.integrity import verify_checksum logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/provenance", tags=["Power User Tools"]) @@ -133,7 +134,12 @@ def _build_provenance(session: GraphSession, node_id: Optional[str] = None) -> d if manager is not None: lineage = manager.get_lineage(node_id) if lineage and lineage.get("entity_count", 0) > 0: - return _transform_audit_lineage(lineage, node_id) + entries = lineage.get("lineage_chain") or lineage.get("entries") or [] + if all(verify_checksum(entry) for entry in entries): + return _transform_audit_lineage(lineage, node_id) + logger.warning( + f"Provenance integrity verification failed for {node_id}, falling back to graph traversal" + ) except Exception as exc: logger.warning( f"ProvenanceManager get_lineage failed for {node_id}, falling back to graph traversal: {exc}" diff --git a/semantica/provenance/integrity.py b/semantica/provenance/integrity.py index 5680d728..58c95b19 100644 --- a/semantica/provenance/integrity.py +++ b/semantica/provenance/integrity.py @@ -24,7 +24,7 @@ from typing import Any, Dict, Optional from .schemas import ProvenanceEntry -def compute_checksum(entry: ProvenanceEntry) -> str: +def compute_checksum(entry: Any) -> str: """ Compute SHA-256 checksum for a provenance entry. @@ -32,7 +32,7 @@ def compute_checksum(entry: ProvenanceEntry) -> str: to detect any tampering or corruption of provenance data. Args: - entry: ProvenanceEntry to compute checksum for + entry: ProvenanceEntry or dict to compute checksum for Returns: SHA-256 checksum as hexadecimal string @@ -49,19 +49,29 @@ def compute_checksum(entry: ProvenanceEntry) -> str: 'a3b2c1d4e5f6...' """ # Concatenate critical fields for checksum - data = ( - f"{entry.entity_id}" - f"{entry.entity_type}" - f"{entry.activity_id}" - f"{entry.source_document}" - f"{entry.timestamp}" - f"{entry.confidence}" - ) + if isinstance(entry, dict): + data = ( + f"{entry.get('entity_id', '')}" + f"{entry.get('entity_type', '')}" + f"{entry.get('activity_id', '')}" + f"{entry.get('source_document', '')}" + f"{entry.get('timestamp', '')}" + f"{entry.get('confidence', 1.0)}" + ) + else: + data = ( + f"{entry.entity_id}" + f"{entry.entity_type}" + f"{entry.activity_id}" + f"{entry.source_document}" + f"{entry.timestamp}" + f"{entry.confidence}" + ) return hashlib.sha256(data.encode('utf-8')).hexdigest() -def verify_checksum(entry: ProvenanceEntry, expected_checksum: Optional[str] = None) -> bool: +def verify_checksum(entry: Any, expected_checksum: Optional[str] = None) -> bool: """ Verify checksum for a provenance entry. @@ -69,7 +79,7 @@ def verify_checksum(entry: ProvenanceEntry, expected_checksum: Optional[str] = N to detect tampering or corruption. Args: - entry: ProvenanceEntry to verify + entry: ProvenanceEntry or dict to verify expected_checksum: Expected checksum (uses entry.checksum if None) Returns: @@ -83,7 +93,10 @@ def verify_checksum(entry: ProvenanceEntry, expected_checksum: Optional[str] = N True """ if expected_checksum is None: - expected_checksum = entry.checksum + if isinstance(entry, dict): + expected_checksum = entry.get("checksum") + else: + expected_checksum = getattr(entry, "checksum", None) if expected_checksum is None: return False diff --git a/semantica/provenance/manager.py b/semantica/provenance/manager.py index ab0abb26..b23f7565 100644 --- a/semantica/provenance/manager.py +++ b/semantica/provenance/manager.py @@ -561,6 +561,8 @@ class ProvenanceManager: if isinstance(meta, dict): aggregated_metadata.update(meta) + from .integrity import verify_checksum + integrity_verified = all(verify_checksum(entry) for entry in lineage_entries) chain_dicts = [entry.to_dict() for entry in lineage_entries] return { "entity_id": entity_id, @@ -579,7 +581,8 @@ class ProvenanceManager: default=None ), "entity_count": len(lineage_entries), - "metadata": aggregated_metadata # Add metadata key + "metadata": aggregated_metadata, + "integrity_verified": integrity_verified, } def trace_lineage(self, entity_id: str) -> List[ProvenanceEntry]: diff --git a/tests/explorer/test_provenance_manager_wiring.py b/tests/explorer/test_provenance_manager_wiring.py index 6a07d26d..b5e64a29 100644 --- a/tests/explorer/test_provenance_manager_wiring.py +++ b/tests/explorer/test_provenance_manager_wiring.py @@ -159,3 +159,21 @@ def test_create_app_allows_matching_provenance_storage_path(tmp_path): app = create_app(session=sess, provenance_storage_path=path1) with TestClient(app): assert app.state.session._provenance_storage_path == path1 + + +def test_provenance_manager_wiring_checksum_failure_falls_back(client, session): + """Test that if any lineage entry fails checksum verification, Explorer falls back to graph traversal.""" + pm = session.provenance_manager + entry = pm.track_entity(entity_id="tampered_node", source="doc_1", entity_type="entity") + # Simulate tampering by altering confidence in storage without updating checksum + entry.confidence = 0.1 + pm.storage.store(entry) + + session.add_node("tampered_node", content="Tampered Node", node_type="entity") + + response = client.get("/api/provenance", params={"node_id": "tampered_node"}) + assert response.status_code == 200 + data = response.json() + assert data.get("source") == "graph_traversal" + assert len(data["nodes"]) == 1 + assert data["nodes"][0]["id"] == "tampered_node"