⚡ Bolt: HMAC-SHA256 Blockchain for Escalation Audits

backend/cache.py

@@ -180,5 +180,6 @@ 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)
+audit_last_hash_cache = ThreadSafeCache(ttl=3600, max_size=2)
 visit_last_hash_cache = ThreadSafeCache(ttl=3600, max_size=2)
 user_issues_cache = ThreadSafeCache(ttl=300, max_size=50) # 5 minutes TTL

backend/escalation_engine.py

@@ -4,11 +4,16 @@
 """
 
 import datetime
+import hashlib
+import hmac
+import os
 from typing import List, Dict, Any, Optional
 from sqlalchemy.orm import Session
 from sqlalchemy import and_, or_
 from backend.models import Grievance, Jurisdiction, EscalationAudit, GrievanceStatus, JurisdictionLevel, EscalationReason, SeverityLevel
 from backend.database import SessionLocal
+from backend.cache import audit_last_hash_cache
+from backend.config import get_config
 from backend.routing_service import RoutingService
 from backend.sla_config_service import SLAConfigService
 
@@ -41,8 +46,10 @@ def evaluate_and_escalate_grievances(self, db: Session = None) -> Dict[str, int]
         Returns:
             Dictionary with escalation statistics
         """
+        should_close = False
         if db is None:
             db = SessionLocal()
+            should_close = True
 
         try:
             # Get grievances that need evaluation
@@ -63,7 +70,7 @@ def evaluate_and_escalate_grievances(self, db: Session = None) -> Dict[str, int]
             }
 
         finally:
