Integrity Verification

Version: 1.0 Date: October 08, 2025 SPDX-License-Identifier: BSD-3-Clause License File: See the LICENSE file in the project root. Copyright: © 2025 Michael Gardner, A Bit of Help, Inc. Authors: Michael Gardner, Claude Code Status: Active

Overview

Integrity verification ensures data hasn't been corrupted or tampered with during processing. The pipeline system uses cryptographic hash functions to calculate checksums at various stages, providing strong guarantees about data integrity.

The checksum service operates in two modes:

• Calculate Mode: Generates checksums for data chunks
• Verify Mode: Validates existing checksums to detect tampering

Supported Algorithms

SHA-256 (Recommended)

Industry-standard cryptographic hash function

#![allow(unused)]
fn main() {
use adaptive_pipeline_domain::value_objects::Algorithm;

let algorithm = Algorithm::sha256();
}

Characteristics:

• Hash Size: 256 bits (32 bytes)
• Security: Cryptographically secure, collision-resistant
• Performance: ~500 MB/s (software), ~2 GB/s (hardware accelerated)
• Use Cases: General-purpose integrity verification

When to Use:

• ✅ General-purpose integrity verification
• ✅ Compliance requirements (FIPS 180-4)
• ✅ Cross-platform compatibility
• ✅ Hardware acceleration available (SHA extensions)

SHA-512

Stronger variant of SHA-2 family

#![allow(unused)]
fn main() {
let algorithm = Algorithm::sha512();
}

Characteristics:

• Hash Size: 512 bits (64 bytes)
• Security: Higher security margin than SHA-256
• Performance: ~400 MB/s (software), faster on 64-bit systems
• Use Cases: High-security requirements, 64-bit optimized systems

When to Use:

• ✅ Maximum security requirements
• ✅ 64-bit systems (better performance)
• ✅ Long-term archival (future-proof security)
• ❌ Resource-constrained systems (larger output)

BLAKE3

Modern, high-performance cryptographic hash

#![allow(unused)]
fn main() {
let algorithm = Algorithm::blake3();
}

Characteristics:

• Hash Size: 256 bits (32 bytes, configurable)
• Security: Based on BLAKE2, ChaCha stream cipher
• Performance: ~3 GB/s (parallelizable, SIMD-optimized)
• Use Cases: High-throughput processing, modern systems

When to Use:

• ✅ Maximum performance requirements
• ✅ Large file processing (highly parallelizable)
• ✅ Modern CPUs with SIMD support
• ✅ No regulatory compliance requirements
• ❌ FIPS compliance needed (not FIPS certified)

Algorithm Comparison

Performance measured on Intel i7-10700K @ 3.8 GHz

Architecture

Service Interface

The domain layer defines the checksum service interface:

#![allow(unused)]
fn main() {
use adaptive_pipeline_domain::services::ChecksumService;
use adaptive_pipeline_domain::entities::ProcessingContext;
use adaptive_pipeline_domain::value_objects::FileChunk;
use adaptive_pipeline_domain::PipelineError;

/// Domain service for integrity verification
pub trait ChecksumService: Send + Sync {
    /// Process a chunk and update the running checksum
    fn process_chunk(
        &self,
        chunk: FileChunk,
        context: &mut ProcessingContext,
        stage_name: &str,
    ) -> Result<FileChunk, PipelineError>;

    /// Get the final checksum value
    fn get_checksum(
        &self,
        context: &ProcessingContext,
        stage_name: &str
    ) -> Option<String>;
}
}

Implementation

The infrastructure layer provides concrete implementations:

#![allow(unused)]
fn main() {
use adaptive_pipeline_domain::services::{ChecksumService, ChecksumProcessor};

/// Concrete checksum processor using SHA-256
pub struct ChecksumProcessor {
    pub algorithm: String,
    pub verify_existing: bool,
}

impl ChecksumProcessor {
    pub fn new(algorithm: String, verify_existing: bool) -> Self {
        Self {
            algorithm,
            verify_existing,
        }
    }

    /// Creates a SHA-256 processor
    pub fn sha256_processor(verify_existing: bool) -> Self {
        Self::new("SHA256".to_string(), verify_existing)
    }
}
}

