forensics_mode.py

"""
Forensic mode for evidence preservation.

Provides:
- Read-only operations enforcement
- File hash calculation and verification
- Audit trail creation
- Evidence integrity verification
"""

import hashlib
import json
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Set


@dataclass
class FileHash:
    """File hash information."""
    file_path: Path
    sha256: str
    md5: str
    size: int
    accessed_at: datetime
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        return {
            "file": str(self.file_path),
            "sha256": self.sha256,
            "md5": self.md5,
            "size": self.size,
            "accessed": self.accessed_at.isoformat(),
        }


@dataclass
class AuditEntry:
    """Audit trail entry."""
    timestamp: datetime
    operation: str
    file_path: Optional[Path]
    details: Dict
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        return {
            "timestamp": self.timestamp.isoformat(),
            "operation": self.operation,
            "file": str(self.file_path) if self.file_path else None,
            "details": self.details,
        }


@dataclass
class ForensicSession:
    """Forensic analysis session information."""
    session_id: str
    start_time: datetime
    end_time: Optional[datetime] = None
    jenkins_home: Optional[Path] = None
    files_accessed: Set[Path] = field(default_factory=set)
    files_hashed: List[FileHash] = field(default_factory=list)
    audit_trail: List[AuditEntry] = field(default_factory=list)
    read_only_mode: bool = True
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        return {
            "session_id": self.session_id,
            "start_time": self.start_time.isoformat(),
            "end_time": self.end_time.isoformat() if self.end_time else None,
            "jenkins_home": str(self.jenkins_home) if self.jenkins_home else None,
            "files_accessed_count": len(self.files_accessed),
            "files_hashed_count": len(self.files_hashed),
            "audit_entries": len(self.audit_trail),
            "read_only": self.read_only_mode,
        }


