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

@@ -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}"

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

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]:

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"