Checksum Error: 23% Files Truncated in S3 Migration

import hashlib
import zlib
import os
import time
from dataclasses import dataclass
from enum import Enum
from typing import Optional, Tuple


class ChecksumAlgorithm(Enum):
    CRC32 = 'crc32'
    MD5 = 'md5'
    SHA1 = 'sha1'
    SHA256 = 'sha256'
    SHA512 = 'sha512'


@dataclass
class ChecksumResult:
    algorithm: ChecksumAlgorithm
    hex_digest: str
    bytes_processed: int
    elapsed_ms: float
    throughput_mbps: float


class ChecksumComparator:
    """Compute and compare checksums across algorithms with performance benchmarks."""

    BUFFER_SIZE = 8 * 1024 * 1024  # 8MB read buffer

    def compute(self, filepath: str, algorithm: ChecksumAlgorithm) -> ChecksumResult:
        """Compute checksum of a file using the specified algorithm."""
        start = time.monotonic()
        bytes_processed = 0

        if algorithm == ChecksumAlgorithm.CRC32:
            crc = 0
            with open(filepath, 'rb') as f:
                while chunk := f.read(self.BUFFER_SIZE):
                    crc = zlib.crc32(chunk, crc)
                    bytes_processed += len(chunk)
            hex_digest = format(crc & 0xFFFFFFFF, '08x')
        else:
            hash_obj = hashlib.new(algorithm.value)
            with open(filepath, 'rb') as f:
                while chunk := f.read(self.BUFFER_SIZE):
                    hash_obj.update(chunk)
                    bytes_processed += len(chunk)
            hex_digest = hash_obj.hexdigest()

        elapsed = time.monotonic() - start
        throughput = (bytes_processed / (1024 * 1024)) / elapsed if elapsed > 0 else 0

        return ChecksumResult(
            algorithm=algorithm,
            hex_digest=hex_digest,
            bytes_processed=bytes_processed,
            elapsed_ms=elapsed * 1000,
            throughput_mbps=round(throughput, 1),
        )

    def verify(self, filepath: str, algorithm: ChecksumAlgorithm, expected: str) -> Tuple[bool, str]:
        """Verify a file's checksum against an expected value."""
        result = self.compute(filepath, algorithm)
        match = result.hex_digest.lower() == expected.lower()
        return match, result.hex_digest

    def benchmark_all(self, filepath: str) -> list:
        """Benchmark all algorithms on a single file."""
        results = []
        for algo in ChecksumAlgorithm:
            result = self.compute(filepath, algo)
            results.append({
                'algorithm': algo.value,
                'hex_digest': result.hex_digest,
                'throughput_mbps': result.throughput_mbps,
                'elapsed_ms': round(result.elapsed_ms, 1),
            })
        return sorted(results, key=lambda r: r['throughput_mbps'], reverse=True)

    def compare_two_files(self, file_a: str, file_b: str, algorithm: ChecksumAlgorithm) -> dict:
        """Compare checksums of two files to detect differences."""
        result_a = self.compute(file_a, algorithm)
        result_b = self.compute(file_b, algorithm)

        return {
            'algorithm': algorithm.value,
            'file_a': file_a,
            'checksum_a': result_a.hex_digest,
            'file_b': file_b,
            'checksum_b': result_b.hex_digest,
            'match': result_a.hex_digest == result_b.hex_digest,
            'size_a': result_a.bytes_processed,
            'size_b': result_b.bytes_processed,
        }

import hashlib
import os
import random
from dataclasses import dataclass
from typing import List, Optional


@dataclass
class CorruptionEvent:
    file_path: str
    offset: int
    original_byte: int
    corrupted_byte: int
    detection_method: str
    likely_cause: str


