"""File hashing utilities for memory dump integrity verification"""
import hashlib
import asyncio
from pathlib import Path
from typing import Dict, List
from database import ForensicsDatabase
async def calculate_hashes(file_path: Path,
algorithms: List[str] = None) -> Dict[str, str]:
"""
Calculate multiple hashes for a file asynchronously
Args:
file_path: Path to file to hash
algorithms: List of hash algorithms (default: ['md5', 'sha1', 'sha256'])
Returns:
Dictionary of algorithm -> hexdigest mapping
Example:
{'md5': 'd41d8cd...', 'sha1': 'da39a3e...', 'sha256': 'e3b0c44...'}
"""
if algorithms is None:
algorithms = ['md5', 'sha1', 'sha256']
# Create hashers for each algorithm
hashers = {alg: hashlib.new(alg) for alg in algorithms}
# Use 8MB chunks for performance with large memory dumps
chunk_size = 8 * 1024 * 1024
# Run hash calculation in executor to avoid blocking
def _calculate():
with open(file_path, 'rb') as f:
while True:
chunk = f.read(chunk_size)
if not chunk:
break
for hasher in hashers.values():
hasher.update(chunk)
return {alg: hasher.hexdigest() for alg, hasher in hashers.items()}
# Run in thread pool to avoid blocking event loop
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(None, _calculate)
return result
async def get_or_calculate_hashes(db: ForensicsDatabase, dump_id: str,
dump_path: Path) -> Dict[str, str]:
"""
Get cached hashes or calculate if not available
Args:
db: Database instance
dump_id: Dump identifier
dump_path: Path to dump file
Returns:
Dictionary of hashes
"""
# Check database cache first
cached_hashes = await db.get_dump_hashes(dump_id)
if cached_hashes:
return cached_hashes
# Calculate hashes
hashes = await calculate_hashes(dump_path)
# Store in database
await db.store_dump_hashes(dump_id, hashes)
return hashes
def format_hashes(hashes: Dict[str, str]) -> str:
"""
Format hashes for display
Args:
hashes: Dictionary of algorithm -> digest mapping
Returns:
Formatted markdown string
"""
result = "**File Hashes:**\n"
if 'md5' in hashes:
result += f"- MD5: {hashes['md5']}\n"
if 'sha1' in hashes:
result += f"- SHA1: {hashes['sha1']}\n"
if 'sha256' in hashes:
result += f"- SHA256: {hashes['sha256']}\n"
return result
