secure_wipe_auditor.py

#!/usr/bin/env python3
"""
Secure Wipe Auditor - Zero Trust Pre-Wipe Check with Immutable Audit Trail
Implements Merkle Tree-based audit logging with Zero Trust Attestation (ZTA)
"""

import hashlib
import json
import time
import os
import sys
import psutil
from typing import Tuple, List, Dict

# Configuration
AUDIT_LOG_FILE = 'audit_chain_log.txt'
MOCK_FORENSIC_TOOLS = ['wireshark', 'gdb', 'volatility']

# Set UTF-8 encoding for Windows console
if sys.platform.startswith('win'):
    import codecs
    sys.stdout = codecs.getwriter('utf-8')(sys.stdout.detach())
    sys.stderr = codecs.getwriter('utf-8')(sys.stderr.detach())

def calculate_hash(data: bytes) -> str:
    """Calculate SHA-256 hash of data"""
    return hashlib.sha256(data).hexdigest()

def mock_wipe_disk(device_id: str, status: str = 'SUCCESS') -> Tuple[bool, str]:
    """
    Mock wipe function that simulates disk wiping operation
    
    Args:
        device_id: Device identifier string
        status: SUCCESS or FAILURE to simulate different outcomes
    
    Returns:
        Tuple of (success_bool, verification_hash)
    """
    if status == 'SUCCESS':
        return (True, '0xDEADBEEF_MOCK_HASH')
    else:
        return (False, '0xBADF00D_MOCK_FAILURE')

def perform_zta_checks(device_id: str) -> Dict:
    """
    Perform Zero Trust Attestation checks
    
    Args:
        device_id: Device identifier to check
        
    Returns:
        Dictionary containing ZTA check results
    """
    print("🔒 Performing Zero Trust Attestation (ZTA) Checks...")
    
    # Operator ID Check
    operator_id = input("Enter Operator ID: ").strip()
    
    # Process Check - Look for forensic tools
    forensic_tool_check = False
    running_processes = []
    
    for proc in psutil.process_iter(['name']):
        try:
            proc_name = proc.info['name'].lower()
            running_processes.append(proc_name)
            
            for tool in MOCK_FORENSIC_TOOLS:
                if tool.lower() in proc_name:
                    forensic_tool_check = True
                    print(f"⚠️  FORENSIC TOOL DETECTED: {proc_name}")
                    break
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue
    
    # Mock Policy Check - Simple rule for demonstration
    policy_check = True
    if 'SSD' in device_id.upper():
        policy_check = True
        print("✅ Policy Check: SSD device approved for wiping")
    else:
        policy_check = False
        print("❌ Policy Check: Non-SSD device requires additional approval")
    
    zta_results = {
        'operator_id': operator_id,
        'forensic_tool_check': forensic_tool_check,
        'policy_check': policy_check,
        'timestamp': time.time(),
        'device_id': device_id
    }
    
    print(f"👤 Operator ID: {operator_id}")
    print(f"🔍 Forensic Tool Check: {'FAIL' if forensic_tool_check else 'PASS'}")
    print(f"📋 Policy Check: {'PASS' if policy_check else 'FAIL'}")
    
    return zta_results

def load_audit_chain(filename: str) -> List[str]:
    """
    Load previous root hashes from audit log file
    
    Args:
        filename: Path to audit log file
        
    Returns:
        List of previous hash entries
    """
    if not os.path.exists(filename):
        return []
    
    try:
        with open(filename, 'r') as f:
            hashes = [line.strip() for line in f.readlines() if line.strip()]
        return hashes
    except Exception as e:
        print(f"⚠️  Error loading audit chain: {e}")
        return []

def compute_merkle_root(leaf_hashes: List[str]) -> str:
    """
    Compute Merkle Tree root from list of leaf hashes
    
    Args:
        leaf_hashes: List of hex SHA-256 hashes
        
    Returns:
        Merkle root hash as hex string
    """
    if not leaf_hashes:
        return calculate_hash(b"EMPTY_TREE")
    
    if len(leaf_hashes) == 1:
        return leaf_hashes[0]
    
    current_level = leaf_hashes.copy()
    
    while len(current_level) > 1:
        next_level = []
        
        # Process pairs
        for i in range(0, len(current_level), 2):
            left = current_level[i]
            
            # If odd number of hashes, duplicate the last one
            if i + 1 < len(current_level):
                right = current_level[i + 1]
            else:
                right = current_level[i]  # Duplicate last hash
            
            # Hash the concatenation
            combined = (left + right).encode('utf-8')
            parent_hash = calculate_hash(combined)
            next_level.append(parent_hash)
        
        current_level = next_level
    
    return current_level[0]