class CorruptionSimulator:
    """Simulate and detect various corruption patterns for testing checksum pipelines."""

    def flip_random_bit(self, data: bytearray, num_flips: int = 1) -> List[int]:
        """Flip random bits in data to simulate cosmic ray bit-flips."""
        offsets = []
        for _ in range(num_flips):
            byte_offset = random.randint(0, len(data) - 1)
            bit_offset = random.randint(0, 7)
            original = data[byte_offset]
            data[byte_offset] ^= (1 << bit_offset)
            offsets.append(byte_offset)
        return offsets

    def truncate_file(self, filepath: str, truncate_bytes: int) -> str:
        """Truncate a file to simulate incomplete writes."""
        with open(filepath, 'rb') as f:
            data = f.read()
        truncated_path = filepath + '.truncated'
        with open(truncated_path, 'wb') as f:
            f.write(data[:-truncate_bytes])
        return truncated_path

    def inject_block_corruption(self, data: bytearray, block_size: int = 4096) -> int:
        """Corrupt an entire block to simulate disk sector failure."""
        block_index = random.randint(0, (len(data) // block_size) - 1)
        offset = block_index * block_size
        for i in range(min(block_size, len(data) - offset)):
            data[offset + i] = 0xFF  # all bits set — classic failing NAND pattern
        return offset

    def verify_integrity(self, filepath: str, expected_sha256: str) -> dict:
        """Verify file integrity against expected SHA-256 hash."""
        sha256 = hashlib.sha256()
        with open(filepath, 'rb') as f:
            while chunk := f.read(8 * 1024 * 1024):
                sha256.update(chunk)

        actual = sha256.hexdigest()
        match = actual == expected_sha256

        return {
            'file': filepath,
            'expected': expected_sha256,
            'actual': actual,
            'match': match,
            'status': 'OK' if match else 'CHECKSUM MISMATCH',
        }

    def diagnose_corruption_pattern(self, original: bytes, corrupted: bytes) -> dict:
        """Analyze corruption pattern to suggest likely cause."""
        if len(original) != len(corrupted):
            return {
                'pattern': 'truncation',
                'likely_cause': 'Incomplete write, network timeout, or filesystem full',
                'severity': 'HIGH',
            }

        bit_flips = 0
        byte_diffs = 0
        consecutive_diffs = 0
        max_consecutive = 0
        in_diff_block = False

        for i in range(len(original)):
            if original[i] != corrupted[i]:
                byte_diffs += 1
                bit_flips += bin(original[i] ^ corrupted[i]).count('1')
                if not in_diff_block:
                    consecutive_diffs += 1
                    in_diff_block = True
                else:
                    consecutive_diffs += 1
                max_consecutive = max(max_consecutive, consecutive_diffs)
            else:
                in_diff_block = False
                consecutive_diffs = 0

        if byte_diffs == 1 and bit_flips == 1:
            return {
                'pattern': 'single_bit_flip',
                'likely_cause': 'Cosmic ray or RAM bit-flip',
                'severity': 'LOW',
            }
        elif max_consecutive >= 4096 and (max_consecutive % 4096 == 0 or max_consecutive % 512 == 0):
            return {
                'pattern': 'block_corruption',
                'likely_cause': 'Disk sector failure or SSD NAND wear',
                'severity': 'CRITICAL',
            }
        elif byte_diffs > 0 and bit_flips > byte_diffs * 4:
            return {
                'pattern': 'multi_bit_burst',
                'likely_cause': 'Network corruption, bad cable, or NIC failure',
                'severity': 'HIGH',
            }
        else:
            return {
                'pattern': 'scattered_corruption',
                'likely_cause': 'Memory corruption, software bug, or concurrent write',
                'severity': 'HIGH',
            }

import hashlib
import json
import os
import sqlite3
import time
from dataclasses import dataclass
from typing import Optional, List, Tuple
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed


@dataclass
class ManifestEntry:
    relative_path: str
    size_bytes: int
    sha256: str
    mtime: float
    verified: bool = False
    verified_at: Optional[float] = None


class MigrationVerifier:
    """Production-grade migration verification with SQLite manifest and parallel checking."""

    def __init__(self, manifest_db_path: str):
        self.manifest_db = manifest_db_path
        self._init_db()

    def _init_db(self):
        """Initialize SQLite manifest database."""
        conn = sqlite3.connect(self.manifest_db)
        conn.execute('''
            CREATE TABLE IF NOT EXISTS manifest (
                relative_path TEXT PRIMARY KEY,
                size_bytes INTEGER,
                sha256 TEXT,
                mtime REAL,
                verified INTEGER DEFAULT 0,
                verified_at REAL,
                destination_sha256 TEXT,
                status TEXT DEFAULT 'pending'
            )
        ''')
        conn.commit()
        conn.close()

    def generate_baseline(self, source_dir: str, max_workers: int = 8) -> dict:
        """Generate SHA-256 manifest for all files in source directory."""
        source_path = Path(source_dir)
        files = []

        for root, dirs, filenames in os.walk(source_path):
            for filename in filenames:
                filepath = Path(root) / filename
                files.append(filepath)

        entries = []
        errors = []

        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {
                executor.submit(self._hash_file, f, source_path): f
                for f in files
            }
            for future in as_completed(futures):
                filepath = futures[future]
                try:
                    entry = future.result()
                    entries.append(entry)
                except Exception as e:
                    errors.append({'file': str(filepath), 'error': str(e)})

        conn = sqlite3.connect(self.manifest_db)
        for entry in entries:
            conn.execute(
                'INSERT OR REPLACE INTO manifest (relative_path, size_bytes, sha256, mtime) VALUES (?, ?, ?, ?)',
                (entry.relative_path, entry.size_bytes, entry.sha256, entry.mtime)
            )
        conn.commit()
        conn.close()

        return {
            'total_files': len(entries),
            'total_bytes': sum(e.size_bytes for e in entries),
            'errors': len(errors),
            'error_files': errors[:10],
        }

    def _hash_file(self, filepath: Path, base_dir: Path) -> ManifestEntry:
        """Compute SHA-256 hash and metadata for a single file."""
        sha256 = hashlib.sha256()
        size = 0
        with open(filepath, 'rb') as f:
            while chunk := f.read(8 * 1024 * 1024):
                sha256.update(chunk)
                size += len(chunk)

        return ManifestEntry(
            relative_path=str(filepath.relative_to(base_dir)),
            size_bytes=size,
            sha256=sha256.hexdigest(),
            mtime=os.path.getmtime(filepath),
        )

    def verify_destination(self, dest_dir: str, max_workers: int = 8) -> dict:
        """Verify all destination files against the manifest."""
        dest_path = Path(dest_dir)
        conn = sqlite3.connect(self.manifest_db)
        cursor = conn.execute('SELECT relative_path, sha256, size_bytes FROM manifest')
        entries = cursor.fetchall()
        conn.close()

        matches = 0
        mismatches = []
        missing = []
        size_errors = []

        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {}
            for rel_path, expected_sha256, expected_size in entries:
                dest_file = dest_path / rel_path
                if not dest_file.exists():
                    missing.append(rel_path)
                    continue

                actual_size = os.path.getsize(dest_file)
                if actual_size != expected_size:
                    size_errors.append({
                        'path': rel_path,
                        'expected_size': expected_size,
                        'actual_size': actual_size,
                    })
                    continue

                future = executor.submit(self._verify_single, dest_file, expected_sha256, rel_path)
                futures[future] = rel_path

            for future in as_completed(futures):
                rel_path = futures[future]
                match, actual_sha256 = future.result()
                if match:
                    matches += 1
                else:
                    mismatches.append({
                        'path': rel_path,
                        'expected': expected_sha256,
                        'actual': actual_sha256,
                    })

        # Update manifest with verification results
        conn = sqlite3.connect(self.manifest_db)
        for m in mismatches:
            conn.execute(
                'UPDATE manifest SET status = ?, destination_sha256 = ?, verified_at = ? WHERE relative_path = ?',
                ('mismatch', m['actual'], time.time(), m['path'])
            )
        for rel_path in missing:
            conn.execute(
                'UPDATE manifest SET status = ?, verified_at = ? WHERE relative_path = ?',
                ('missing', time.time(), rel_path)
            )
        conn.commit()
        conn.close()

        return {
            'total_checked': len(entries),
            'matches': matches,
            'mismatches': len(mismatches),
            'missing': len(missing),
            'size_errors': len(size_errors),
            'mismatch_details': mismatches[:20],
            'missing_files': missing[:20],
            'success_rate': f'{(matches / len(entries) * 100):.2f}%' if entries else 'N/A',
        }

    def _verify_single(self, filepath: Path, expected_sha256: str, rel_path: str) -> Tuple[bool, str]:
        """Verify a single file's checksum."""
        sha256 = hashlib.sha256()
        with open(filepath, 'rb') as f:
            while chunk := f.read(8 * 1024 * 1024):
                sha256.update(chunk)
        actual = sha256.hexdigest()
        return actual == expected_sha256, actual

import struct
import socket
from dataclasses import dataclass
from typing import Optional, Tuple


@dataclass
class ChecksumValidation:
    layer: str
    computed: int
    received: int
    match: bool
    algorithm: str


class NetworkChecksumAnalyzer:
    """Analyze and validate checksums in network protocol headers."""

    def compute_ip_checksum(self, header: bytes) -> int:
        """Compute IP header checksum (RFC 791 one's complement sum)."""
        if len(header) % 2 != 0:
            header += b'\x00'

        total = 0
        for i in range(0, len(header), 2):
            word = (header[i] << 8) + header[i + 1]
            total += word

        # Fold 32-bit sum to 16 bits
        while total >> 16:
            total = (total & 0xFFFF) + (total >> 16)

        return ~total & 0xFFFF

    def compute_tcp_checksum(self, pseudo_header: bytes, tcp_segment: bytes) -> int:
        """Compute TCP checksum including pseudo-header (RFC 793)."""
        data = pseudo_header + tcp_segment
        if len(data) % 2 != 0:
            data += b'\x00'

        total = 0
        for i in range(0, len(data), 2):
            word = (data[i] << 8) + data[i + 1]
            total += word

        while total >> 16:
            total = (total & 0xFFFF) + (total >> 16)

        return ~total & 0xFFFF

    def validate_ip_packet(self, packet: bytes) -> ChecksumValidation:
        """Validate IP header checksum of a raw packet."""
        header_length = (packet[0] & 0x0F) * 4
        header = bytearray(packet[:header_length])

        # Zero out checksum field for computation
        received_checksum = (header[10] << 8) + header[11]
        header[10] = 0
        header[11] = 0

        computed = self.compute_ip_checksum(bytes(header))

        return ChecksumValidation(
            layer='IP',
            computed=computed,
            received=received_checksum,
            match=computed == received_checksum,
            algorithm='one\'s complement sum (16-bit)',
        )

    def detect_offload_artifact(self, packet: bytes) -> dict:
        """Detect if a checksum error is caused by NIC offloading rather than real corruption."""
        ip_result = self.validate_ip_packet(packet)

        # Check if checksum field is zero — common sign of offloading
        header_length = (packet[0] & 0x0F) * 4
        checksum_field = (packet[10] << 8) + packet[11]

        if checksum_field == 0:
            return {
                'diagnosis': 'CHECKSUM_OFFLOAD',
                'explanation': 'NIC computed checksum after capture. The zero checksum field indicates hardware offloading is enabled.',
                'action': 'Disable offloading with ethtool -K <iface> tx-checksumming off to capture real checksums.',
                'real_corruption': False,
            }

        if not ip_result.match:
            return {
                'diagnosis': 'REAL_CORRUPTION',
                'explanation': f'IP checksum mismatch: computed={ip_result.computed:#06x}, received={ip_result.received:#06x}',
                'action': 'Check network hardware: cables, NIC, switch ports. Run cable tester if possible.',
                'real_corruption': True,
            }

        return {
            'diagnosis': 'OK',
            'explanation': 'IP checksum valid. No corruption detected at this layer.',
            'action': 'No action required.',
            'real_corruption': False,
        }

import subprocess
import json
import re
from dataclasses import dataclass
from typing import Optional, List, Dict


@dataclass
class StorageIntegrityReport:
    filesystem: str
    checksum_enabled: bool
    scrub_status: Optional[str]
    errors_found: int
    errors_corrected: int
    recommendations: List[str]


class StorageIntegrityChecker:
    """Check and report on filesystem-level checksum configuration and integrity status."""

    def check_zfs_integrity(self, pool_name: str) -> StorageIntegrityReport:
        """Check ZFS pool integrity status and scrub history."""
        recommendations = []

        # Get pool status
        try:
            result = subprocess.run(
                ['zpool', 'status', '-v', pool_name],
                capture_output=True, text=True, timeout=30
            )
            output = result.stdout
        except (subprocess.TimeoutExpired, FileNotFoundError) as e:
            return StorageIntegrityReport(
                filesystem='zfs',
                checksum_enabled=True,
                scrub_status=f'ERROR: {e}',
                errors_found=-1,
                errors_corrected=-1,
                recommendations=['Cannot query ZFS pool status'],
            )

        # Parse errors
        errors_found = 0
        errors_corrected = 0
        if 'No known data errors' in output:
            errors_found = 0
        else:
            error_match = re.search(r'(\d+) data errors?', output)
            if error_match:
                errors_found = int(error_match.group(1))

        # Check scrub status
        scrub_status = 'unknown'
        if 'scrub repaired' in output:
            scrub_match = re.search(r'scrub repaired (\S+) in', output)
            if scrub_match:
                scrub_status = f'last scrub repaired {scrub_match.group(1)}'
        elif 'scrub in progress' in output:
            scrub_status = 'scrub in progress'
        else:
            scrub_status = 'no recent scrub found'
            recommendations.append('Run zpool scrub to verify all blocks')

        # Check for degraded pool
        if 'DEGRADED' in output:
            recommendations.append('Pool is DEGRADED — replace failed disk immediately')

        # Check checksum algorithm
        if 'sha256' in output.lower() or 'skein' in output.lower():
            recommendations.append('Using strong checksum algorithm (SHA-256 or Skein)')
        elif 'fletcher4' in output.lower():
            recommendations.append('Using fletcher4 — consider upgrading to SHA-256 for better collision resistance')

        return StorageIntegrityReport(
            filesystem='zfs',
            checksum_enabled=True,
            scrub_status=scrub_status,
            errors_found=errors_found,
            errors_corrected=errors_corrected,
            recommendations=recommendations,
        )

    def check_ext4_integrity(self, device: str) -> StorageIntegrityReport:
        """Check ext4 metadata checksum configuration."""
        recommendations = []

        try:
            result = subprocess.run(
                ['tune2fs', '-l', device],
                capture_output=True, text=True, timeout=30
            )
            output = result.stdout
        except (subprocess.TimeoutExpired, FileNotFoundError) as e:
            return StorageIntegrityReport(
                filesystem='ext4',
                checksum_enabled=False,
                scrub_status=f'ERROR: {e}',
                errors_found=-1,
                errors_corrected=-1,
                recommendations=['Cannot query ext4 filesystem'],
            )

        metadata_csum = 'metadata_csum' in output
        journal_checksum = 'journal_checksum' in output

        if not metadata_csum:
            recommendations.append('CRITICAL: metadata_csum not enabled — ext4 cannot detect metadata corruption')
            recommendations.append('Enable with: tune2fs -O metadata_csum ' + device)

        if not journal_checksum:
            recommendations.append('journal_checksum not enabled — journal corruption may be silent')

        recommendations.append('ext4 has no data checksums — consider ZFS or btrfs for integrity-critical workloads')

        return StorageIntegrityReport(
            filesystem='ext4',
            checksum_enabled=metadata_csum,
            scrub_status='ext4 has no scrub — use e2fsck -f for manual check',
            errors_found=0,
            errors_corrected=0,
            recommendations=recommendations,
        )

    def check_s3_integrity(self, bucket: str, key: str, s3_client) -> dict:
        """Check S3 object integrity using available checksum methods."""
        response = s3_client.head_object(Bucket=bucket, Key=key)

        result = {
            'bucket': bucket,
            'key': key,
            'etag': response.get('ETag', '').strip('"'),
            'content_length': response.get('ContentLength', 0),
            'checksums': {},
            'recommendations': [],
        }

        # Check for additional checksum headers
        for algo in ['sha256', 'sha1', 'crc32', 'crc32c']:
            header = f'Checksum{algo.upper()}' if algo != 'sha256' else 'ChecksumSHA256'
            value = response.get(header) or response.get(f'x-amz-checksum-{algo}')
            if value:
                result['checksums'][algo] = value

        if not result['checksums']:
            result['recommendations'].append(
                'No additional checksum headers found. Only ETag (MD5) available. '
                'Consider uploading with x-amz-checksum-sha256 for stronger verification.'
            )

        if '-' in response.get('ETag', ''):
            result['recommendations'].append(
                'ETag contains "-" indicating multipart upload. '
                'ETag is a composite MD5, not a simple MD5 of the object content.'
            )

        return result

import hashlib
import zlib
import os
import time
import tempfile
from dataclasses import dataclass
from typing import Dict, List
from concurrent.futures import ThreadPoolExecutor


@dataclass
class BenchmarkResult:
    algorithm: str
    file_size_mb: float
    elapsed_ms: float
    throughput_mbps: float
    cpu_efficiency: str


class ChecksumBenchmark:
    """Benchmark checksum algorithms with realistic workloads."""

    def generate_test_file(self, size_mb: int) -> str:
        """Generate a test file with pseudo-random data."""
        filepath = os.path.join(tempfile.gettempdir(), f'checksum_bench_{size_mb}mb.dat')
        chunk_size = 8 * 1024 * 1024  # 8MB chunks
        bytes_written = 0

        with open(filepath, 'wb') as f:
            while bytes_written < size_mb * 1024 * 1024:
                remaining = min(chunk_size, size_mb * 1024 * 1024 - bytes_written)
                f.write(os.urandom(remaining))
                bytes_written += remaining

        return filepath

    def benchmark_single(self, filepath: str, algorithm: str) -> BenchmarkResult:
        """Benchmark a single algorithm on a file."""
        file_size = os.path.getsize(filepath)
        start = time.monotonic()

        if algorithm == 'crc32':
            crc = 0
            with open(filepath, 'rb') as f:
                while chunk := f.read(8 * 1024 * 1024):
                    crc = zlib.crc32(chunk, crc)
        else:
            h = hashlib.new(algorithm)
            with open(filepath, 'rb') as f:
                while chunk := f.read(8 * 1024 * 1024):
                    h.update(chunk)

        elapsed = time.monotonic() - start
        throughput = (file_size / (1024 * 1024)) / elapsed

        return BenchmarkResult(
            algorithm=algorithm,
            file_size_mb=file_size / (1024 * 1024),
            elapsed_ms=elapsed * 1000,
            throughput_mbps=round(throughput, 1),
            cpu_efficiency='hardware' if algorithm == 'crc32' else 'software',
        )

    def benchmark_parallel(self, filepath: str, algorithm: str, num_threads: int) -> BenchmarkResult:
        """Benchmark checksum computation with parallel chunk processing."""
        file_size = os.path.getsize(filepath)
        chunk_size = file_size // num_threads

        def hash_chunk(offset: int, size: int) -> str:
            h = hashlib.new(algorithm) if algorithm != 'crc32' else None
            crc = 0 if algorithm == 'crc32' else None
            with open(filepath, 'rb') as f:
                f.seek(offset)
                remaining = size
                while remaining > 0:
                    read_size = min(8 * 1024 * 1024, remaining)
                    chunk = f.read(read_size)
                    if algorithm == 'crc32':
                        crc = zlib.crc32(chunk, crc)
                    else:
                        h.update(chunk)
                    remaining -= len(chunk)
            return format(crc & 0xFFFFFFFF, '08x') if algorithm == 'crc32' else h.hexdigest()

        start = time.monotonic()

        with ThreadPoolExecutor(max_workers=num_threads) as executor:
            futures = []
            for i in range(num_threads):
                offset = i * chunk_size
                size = chunk_size if i < num_threads - 1 else file_size - offset
                futures.append(executor.submit(hash_chunk, offset, size))
            results = [f.result() for f in futures]

        elapsed = time.monotonic() - start
        throughput = (file_size / (1024 * 1024)) / elapsed

        return BenchmarkResult(
            algorithm=f'{algorithm}_parallel_{num_threads}',
            file_size_mb=file_size / (1024 * 1024),
            elapsed_ms=elapsed * 1000,
            throughput_mbps=round(throughput, 1),
            cpu_efficiency=f'{num_threads} threads',
        )

    def run_full_benchmark(self, size_mb: int = 1024) -> List[Dict]:
        """Run comprehensive benchmark across all algorithms."""
        filepath = self.generate_test_file(size_mb)
        algorithms = ['crc32', 'md5', 'sha1', 'sha256', 'sha512']
        results = []

        for algo in algorithms:
            result = self.benchmark_single(filepath, algo)
            results.append({
                'algorithm': algo,
                'throughput_mbps': result.throughput_mbps,
                'elapsed_ms': round(result.elapsed_ms, 1),
            })

        # Parallel benchmarks
        for threads in [2, 4, 8]:
            for algo in ['sha256', 'sha512']:
                result = self.benchmark_parallel(filepath, algo, threads)
                results.append({
                    'algorithm': result.algorithm,
                    'throughput_mbps': result.throughput_mbps,
                    'elapsed_ms': round(result.elapsed_ms, 1),
                })

        os.remove(filepath)
        return sorted(results, key=lambda r: r['throughput_mbps'], reverse=True)

// io.thecodeforge — cs-fundamentals tutorial

import hashlib
import struct

def position_dependent_checksum(data: bytes, block_number: int) -> int:
    """Generate a position-dependent checksum to prevent block-swapping attacks."""
    # Incorporate the block number into the hash input
    salted_data = struct.pack('>Q', block_number) + data
    # Use SHA-256 truncated to 64 bits for reasonable collision resistance
    hash_bytes = hashlib.sha256(salted_data).digest()[:8]
    return struct.unpack('>Q', hash_bytes)[0]

def verify_position_checksum(data: bytes, block_number: int, expected_checksum: int) -> bool:
    computed = position_dependent_checksum(data, block_number)
    return computed == expected_checksum

# Simulate a production block
block_data = b"This is a critical financial transaction record."
original_checksum = position_dependent_checksum(block_data, block_number=42)
print(f"Block 42 checksum: {original_checksum}")

# Now swap with block 99's data (simulated corruption)
fake_data = b"Fake transaction record."
print(f"Swapped block data passes? {verify_position_checksum(fake_data, 42, original_checksum)}")

# The attacker would need to recompute the checksum for the swapped position
print(f"Attacker recomputed checksum? {verify_position_checksum(fake_data, 42, position_dependent_checksum(fake_data, 42))}")

# Output shows that the original checksum does NOT match swapped data at the same position

// io.thecodeforge — cs-fundamentals tutorial

import hashlib
import struct

def validate_and_receive_packet(packet: bytes) -> tuple[bool, bytes]:
    """Atomic receiver-side verification: validate before releasing data."""
    # Packet format: 8-byte checksum | payload
    if len(packet) < 8:
        return False, b""

    received_checksum = struct.unpack('>Q', packet[:8])[0]
    payload = packet[8:]

    # Recompute checksum on the received payload only
    computed_checksum = struct.unpack('>Q', hashlib.sha256(payload).digest()[:8])[0]

    if computed_checksum != received_checksum:
        # CRITICAL: Do NOT return the payload even for logging
        return False, b""

    # Only now release the payload to the application
    return True, payload

# Simulate a corrupted packet
raw_packet = struct.pack('>Q', 12345678901234567890) + b"Hello, world!"
# Introduce a single bit flip
corrupted_packet = raw_packet[:10] + bytes([raw_packet[10] ^ 0x01]) + raw_packet[11:]

data_valid, data = validate_and_receive_packet(corrupted_packet)
print(f"Validation result: {data_valid}")
print(f"Released data (should be empty): {repr(data)}")

# Test with a valid packet
valid_checksum = struct.unpack('>Q', hashlib.sha256(b"Valid data").digest()[:8])[0]
valid_packet = struct.pack('>Q', valid_checksum) + b"Valid data"
data_valid2, data2 = validate_and_receive_packet(valid_packet)
print(f"Valid packet result: {data_valid2}")
print(f"Released data: {repr(data2)}")

// io.thecodeforge — cs-fundamentals tutorial

import hashlib

def hamming_distance(a: bytes, b: bytes) -> int:
    """Count differing bits between two byte strings."""
    return sum(bin(x ^ y).count('1') for x, y in zip(a, b))

def fuzzy_checksum(data: bytes, threshold_bits: int = 1) -> bool:
    """Horrible idea: accept data if it's 'close' to the expected checksum."""
    expected_hash = hashlib.sha256(b"Original transaction $1000").digest()
    actual_hash = hashlib.sha256(data).digest()
    distance = hamming_distance(expected_hash, actual_hash)
    # Allow up to 'threshold_bits' mismatches — this is a data integrity disaster
    return distance <= threshold_bits

# Original transaction
original = b"Original transaction $1000"
# Corrupted transaction: $1,000,000
corrupted = b"Original transaction $1000000"

# Fuzzy checksum — this will likely pass because only a few bytes changed
passed = fuzzy_checksum(corrupted, threshold_bits=5)
print(f"Fuzzy checksum passed for corrupted transaction? {passed}")

# Exact checksum — should fail
full_match = hashlib.sha256(original).digest() == hashlib.sha256(corrupted).digest()
print(f"Exact checksum match? {full_match}")

# Output shows the fuzzy checksum accepted a massive data change

// io.thecodeforge — cs-fundamentals tutorial

def xor_checksum(data: bytes) -> int:
    result = 0
    for byte in data:
        result ^= byte
    return result

def crc32_checksum(data: bytes) -> int:
    poly = 0xEDB88320
    crc = 0xFFFFFFFF
    for byte in data:
        crc ^= byte
        for _ in range(8):
            if crc & 1:
                crc = (crc >> 1) ^ poly
            else:
                crc >>= 1
    return crc ^ 0xFFFFFFFF

# Example: both detect single-bit flip
data = b"TheCodeForge"
print(xor_checksum(data))
print(crc32_checksum(data))

// io.thecodeforge — cs-fundamentals tutorial

def ones_complement_sum(words: list[int]) -> int:
    total = sum(words)
    while total > 0xFFFF:
        total = (total & 0xFFFF) + (total >> 16)
    return total

# Sender: data = 0xA9 (10101001) and 0x39 (00111001)
words = [0xA9, 0x39]
checksum = ~ones_complement_sum(words) & 0xFFFF
print(f"Transmit checksum: 0x{checksum:04X}")

# Receiver: include checksum word
received = words + [checksum]
result = ones_complement_sum(received)
ok = (result == 0xFFFF)
print(f"Receiver sum: 0x{result:04X}, data OK: {ok}")