Algorithm Implementations

SHA-256 Implementation

#![allow(unused)]
fn main() {
use sha2::{Digest, Sha256};

impl ChecksumProcessor {
    /// Calculate SHA-256 checksum
    pub fn calculate_sha256(&self, data: &[u8]) -> String {
        let mut hasher = Sha256::new();
        hasher.update(data);
        format!("{:x}", hasher.finalize())
    }

    /// Incremental SHA-256 hashing
    pub fn update_hash(&self, hasher: &mut Sha256, chunk: &FileChunk) {
        hasher.update(chunk.data());
    }

    /// Finalize hash and return hex string
    pub fn finalize_hash(&self, hasher: Sha256) -> String {
        format!("{:x}", hasher.finalize())
    }
}
}

Key Features:

• Incremental hashing for streaming large files
• Memory-efficient (constant 32-byte state)
• Hardware acceleration with SHA-NI instructions

SHA-512 Implementation

#![allow(unused)]
fn main() {
use sha2::{Sha512};

impl ChecksumProcessor {
    /// Calculate SHA-512 checksum
    pub fn calculate_sha512(&self, data: &[u8]) -> String {
        let mut hasher = Sha512::new();
        hasher.update(data);
        format!("{:x}", hasher.finalize())
    }
}
}

Key Features:

• 512-bit output for higher security margin
• Optimized for 64-bit architectures
• Suitable for long-term archival

BLAKE3 Implementation

#![allow(unused)]
fn main() {
use blake3::Hasher;

impl ChecksumProcessor {
    /// Calculate BLAKE3 checksum
    pub fn calculate_blake3(&self, data: &[u8]) -> String {
        let mut hasher = Hasher::new();
        hasher.update(data);
        hasher.finalize().to_hex().to_string()
    }

    /// Parallel BLAKE3 hashing
    pub fn calculate_blake3_parallel(&self, chunks: &[&[u8]]) -> String {
        let mut hasher = Hasher::new();
        for chunk in chunks {
            hasher.update(chunk);
        }
        hasher.finalize().to_hex().to_string()
    }
}
}

Key Features:

• Highly parallelizable (uses Rayon internally)
• SIMD-optimized for modern CPUs
• Incremental and streaming support
• Up to 6x faster than SHA-256

Chunk Processing

ChunkProcessor Trait

The checksum service implements the ChunkProcessor trait for integration with the pipeline:

#![allow(unused)]
fn main() {
use adaptive_pipeline_domain::services::file_processor_service::ChunkProcessor;

impl ChunkProcessor for ChecksumProcessor {
    /// Process chunk with checksum calculation/verification
    fn process_chunk(&self, chunk: &FileChunk) -> Result<FileChunk, PipelineError> {
        // Step 1: Verify existing checksum if requested
        if self.verify_existing && chunk.checksum().is_some() {
            let is_valid = chunk.verify_integrity()?;
            if !is_valid {
                return Err(PipelineError::IntegrityError(format!(
                    "Checksum verification failed for chunk {}",
                    chunk.sequence_number()
                )));
            }
        }

        // Step 2: Ensure chunk has checksum (calculate if missing)
        if chunk.checksum().is_none() {
            chunk.with_calculated_checksum()
        } else {
            Ok(chunk.clone())
        }
    }

    fn name(&self) -> &str {
        "ChecksumProcessor"
    }

    fn modifies_data(&self) -> bool {
        false // Only modifies metadata
    }
}
}

Integrity Verification

The FileChunk value object provides built-in integrity verification:

#![allow(unused)]
fn main() {
impl FileChunk {
    /// Verify chunk integrity against stored checksum
    pub fn verify_integrity(&self) -> Result<bool, PipelineError> {
        match &self.checksum {
            Some(stored_checksum) => {
                let calculated = Self::calculate_checksum(self.data());
                Ok(*stored_checksum == calculated)
            }
            None => Err(PipelineError::InvalidConfiguration(
                "No checksum to verify".to_string()
            )),
        }
    }

    /// Calculate checksum for chunk data
    fn calculate_checksum(data: &[u8]) -> String {
        use sha2::{Digest, Sha256};
        let mut hasher = Sha256::new();
        hasher.update(data);
        format!("{:x}", hasher.finalize())
    }

    /// Create new chunk with calculated checksum
    pub fn with_calculated_checksum(&self) -> Result<FileChunk, PipelineError> {
        let checksum = Self::calculate_checksum(self.data());
        Ok(FileChunk {
            sequence_number: self.sequence_number,
            data: self.data.clone(),
            checksum: Some(checksum),
            metadata: self.metadata.clone(),
        })
    }
}
}

Performance Optimizations

Parallel Chunk Processing

Process multiple chunks in parallel using Rayon:

#![allow(unused)]
fn main() {
use rayon::prelude::*;

impl ChecksumProcessor {
    /// Process chunks in parallel for maximum throughput
    pub fn process_chunks_parallel(
        &self,
        chunks: &[FileChunk]
    ) -> Result<Vec<FileChunk>, PipelineError> {
        chunks
            .par_iter()
            .map(|chunk| self.process_chunk(chunk))
            .collect()
    }
}
}

Performance Benefits:

• Linear Scaling: Performance scales with CPU cores
• No Contention: Each chunk processed independently
• 2-4x Speedup: On typical multi-core systems

Hardware Acceleration

Leverage CPU crypto extensions when available:

#![allow(unused)]
fn main() {
/// Check for SHA hardware acceleration
pub fn has_sha_extensions() -> bool {
    #[cfg(target_arch = "x86_64")]
    {
        is_x86_feature_detected!("sha")
    }
    #[cfg(not(target_arch = "x86_64"))]
    {
        false
    }
}

/// Select optimal algorithm based on hardware
pub fn optimal_hash_algorithm() -> Algorithm {
    if has_sha_extensions() {
        Algorithm::sha256() // Hardware accelerated
    } else {
        Algorithm::blake3() // Software optimized
    }
}
}

Memory Management

Minimize allocations during hash calculation:

#![allow(unused)]
fn main() {
impl ChecksumProcessor {
    /// Reuse buffer for hash calculations
    pub fn calculate_with_buffer(
        &self,
        data: &[u8],
        buffer: &mut Vec<u8>
    ) -> String {
        buffer.clear();
        buffer.extend_from_slice(data);
        self.calculate_sha256(buffer)
    }
}
}

Configuration

Stage Configuration

Configure integrity stages in your pipeline:

#![allow(unused)]
fn main() {
use adaptive_pipeline_domain::entities::PipelineStage;
use adaptive_pipeline_domain::value_objects::{Algorithm, StageType};

// Input integrity verification
let input_stage = PipelineStage::new(
    "input_checksum",
    StageType::Integrity,
    Algorithm::sha256(),
)?;

// Output integrity verification
let output_stage = PipelineStage::new(
    "output_checksum",
    StageType::Integrity,
    Algorithm::blake3(), // Faster for final verification
)?;
}

Verification Mode

Enable checksum verification for existing data:

#![allow(unused)]
fn main() {
// Calculate checksums only (default)
let processor = ChecksumProcessor::new("SHA256".to_string(), false);

// Verify existing checksums before processing
let verifying_processor = ChecksumProcessor::new("SHA256".to_string(), true);
}

Algorithm Selection

Choose algorithm based on requirements:

#![allow(unused)]
fn main() {
pub fn select_hash_algorithm(
    security_level: SecurityLevel,
    performance_priority: bool,
) -> Algorithm {
    match (security_level, performance_priority) {
        (SecurityLevel::Maximum, _) => Algorithm::sha512(),
        (SecurityLevel::High, false) => Algorithm::sha256(),
        (SecurityLevel::High, true) => Algorithm::blake3(),
        (SecurityLevel::Standard, _) => Algorithm::blake3(),
    }
}
}

Error Handling

Error Types

The service handles various error conditions:

