⚡ Bolt: Implement blockchain integrity for ClosureConfirmation

.jules/bolt.md

@@ -61,3 +61,7 @@
 ## 2025-02-13 - API Route Prefix Consistency
 **Learning:** Inconsistent application of `/api` prefixes between `main.py` router mounting and test suite request paths can lead to 404 errors during testing, even if the logic is correct. This is especially prevalent when multiple agents work on the same codebase with different assumptions about global prefixes.
 **Action:** Always verify that `app.include_router` in `backend/main.py` uses `prefix="/api"` if the test suite (e.g., `tests/test_blockchain.py`) expects it. If a router is mounted without a prefix, ensure tests are updated or the prefix is added to `main.py` to maintain repository-wide consistency.
+
+## 2026-02-12 - Atomicity of Blockchain Cache Updates
+**Learning:** When using in-memory caches to store the "latest hash" for blockchain chaining, updating the cache before a successful database commit can lead to "cache poisoning" if the transaction fails. This results in future records being chained to a hash that doesn't exist in the database, breaking the chain's integrity.
+**Action:** Always perform `cache.set()` operations for blockchain hashes strictly after `db.commit()` has succeeded.

backend/cache.py

@@ -180,6 +180,7 @@ def invalidate(self):
 blockchain_last_hash_cache = ThreadSafeCache(ttl=3600, max_size=1)
 grievance_last_hash_cache = ThreadSafeCache(ttl=3600, max_size=1)
 resolution_last_hash_cache = ThreadSafeCache(ttl=3600, max_size=1)
+closure_last_hash_cache = ThreadSafeCache(ttl=3600, max_size=1)
 visit_last_hash_cache = ThreadSafeCache(ttl=3600, max_size=2)
 audit_last_hash_cache = ThreadSafeCache(ttl=3600, max_size=2)
 user_issues_cache = ThreadSafeCache(ttl=300, max_size=50) # 5 minutes TTL

backend/closure_service.py

@@ -3,6 +3,10 @@
 from datetime import datetime, timedelta, timezone
 from backend.models import Grievance, GrievanceFollower, ClosureConfirmation, GrievanceStatus
 import logging
+import hashlib
+import hmac
+from backend.config import get_auth_config
+from backend.cache import closure_last_hash_cache
 
 logger = logging.getLogger(__name__)
 
