Cryptographic Hash Generation: Implementation & Validation in ESI Processing Pipelines

import asyncio
import hashlib
import json
import logging
import os
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import AsyncGenerator, List, Optional

import aiofiles

# ---------------------------------------------------------------------------
# Structured JSON Logging Configuration
# ---------------------------------------------------------------------------
class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        return json.dumps({
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName
        })

logger = logging.getLogger("esi_hash_pipeline")
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)

# ---------------------------------------------------------------------------
# Data Models
# ---------------------------------------------------------------------------
@dataclass
class HashResult:
    file_path: str
    sha256: str
    file_size_bytes: int
    chunk_size_bytes: int
    processed_at: float
    status: str = "SUCCESS"
    fallback_algorithm: Optional[str] = None
    error_context: Optional[str] = None

@dataclass
class DeadLetterRecord:
    file_path: str
    error_type: str
    error_message: str
    timestamp: float
    retry_count: int = 0

# ---------------------------------------------------------------------------
# Pipeline Constants
# ---------------------------------------------------------------------------
CHUNK_SIZE = 4 * 1024 * 1024  # 4 MiB optimal for NVMe/SSD throughput vs RAM
MAX_CONCURRENCY = 16          # Prevents thread pool exhaustion on high-core nodes

# ---------------------------------------------------------------------------
# Core Hash Computation
# ---------------------------------------------------------------------------
async def compute_file_hash(
    file_path: Path, 
    chunk_size: int = CHUNK_SIZE
) -> HashResult:
    """
    Memory-aware async hash computation with deterministic fallback.
    Aligns with NIST SP 800-107 Rev 1 recommendations for cryptographic hashing.
    """
    sha256 = hashlib.sha256()
    md5 = hashlib.md5()
    fallback_active = False

    try:
        async with aiofiles.open(file_path, mode="rb") as fh:
            while True:
                chunk = await fh.read(chunk_size)
                if not chunk:
                    break
                sha256.update(chunk)
                md5.update(chunk)
    except (PermissionError, OSError, asyncio.TimeoutError) as exc:
        logger.warning("Async read failed for %s, initiating synchronous fallback: %s", file_path, exc)
        fallback_active = True
        # Reset both hashers: the async attempt may have consumed partial data,
        # so re-read the file from the start to produce a clean, deterministic digest.
        sha256 = hashlib.sha256()
        md5 = hashlib.md5()
        with open(file_path, "rb") as fh:
            while chunk := fh.read(chunk_size // 2):
                sha256.update(chunk)
                md5.update(chunk)
    except Exception as exc:
        raise RuntimeError(f"Unrecoverable I/O failure during hash computation: {exc}") from exc

    digest = sha256.hexdigest()
    if len(digest) != 64 or any(c not in "0123456789abcdef" for c in digest):
        raise ValueError("Invalid SHA-256 digest generated; algorithm mismatch detected")

    if fallback_active:
        # The parallel MD5 digest provides an independent integrity signal for
        # forensic cross-validation when the primary async read path failed.
        logger.info("Fallback MD5 digest computed for %s: %s", file_path, md5.hexdigest())

    return HashResult(
        file_path=str(file_path.resolve()),
        sha256=digest,
        file_size_bytes=os.path.getsize(file_path),
        chunk_size_bytes=chunk_size,
        processed_at=time.time(),
        fallback_algorithm="SYNC_STREAM_FALLBACK" if fallback_active else None,
        status="SUCCESS"
    )

# ---------------------------------------------------------------------------
# Async Batch Processor with Concurrency Control
# ---------------------------------------------------------------------------
async def batch_hash_processor(
    file_queue: AsyncGenerator[Path, None],
    semaphore: asyncio.Semaphore,
    dead_letter_queue: List[DeadLetterRecord]
) -> AsyncGenerator[HashResult, None]:
    """
    Concurrency-controlled batch processor with explicit DLQ routing.
    Yields validated HashResult objects or logs failures for forensic triage.
    """
    async for file_path in file_queue:
        async with semaphore:
            try:
                result = await compute_file_hash(file_path)
                logger.info(json.dumps({
                    "event": "hash_complete", 
                    "file": str(file_path), 
                    "sha256": result.sha256
                }))
                yield result
            except Exception as exc:
                dlq_record = DeadLetterRecord(
                    file_path=str(file_path),
                    error_type=type(exc).__name__,
                    error_message=str(exc),
                    timestamp=time.time()
                )
                dead_letter_queue.append(dlq_record)
                logger.error(json.dumps({
                    "event": "hash_failure", 
                    "file": str(file_path), 
                    "error": str(exc)
                }))

# ---------------------------------------------------------------------------
# Execution Entry Point
# ---------------------------------------------------------------------------
async def run_pipeline(source_dir: Path) -> None:
    semaphore = asyncio.Semaphore(MAX_CONCURRENCY)
    dlq: List[DeadLetterRecord] = []
    
    async def file_generator():
        for path in source_dir.rglob("*"):
            if path.is_file() and not path.name.startswith("."):
                yield path

    async for result in batch_hash_processor(file_generator(), semaphore, dlq):
        # Downstream routing: register hash in deduplication index or metadata store
        pass

    if dlq:
        logger.warning(f"Pipeline completed. {len(dlq)} files routed to dead-letter queue for review.")