-            if db is not SessionLocal():
+            if should_close:
                 db.close()
 
     def escalate_grievance_severity(self, grievance_id: int, new_severity: SeverityLevel,
@@ -80,8 +87,10 @@ def escalate_grievance_severity(self, grievance_id: int, new_severity: SeverityL
         Returns:
             True if escalation successful
         """
+        should_close = False
         if db is None:
             db = SessionLocal()
+            should_close = True
 
         try:
             grievance = db.query(Grievance).filter(Grievance.id == grievance_id).first()
@@ -108,7 +117,7 @@ def escalate_grievance_severity(self, grievance_id: int, new_severity: SeverityL
             print(f"Error escalating grievance severity: {e}")
             return False
         finally:
-            if db is not SessionLocal():
+            if should_close:
                 db.close()
 
     def manual_escalate(self, grievance_id: int, reason: str = "", db: Session = None) -> bool:
@@ -123,8 +132,10 @@ def manual_escalate(self, grievance_id: int, reason: str = "", db: Session = Non
         Returns:
             True if escalation successful
         """
+        should_close = False
         if db is None:
             db = SessionLocal()
+            should_close = True
 
         try:
             grievance = db.query(Grievance).filter(Grievance.id == grievance_id).first()
@@ -134,7 +145,7 @@ def manual_escalate(self, grievance_id: int, reason: str = "", db: Session = Non
             return self._escalate_grievance(grievance, EscalationReason.MANUAL, db, reason)
 
         finally:
-            if db is not SessionLocal():
+            if should_close:
                 db.close()
 
     def _get_grievances_for_evaluation(self, db: Session) -> List[Grievance]:
@@ -170,7 +181,13 @@ def _should_escalate(self, grievance: Grievance, db: Session) -> bool:
         """
         # Check if SLA is breached
         now = datetime.datetime.now(datetime.timezone.utc)
-        if grievance.sla_deadline >= now:
+
+        # Handle naive datetimes from SQLite
+        deadline = grievance.sla_deadline
+        if deadline and deadline.tzinfo is None:
+            deadline = deadline.replace(tzinfo=datetime.timezone.utc)
+
+        if deadline >= now:
             return False
 
         # Check if escalation is possible
@@ -248,18 +265,45 @@ def _escalate_grievance(self, grievance: Grievance, reason: EscalationReason,
             # Recalculate SLA
             self._recalculate_sla(grievance, db)
 
+            # Optimized Blockchain logic: Cache-first retrieval to ensure O(1) creation path
+            prev_hash = audit_last_hash_cache.get("last_hash")
+            if prev_hash is None:
+                # Cache miss: Fetch only the last hash from DB
+                last_audit = db.query(EscalationAudit.integrity_hash).order_by(EscalationAudit.id.desc()).first()
+                prev_hash = last_audit[0] if last_audit and last_audit[0] else ""
+                # Populate cache for subsequent escalations
+                audit_last_hash_cache.set(data=prev_hash, key="last_hash")
+
+            # HMAC-SHA256 chaining: hash(grievance_id|prev_auth|new_auth|reason|prev_hash)
+            # Using centralized config to avoid hardcoded secret fallbacks (Security compliance)
+            app_config = get_config()
+            secret_key = app_config.secret_key.encode('utf-8')
+            reason_val = reason.value if hasattr(reason, 'value') else reason
+            hash_content = f"{grievance.id}|{previous_authority}|{grievance.assigned_authority}|{reason_val}|{prev_hash}"
+
+            integrity_hash = hmac.new(
+                secret_key,
+                hash_content.encode('utf-8'),
+                hashlib.sha256
+            ).hexdigest()
+
             # Create audit log
             audit_log = EscalationAudit(
                 grievance_id=grievance.id,
                 previous_authority=previous_authority,
                 new_authority=grievance.assigned_authority,
                 reason=reason,
-                notes=notes
+                notes=notes,
+                integrity_hash=integrity_hash,
+                previous_integrity_hash=prev_hash
             )
 
             db.add(audit_log)
             db.commit()
 
+            # Update cache after successful commit to maintain chain integrity
+            audit_last_hash_cache.set(data=integrity_hash, key="last_hash")
+
             return True
 
         except Exception as e:

backend/init_db.py

@@ -219,6 +219,19 @@ def index_exists(table, index_name):
                 if not index_exists("resolution_evidence", "ix_resolution_evidence_previous_integrity_hash"):
                     conn.execute(text("CREATE INDEX IF NOT EXISTS ix_resolution_evidence_previous_integrity_hash ON resolution_evidence (previous_integrity_hash)"))
 
+            # Escalation Audits Table Migrations
+            if inspector.has_table("escalation_audits"):
+                if not column_exists("escalation_audits", "integrity_hash"):
+                    conn.execute(text("ALTER TABLE escalation_audits ADD COLUMN integrity_hash VARCHAR"))
+                    logger.info("Added integrity_hash column to escalation_audits")
+
+                if not column_exists("escalation_audits", "previous_integrity_hash"):
+                    conn.execute(text("ALTER TABLE escalation_audits ADD COLUMN previous_integrity_hash VARCHAR"))
+                    logger.info("Added previous_integrity_hash column to escalation_audits")
+
+                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)"))
+
             # Resolution Proof Tokens Table Migrations
             if inspector.has_table("resolution_proof_tokens"):
                 if not column_exists("resolution_proof_tokens", "nonce"):

backend/models.py

@@ -125,6 +125,10 @@ class EscalationAudit(Base):
     reason = Column(Enum(EscalationReason), nullable=False)
     notes = Column(Text, nullable=True)  # Additional context
 
+    # Blockchain integrity fields
+    integrity_hash = Column(String, nullable=True)
+    previous_integrity_hash = Column(String, nullable=True, index=True)
+
     # Relationships
     grievance = relationship("Grievance", back_populates="audit_logs")
 

backend/routers/grievances.py

@@ -6,6 +6,7 @@
 import json
 import logging
 import hashlib
+import hmac
 from datetime import datetime, timezone
 
 from backend.database import get_db
@@ -21,6 +22,7 @@
 )
 from backend.grievance_service import GrievanceService
 from backend.closure_service import ClosureService
+from backend.config import get_config
 
 logger = logging.getLogger(__name__)
 
@@ -192,9 +194,8 @@ def manual_escalate_grievance(
             raise HTTPException(status_code=404, detail="Grievance not found")
 
         # Perform manual escalation
-        success = grievance_service.escalation_engine.escalate_grievance_severity(
+        success = grievance_service.escalation_engine.manual_escalate(
             grievance_id=grievance_id,
-            new_severity=grievance.severity,  # Keep same severity, just escalate jurisdiction
             reason=reason,
             db=db
         )
@@ -496,3 +497,59 @@ 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("/audit/{audit_id}/blockchain-verify", response_model=BlockchainVerificationResponse)
+def verify_audit_blockchain(
+    audit_id: int,
+    db: Session = Depends(get_db)
+):
+    """
+    Verify the cryptographic integrity of an escalation audit record using HMAC-SHA256 chaining.
+    Optimized: Uses previous_integrity_hash column for O(1) verification.
+    """
+    try:
+        audit = db.query(EscalationAudit).filter(EscalationAudit.id == audit_id).first()
+
+        if not audit:
+            raise HTTPException(status_code=404, detail="Audit record not found")
+
+        # Determine previous hash (O(1) from stored column)
+        prev_hash = audit.previous_integrity_hash or ""
+
+        # HMAC-SHA256 chaining: hash(grievance_id|prev_auth|new_auth|reason|prev_hash)
+        # Using centralized config for secret key to ensure security compliance
+        app_config = get_config()
+        secret_key = app_config.secret_key.encode('utf-8')
+        reason_val = audit.reason.value if hasattr(audit.reason, 'value') else audit.reason
+        hash_content = f"{audit.grievance_id}|{audit.previous_authority}|{audit.new_authority}|{reason_val}|{prev_hash}"
+
+        computed_hash = hmac.new(
+            secret_key,
+            hash_content.encode('utf-8'),
+            hashlib.sha256
+        ).hexdigest()
+
+        if audit.integrity_hash is None:
+            is_valid = False
+            message = "No integrity hash present for this audit record; cryptographic integrity cannot be verified."
+        else:
+            is_valid = hmac.compare_digest(computed_hash, audit.integrity_hash)
+            message = (
+                "Integrity verified. This escalation audit record is cryptographically sealed."
+                if is_valid
+                else "Integrity check failed! The audit data does not match its cryptographic seal."
+            )
+
+        return BlockchainVerificationResponse(
+            is_valid=is_valid,
+            current_hash=audit.integrity_hash,
+            computed_hash=computed_hash,
+            message=message
+        )
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Error verifying audit blockchain for {audit_id}: {e}", exc_info=True)
+        raise HTTPException(status_code=500, detail="Failed to verify audit integrity")

backend/schemas.py

@@ -223,6 +223,8 @@ class EscalationAuditResponse(BaseModel):
     new_authority: str = Field(..., description="New authority after escalation")
     timestamp: datetime = Field(..., description="When the escalation occurred")
     reason: str = Field(..., description="Reason for escalation (SLA_BREACH, SEVERITY_UPGRADE, MANUAL)")
+    integrity_hash: Optional[str] = Field(None, description="Cryptographic integrity seal")
+    previous_integrity_hash: Optional[str] = Field(None, description="Linked hash for O(1) verification")
 
 class GrievanceSummaryResponse(BaseModel):
     id: int = Field(..., description="Grievance ID")