@@ -86,16 +90,40 @@ def submit_confirmation(grievance_id: int, user_email: str, confirmation_type: s
         if existing:
             raise ValueError("You have already submitted a response for this closure")
 
+        # Blockchain feature: calculate integrity hash for the confirmation
+        # Performance Boost: Use thread-safe cache to eliminate DB query for last hash
+        prev_hash = closure_last_hash_cache.get("last_hash")
+        if prev_hash is None:
+            # Cache miss: Fetch only the last hash from DB
+            last_conf = db.query(ClosureConfirmation.integrity_hash).order_by(ClosureConfirmation.id.desc()).first()
+            prev_hash = last_conf[0] if last_conf and last_conf[0] else ""
+            closure_last_hash_cache.set(data=prev_hash, key="last_hash")
+
+        # Chaining logic: hash(grievance_id|user_email|confirmation_type|prev_hash)
+        hash_content = f"{grievance_id}|{user_email}|{confirmation_type}|{reason or ''}|{prev_hash}"
+
+        secret_key = get_auth_config().secret_key
+        integrity_hash = hmac.new(
+            secret_key.encode('utf-8'),
+            hash_content.encode('utf-8'),
+            hashlib.sha256
+        ).hexdigest()
+
         # Create confirmation record
         confirmation = ClosureConfirmation(
             grievance_id=grievance_id,
             user_email=user_email,
             confirmation_type=confirmation_type,
-            reason=reason
+            reason=reason,
+            integrity_hash=integrity_hash,
+            previous_integrity_hash=prev_hash
         )
         db.add(confirmation)
         db.commit()
 
+        # Update cache for next confirmation AFTER successful DB commit
+        closure_last_hash_cache.set(data=integrity_hash, key="last_hash")
+
         # Check if threshold is met
         return ClosureService.check_and_finalize_closure(grievance_id, db)
 

backend/init_db.py

@@ -232,6 +232,19 @@ def index_exists(table, index_name):
                 if not index_exists("escalation_audits", "ix_escalation_audits_previous_integrity_hash"):
                     conn.execute(text("CREATE INDEX IF NOT EXISTS ix_escalation_audits_previous_integrity_hash ON escalation_audits (previous_integrity_hash)"))
 
+            # Closure Confirmations Table Migrations
+            if inspector.has_table("closure_confirmations"):
+                if not column_exists("closure_confirmations", "integrity_hash"):
+                    conn.execute(text("ALTER TABLE closure_confirmations ADD COLUMN integrity_hash VARCHAR"))
+                    logger.info("Added integrity_hash column to closure_confirmations")
+
+                if not column_exists("closure_confirmations", "previous_integrity_hash"):
+                    conn.execute(text("ALTER TABLE closure_confirmations ADD COLUMN previous_integrity_hash VARCHAR"))
+                    logger.info("Added previous_integrity_hash column to closure_confirmations")
+
+                if not index_exists("closure_confirmations", "ix_closure_confirmations_previous_integrity_hash"):
+                    conn.execute(text("CREATE INDEX IF NOT EXISTS ix_closure_confirmations_previous_integrity_hash ON closure_confirmations (previous_integrity_hash)"))
+
             # Resolution Proof Tokens Table Migrations
             if inspector.has_table("resolution_proof_tokens"):
                 if not column_exists("resolution_proof_tokens", "nonce"):

backend/models.py

@@ -203,6 +203,10 @@ class ClosureConfirmation(Base):
     reason = Column(Text, nullable=True)  # Optional reason for dispute
     created_at = Column(DateTime, default=lambda: datetime.datetime.now(datetime.timezone.utc))
 
+    # Blockchain integrity fields
+    integrity_hash = Column(String, nullable=True)
+    previous_integrity_hash = Column(String, nullable=True, index=True)
+
     # Relationship
     grievance = relationship("Grievance", back_populates="closure_confirmations")
 

backend/routers/grievances.py

@@ -560,3 +560,63 @@ def verify_grievance_blockchain(
     except Exception as e:
         logger.error(f"Error verifying grievance blockchain for {grievance_id}: {e}", exc_info=True)
         raise HTTPException(status_code=500, detail="Failed to verify grievance integrity")
+
+
+@router.get("/closure-confirmation/{confirmation_id}/blockchain-verify", response_model=BlockchainVerificationResponse)
+def verify_closure_confirmation_blockchain(
+    confirmation_id: int,
+    db: Session = Depends(get_db)
+):
+    """
+    Verify the cryptographic integrity of a closure confirmation record using blockchain-style chaining.
+    Optimized: Uses previous_integrity_hash column for O(1) verification.
+    """
+    try:
+        confirmation = db.query(
+            ClosureConfirmation.grievance_id,
+            ClosureConfirmation.user_email,
+            ClosureConfirmation.confirmation_type,
+            ClosureConfirmation.integrity_hash,
+            ClosureConfirmation.previous_integrity_hash
+        ).filter(ClosureConfirmation.id == confirmation_id).first()
+
+        if not confirmation:
+            raise HTTPException(status_code=404, detail="Closure confirmation not found")
+
+        # Determine previous hash (O(1) from stored column)
+        prev_hash = confirmation.previous_integrity_hash or ""
+
+        # Recompute hash based on current data and previous hash
+        # Chaining logic: hash(grievance_id|user_email|confirmation_type|prev_hash)
+        hash_content = f"{confirmation.grievance_id}|{confirmation.user_email}|{confirmation.confirmation_type}|{prev_hash}"
+
+        secret_key = get_auth_config().secret_key
+        computed_hash = hmac.new(
+            secret_key.encode('utf-8'),
+            hash_content.encode('utf-8'),
+            hashlib.sha256
+        ).hexdigest()
+
+        if confirmation.integrity_hash is None:
+            is_valid = False
+            message = "No integrity hash present for this confirmation record; cryptographic integrity cannot be verified."
+        else:
+            is_valid = hmac.compare_digest(computed_hash, confirmation.integrity_hash)
+            message = (
+                "Integrity verified. This closure confirmation record is cryptographically sealed."
+                if is_valid
+                else "Integrity check failed! The confirmation data does not match its cryptographic seal."
+            )
+
+        return BlockchainVerificationResponse(
+            is_valid=is_valid,
+            current_hash=confirmation.integrity_hash,
+            computed_hash=computed_hash,
+            message=message
+        )
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Error verifying closure confirmation blockchain for {confirmation_id}: {e}", exc_info=True)
+        raise HTTPException(status_code=500, detail="Failed to verify confirmation integrity")

backend/tests/test_closure_blockchain.py

@@ -0,0 +1,118 @@
+import pytest
+from sqlalchemy.orm import Session
+from fastapi.testclient import TestClient
+from datetime import datetime, timezone, timedelta
+
+from backend.main import app
+from backend.database import Base, get_db, engine
+from backend.models import Grievance, GrievanceStatus, SeverityLevel, GrievanceFollower, Jurisdiction, JurisdictionLevel, ClosureConfirmation
+from backend.closure_service import ClosureService
+
+# Setup test database
+@pytest.fixture(name="db_session")
+def fixture_db_session():
+    Base.metadata.create_all(bind=engine)
+    session = Session(bind=engine)
+    yield session
+    session.close()
+    Base.metadata.drop_all(bind=engine)
+
+@pytest.fixture(name="client")
+def fixture_client(db_session):
+    def override_get_db():
+        try:
+            yield db_session
+        finally:
+            pass
+    app.dependency_overrides[get_db] = override_get_db
+    with TestClient(app) as c:
+        yield c
+    app.dependency_overrides.clear()
+
+def test_closure_confirmation_blockchain_chaining(client, db_session):
+    # 1. Setup a grievance and followers
+    jurisdiction = Jurisdiction(
+        level=JurisdictionLevel.LOCAL,
+        geographic_coverage={"cities": ["Mumbai"]},
+        responsible_authority="BMC",
+        default_sla_hours=24
+    )
+    db_session.add(jurisdiction)
+    db_session.flush()
+
+    grievance = Grievance(
+        unique_id="TEST-123",
+        category="Water",
+        severity=SeverityLevel.HIGH,
+        current_jurisdiction_id=jurisdiction.id,
+        assigned_authority="BMC",
+        sla_deadline=datetime.now(timezone.utc) + timedelta(hours=24),
+        status=GrievanceStatus.OPEN
+    )
+    db_session.add(grievance)
+    db_session.flush()
+
+    followers = [
+        GrievanceFollower(grievance_id=grievance.id, user_email=f"user{i}@example.com")
+        for i in range(5)
+    ]
+    for f in followers:
+        db_session.add(f)
+    db_session.commit()
+
+    # 2. Request closure
+    ClosureService.request_closure(grievance.id, db_session)
+
+    # 3. Submit multiple confirmations and verify chaining
+    emails = [f"user{i}@example.com" for i in range(3)]
+    conf_ids = []
+
+    for email in emails:
+        result = ClosureService.submit_confirmation(
+            grievance_id=grievance.id,
+            user_email=email,
+            confirmation_type="confirmed",
+            reason="Verified resolved",
+            db=db_session
+        )
+
+        # Get the record to check hashes
+        conf = db_session.query(ClosureConfirmation).filter(
+            ClosureConfirmation.user_email == email,
+            ClosureConfirmation.grievance_id == grievance.id
+        ).first()
+        conf_ids.append(conf.id)
+
+    # Verify the chain
+    conf1 = db_session.query(ClosureConfirmation).filter(ClosureConfirmation.id == conf_ids[0]).first()
+    conf2 = db_session.query(ClosureConfirmation).filter(ClosureConfirmation.id == conf_ids[1]).first()
+    conf3 = db_session.query(ClosureConfirmation).filter(ClosureConfirmation.id == conf_ids[2]).first()
+
+    assert conf1.previous_integrity_hash == ""
+    assert conf2.previous_integrity_hash == conf1.integrity_hash
+    assert conf3.previous_integrity_hash == conf2.integrity_hash
+
+    # 4. Verify via API endpoint
+    for cid in conf_ids:
+        response = client.get(f"/api/closure-confirmation/{cid}/blockchain-verify")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["is_valid"] is True
+        assert "Integrity verified" in data["message"]
+
+    # 5. Tamper with data and verify failure
+    conf2.confirmation_type = "disputed"
+    db_session.commit()
+
+    response = client.get(f"/api/closure-confirmation/{conf_ids[1]}/blockchain-verify")
+    assert response.status_code == 200
+    assert response.json()["is_valid"] is False
+    assert "Integrity check failed" in response.json()["message"]
+
+    # Subsequent record should still be valid if its OWN data and recorded previous_integrity_hash match.
+    # Blockchain integrity check for a single record only verifies that IT matches its seal.
+    # To detect a break in the chain, you would need to verify the previous record as well.
+    # This is consistent with O(1) single-record verification.
+    response = client.get(f"/api/closure-confirmation/{conf_ids[2]}/blockchain-verify")
+    assert response.status_code == 200
+    assert response.json()["is_valid"] is True