def generate_audit_proof(metadata: Dict, old_root: str, audit_hash: str) -> Tuple[str, str]:
    """
    Generate audit proof with new Merkle root
    
    Args:
        metadata: Audit metadata dictionary
        old_root: Previous Merkle root
        audit_hash: Current audit hash
        
    Returns:
        Tuple of (new_audit_hash, new_merkle_root)
    """
    # Convert metadata to canonical JSON
    canonical_json = json.dumps(metadata, sort_keys=True, separators=(',', ':'))
    
    # Calculate audit hash of canonical JSON
    new_audit_hash = calculate_hash(canonical_json.encode('utf-8'))
    
    # Load previous audit chain
    previous_hashes = load_audit_chain(AUDIT_LOG_FILE)
    
    # Add new audit hash to the chain
    all_hashes = previous_hashes + [new_audit_hash]
    
    # Compute new Merkle root
    new_merkle_root = compute_merkle_root(all_hashes)
    
    return new_audit_hash, new_merkle_root

def main():
    """Main execution function"""
    print("🚀 Secure Wipe Auditor - Zero Trust Edition")
    print("=" * 50)
    
    # Check for mock mode flag
    MOCK_MODE = '--mock' in sys.argv
    if MOCK_MODE:
        print("🧪 Running in MOCK MODE for testing")
    
    # Define target device
    device_id = 'sda1_SSD_SN12345'
    print(f"🎯 Target Device: {device_id}")
    
    # Phase 1: Zero Trust Attestation
    print(f"\n📋 Phase 1: Zero Trust Attestation")
    print("-" * 30)
    
    zta_results = perform_zta_checks(device_id)
    
    # Zero Trust Enforcement - Exit if forensic tools detected
    if zta_results['forensic_tool_check']:
        print(f"\n❌ ZERO TRUST VIOLATION: Forensic tools detected!")
        print(f"🚫 Wipe operation DENIED for security reasons")
        print(f"🔒 System must be clean before proceeding")
        sys.exit(1)
    
    print(f"\n✅ Zero Trust Attestation PASSED")
    
    # Wipe Operation (Mock)
    print(f"\n💽 Phase 2: Secure Wipe Operation")
    print("-" * 30)
    
    print(f"🔄 Initiating secure wipe of {device_id}...")
    wipe_success, final_verification_hash = mock_wipe_disk(device_id, 'SUCCESS')
    
    if wipe_success:
        print(f"✅ Wipe operation completed successfully")
        print(f"🔐 Verification Hash: {final_verification_hash}")
    else:
        print(f"❌ Wipe operation failed")
        print(f"💥 Error Hash: {final_verification_hash}")
    
    # Phase 3: Audit Chain Generation
    print(f"\n🔗 Phase 3: Immutable Audit Trail")
    print("-" * 30)
    
    # Compile final audit metadata
    audit_metadata = {
        'device_id': device_id,
        'zta_results': zta_results,
        'wipe_status': 'SUCCESS' if wipe_success else 'FAILURE',
        'final_verification_hash': final_verification_hash,
        'timestamp': time.time(),
        'audit_version': '1.0',
        'operator_attestation': True
    }
    
    # Load previous audit chain
    previous_chain = load_audit_chain(AUDIT_LOG_FILE)
    old_root = previous_chain[-1] if previous_chain else "GENESIS_BLOCK"
    
    print(f"📚 Previous audit chain entries: {len(previous_chain)}")
    print(f"🌳 Previous Merkle Root: {old_root}")
    
    # Generate new audit proof
    new_audit_hash, new_merkle_root = generate_audit_proof(
        audit_metadata, old_root, ""
    )
    
    # Display results
    print(f"\n🎯 AUDIT TRAIL RESULTS")
    print("=" * 30)
    print(f"📝 New Audit Hash: {new_audit_hash}")
    print(f"🌳 New Merkle Root: {new_merkle_root}")
    print(f"🔗 Chain Length: {len(previous_chain) + 1}")
    
    # Write new Merkle root to audit log
    try:
        with open(AUDIT_LOG_FILE, 'a') as f:
            f.write(f"{new_audit_hash}\n")
        print(f"✅ Audit chain updated: {AUDIT_LOG_FILE}")
    except Exception as e:
        print(f"❌ Error writing audit log: {e}")
    
    # Final summary
    print(f"\n🎉 SECURE WIPE AUDIT COMPLETE")
    print("=" * 30)
    print(f"Device: {device_id}")
    print(f"Status: {'SUCCESS' if wipe_success else 'FAILURE'}")
    print(f"Operator: {zta_results['operator_id']}")
    print(f"Audit Hash: {new_audit_hash[:16]}...")
    print(f"Merkle Root: {new_merkle_root[:16]}...")
    print(f"Zero Trust: ENFORCED")
    print(f"Immutability: GUARANTEED")

if __name__ == "__main__":
    main()