#![allow(unused)]
fn main() {
pub enum IntegrityError {
    /// Checksum verification failed
    ChecksumMismatch {
        expected: String,
        actual: String,
        chunk: u64,
    },

    /// Invalid algorithm specified
    UnsupportedAlgorithm(String),

    /// Hash calculation failed
    HashCalculationError(String),

    /// Chunk data corrupted
    CorruptedData {
        chunk: u64,
        reason: String,
    },
}
}

Error Recovery

Handle integrity errors gracefully:

#![allow(unused)]
fn main() {
impl ChecksumProcessor {
    pub fn process_with_retry(
        &self,
        chunk: &FileChunk,
        max_retries: u32
    ) -> Result<FileChunk, PipelineError> {
        let mut attempts = 0;

        loop {
            match self.process_chunk(chunk) {
                Ok(result) => return Ok(result),
                Err(PipelineError::IntegrityError(msg)) if attempts < max_retries => {
                    attempts += 1;
                    eprintln!("Integrity check failed (attempt {}/{}): {}",
                        attempts, max_retries, msg);
                    continue;
                }
                Err(e) => return Err(e),
            }
        }
    }
}
}

Usage Examples

Basic Checksum Calculation

Calculate SHA-256 checksums for data:

#![allow(unused)]
fn main() {
use adaptive_pipeline_domain::services::ChecksumProcessor;

fn calculate_file_checksum(data: &[u8]) -> Result<String, PipelineError> {
    let processor = ChecksumProcessor::sha256_processor(false);
    let checksum = processor.calculate_sha256(data);
    Ok(checksum)
}

// Example usage
let data = b"Hello, world!";
let checksum = calculate_file_checksum(data)?;
println!("SHA-256: {}", checksum);
// Output: SHA-256: 315f5bdb76d078c43b8ac0064e4a0164612b1fce77c869345bfc94c75894edd3
}

Integrity Verification

Verify data hasn't been tampered with:

#![allow(unused)]
fn main() {
use adaptive_pipeline_domain::value_objects::FileChunk;

fn verify_chunk_integrity(chunk: &FileChunk) -> Result<bool, PipelineError> {
    let processor = ChecksumProcessor::sha256_processor(true);

    // Process with verification enabled
    match processor.process_chunk(chunk) {
        Ok(_) => Ok(true),
        Err(PipelineError::IntegrityError(_)) => Ok(false),
        Err(e) => Err(e),
    }
}

// Example usage
let chunk = FileChunk::new(0, data.to_vec())?
    .with_calculated_checksum()?;

if verify_chunk_integrity(&chunk)? {
    println!("✓ Chunk integrity verified");
} else {
    println!("✗ Chunk has been tampered with!");
}
}

Pipeline Integration

Integrate checksums into processing pipeline:

#![allow(unused)]
fn main() {
use adaptive_pipeline_domain::entities::{Pipeline, PipelineStage};

fn create_verified_pipeline() -> Result<Pipeline, PipelineError> {
    let stages = vec![
        // Input verification
        PipelineStage::new(
            "input_checksum",
            StageType::Integrity,
            Algorithm::sha256(),
        )?,

        // Processing stages...
        PipelineStage::new(
            "compression",
            StageType::Compression,
            Algorithm::zstd(),
        )?,

        // Output verification
        PipelineStage::new(
            "output_checksum",
            StageType::Integrity,
            Algorithm::sha256(),
        )?,
    ];

    Pipeline::new("verified-pipeline".to_string(), stages)
}
}

Parallel Processing

Process multiple chunks with maximum performance:

#![allow(unused)]
fn main() {
use rayon::prelude::*;

fn hash_large_file(chunks: Vec<FileChunk>) -> Result<Vec<String>, PipelineError> {
    let processor = ChecksumProcessor::sha256_processor(false);

    chunks.par_iter()
        .map(|chunk| processor.calculate_sha256(chunk.data()))
        .collect()
}

// Example: Hash 1000 chunks in parallel
let checksums = hash_large_file(chunks)?;
println!("Processed {} chunks", checksums.len());
}

Benchmarks

SHA-256 Performance

File Size: 100 MB, Chunk Size: 1 MB

SHA-512 Performance

File Size: 100 MB, Chunk Size: 1 MB

BLAKE3 Performance