class ForensicMode:
    """
    Forensic mode manager for evidence preservation.
    
    Ensures read-only operations and creates audit trails.
    """
    
    def __init__(
        self, 
        jenkins_home: Path,
        session_id: Optional[str] = None,
        read_only: bool = True
    ):
        """
        Initialize forensic mode.
        
        Args:
            jenkins_home: Path to Jenkins home directory
            session_id: Session identifier (auto-generated if None)
            read_only: Enforce read-only operations
        """
        self.jenkins_home = Path(jenkins_home)
        self.read_only = read_only
        
        if session_id is None:
            session_id = f"forensic_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
        
        self.session = ForensicSession(
            session_id=session_id,
            start_time=datetime.now(),
            jenkins_home=self.jenkins_home,
            read_only_mode=read_only,
        )
    
    def calculate_file_hash(self, file_path: Path) -> Optional[FileHash]:
        """
        Calculate hashes for a file.
        
        Args:
            file_path: Path to file
            
        Returns:
            File hash information
        """
        if not file_path.exists():
            return None
        
        try:
            sha256_hash = hashlib.sha256()
            md5_hash = hashlib.md5()
            
            with open(file_path, 'rb') as f:
                while chunk := f.read(8192):
                    sha256_hash.update(chunk)
                    md5_hash.update(chunk)
            
            file_hash = FileHash(
                file_path=file_path,
                sha256=sha256_hash.hexdigest(),
                md5=md5_hash.hexdigest(),
                size=file_path.stat().st_size,
                accessed_at=datetime.now(),
            )
            
            self.session.files_hashed.append(file_hash)
            self.log_audit("hash_calculated", file_path, {
                "sha256": file_hash.sha256,
                "size": file_hash.size
            })
            
            return file_hash
        
        except (IOError, OSError) as e:
            self.log_audit("hash_failed", file_path, {"error": str(e)})
            return None
    
    def read_file(self, file_path: Path) -> Optional[bytes]:
        """
        Read a file in forensic mode.
        
        Args:
            file_path: Path to file
            
        Returns:
            File content or None
        """
        if not file_path.exists():
            self.log_audit("read_failed", file_path, {"reason": "file_not_found"})
            return None
        
        try:
            # Calculate hash before reading
            file_hash = self.calculate_file_hash(file_path)
            
            # Read file
            content = file_path.read_bytes()
            
            # Track access
            self.session.files_accessed.add(file_path)
            self.log_audit("file_read", file_path, {
                "size": len(content),
                "hash": file_hash.sha256 if file_hash else None
            })
            
            return content
        
        except (IOError, OSError) as e:
            self.log_audit("read_failed", file_path, {"error": str(e)})
            return None
    
    def verify_no_modifications(self) -> Dict[Path, bool]:
        """
        Verify that no files were modified during session.
        
        Returns:
            Dictionary mapping file paths to verification status
        """
        verification_results = {}
        
        for file_hash in self.session.files_hashed:
            current_hash = self.calculate_file_hash(file_hash.file_path)
            
            if current_hash is None:
                verification_results[file_hash.file_path] = False
            else:
                # Compare SHA256 hashes
                verification_results[file_hash.file_path] = (
                    current_hash.sha256 == file_hash.sha256
                )
        
        return verification_results
    
    def log_audit(
        self, 
        operation: str, 
        file_path: Optional[Path] = None,
        details: Optional[Dict] = None
    ) -> None:
        """
        Log an audit trail entry.
        
        Args:
            operation: Operation type
            file_path: Optional file path
            details: Optional additional details
        """
        entry = AuditEntry(
            timestamp=datetime.now(),
            operation=operation,
            file_path=file_path,
            details=details or {},
        )
        
        self.session.audit_trail.append(entry)
    
    def end_session(self) -> None:
        """End the forensic session."""
        self.session.end_time = datetime.now()
        self.log_audit("session_ended", details={
            "duration_seconds": (
                self.session.end_time - self.session.start_time
            ).total_seconds()
        })
    
    def export_audit_trail(self, output_path: Path) -> bool:
        """
        Export audit trail to JSON file.
        
        Args:
            output_path: Path to output file
            
        Returns:
            True if successful
        """
        try:
            audit_data = {
                "session": self.session.to_dict(),
                "files_accessed": [str(f) for f in self.session.files_accessed],
                "file_hashes": [h.to_dict() for h in self.session.files_hashed],
                "audit_trail": [e.to_dict() for e in self.session.audit_trail],
            }
            
            output_path.write_text(json.dumps(audit_data, indent=2))
            
            self.log_audit("audit_exported", output_path, {
                "entries": len(self.session.audit_trail)
            })
            
            return True
        
        except (IOError, OSError) as e:
            self.log_audit("audit_export_failed", output_path, {"error": str(e)})
            return False
    
    def generate_integrity_report(self) -> Dict:
        """
        Generate an integrity report.
        
        Returns:
            Integrity report dictionary
        """
        verification = self.verify_no_modifications()
        
        modified_files = [
            str(path) for path, verified in verification.items() 
            if not verified
        ]
        
        return {
            "session_id": self.session.session_id,
            "read_only_mode": self.read_only,
            "files_verified": len(verification),
            "files_modified": len(modified_files),
            "modified_files": modified_files,
            "integrity_maintained": len(modified_files) == 0,
            "start_time": self.session.start_time.isoformat(),
            "end_time": self.session.end_time.isoformat() if self.session.end_time else None,
        }
    
    def __enter__(self):
        """Enter context."""
        self.log_audit("session_started", details={
            "jenkins_home": str(self.jenkins_home),
            "read_only": self.read_only
        })
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        """Exit context."""
        self.end_session()
        
        if exc_type is not None:
            self.log_audit("session_error", details={
                "exception": str(exc_type),
                "message": str(exc_val)
            })
        
        return False


class ChainOfCustody:
    """
    Chain of custody tracker for forensic evidence.
    """
    
    def __init__(self):
        """Initialize chain of custody."""
        self.entries: List[Dict] = []
    
    def add_entry(
        self, 
        action: str, 
        operator: str,
        file_path: Optional[Path] = None,
        notes: Optional[str] = None
    ) -> None:
        """
        Add a chain of custody entry.
        
        Args:
            action: Action performed
            operator: Person/system performing action
            file_path: Optional file path
            notes: Optional notes
        """
        entry = {
            "timestamp": datetime.now().isoformat(),
            "action": action,
            "operator": operator,
            "file": str(file_path) if file_path else None,
            "notes": notes,
        }
        
        self.entries.append(entry)
    
    def export(self, output_path: Path) -> bool:
        """
        Export chain of custody to file.
        
        Args:
            output_path: Output file path
            
        Returns:
            True if successful
        """
        try:
            output_path.write_text(json.dumps({
                "chain_of_custody": self.entries,
                "entry_count": len(self.entries)
            }, indent=2))
            return True
        except (IOError, OSError):
            return False