File Size: 100 MB, Chunk Size: 1 MB

Test Environment: Intel i7-10700K @ 3.8 GHz, 32GB RAM, Ubuntu 22.04

Algorithm Recommendations by Use Case

Best Practices

Algorithm Selection

Choose the right algorithm for your requirements:

#![allow(unused)]
fn main() {
// Compliance requirements
if needs_fips_compliance {
    Algorithm::sha256() // FIPS 180-4 certified
}
// Maximum security
else if security_level == SecurityLevel::Maximum {
    Algorithm::sha512() // Stronger security margin
}
// Performance critical
else if throughput_priority {
    Algorithm::blake3() // 3-6x faster
}
// Default
else {
    Algorithm::sha256() // Industry standard
}
}

Verification Strategy

Implement defense-in-depth verification:

#![allow(unused)]
fn main() {
// 1. Input verification (detect source corruption)
let input_checksum_stage = PipelineStage::new(
    "input_verify",
    StageType::Integrity,
    Algorithm::sha256(),
)?;

// 2. Processing stages...

// 3. Output verification (detect processing corruption)
let output_checksum_stage = PipelineStage::new(
    "output_verify",
    StageType::Integrity,
    Algorithm::sha256(),
)?;
}

Performance Optimization

Optimize for your workload:

#![allow(unused)]
fn main() {
// Small files (<10 MB): Use single-threaded
if file_size < 10 * 1024 * 1024 {
    processor.calculate_sha256(data)
}
// Large files: Use parallel processing
else {
    processor.process_chunks_parallel(&chunks)
}

// Hardware acceleration available: Use SHA-256
if has_sha_extensions() {
    Algorithm::sha256()
}
// No hardware acceleration: Use BLAKE3
else {
    Algorithm::blake3()
}
}

Error Handling

Handle integrity failures appropriately:

#![allow(unused)]
fn main() {
match processor.process_chunk(&chunk) {
    Ok(verified_chunk) => {
        // Integrity verified, continue processing
        process_chunk(verified_chunk)
    }
    Err(PipelineError::IntegrityError(msg)) => {
        // Log error and attempt recovery
        eprintln!("Integrity failure: {}", msg);

        // Option 1: Retry from source
        let fresh_chunk = reload_chunk_from_source()?;
        processor.process_chunk(&fresh_chunk)
    }
    Err(e) => return Err(e),
}
}

Security Considerations

Cryptographic Strength

All supported algorithms are cryptographically secure:

• SHA-256: 128-bit security level (2^128 operations for collision)
• SHA-512: 256-bit security level (2^256 operations for collision)
• BLAKE3: 128-bit security level (based on ChaCha20)

Collision Resistance

Practical collision resistance:

#![allow(unused)]
fn main() {
// SHA-256 collision resistance: ~2^128 operations
// Effectively impossible with current technology
let sha256_security_bits = 128;

// SHA-512 collision resistance: ~2^256 operations
// Provides future-proof security margin
let sha512_security_bits = 256;
}

Tampering Detection

Checksums detect any modification:

#![allow(unused)]
fn main() {
// Even single-bit changes produce completely different hashes
let original = "Hello, World!";
let tampered = "Hello, world!"; // Changed 'W' to 'w'

let hash1 = processor.calculate_sha256(original.as_bytes());
let hash2 = processor.calculate_sha256(tampered.as_bytes());

assert_ne!(hash1, hash2); // Completely different hashes
}

Not for Authentication

Important: Checksums alone don't provide authentication:

#![allow(unused)]
fn main() {
// ❌ WRONG: Checksum alone doesn't prove authenticity
let checksum = calculate_sha256(data);
// Attacker can modify data AND update checksum

// ✅ CORRECT: Use HMAC for authentication
let hmac = calculate_hmac_sha256(data, secret_key);
// Attacker cannot forge HMAC without secret key
}

Use HMAC or digital signatures for authentication.

Next Steps

Now that you understand integrity verification:

• Repositories - Data persistence patterns
• Binary Format - File format with embedded checksums
• Error Handling - Comprehensive error strategies
• Performance - Advanced optimization techniques