forensics

# Forensics ## What I do I enable digital forensics capabilities including evidence acquisition, preservation, analysis, and reporting. I help with disk forensics, memory forensics, network forensics, and mobile device forensics. ## When to use me - Collecting and preserving evidence after a security incident - Analyzing compromised systems for malware and IOCs - Conducting employee investigations - Supporting legal proceedings with digital evidence - Recovering deleted files and artifacts - Timeline analysis of security events - Malware analysis and reverse engineering - Email and communication forensics ## Core Concepts - **Chain of Custody**: Documenting evidence handling and transfers - **Write Blockers**: Hardware/software to prevent evidence modification - **Hash Verification**: Ensuring evidence integrity (MD5, SHA-256) - **Live vs Dead Forensics**: Acquiring memory vs powered-off systems - **File System Forensics**: NTFS, ext4, APFS analysis - **Registry Analysis**: Windows registry artifacts - **Memory Forensics**: Volatile memory analysis (processes, connections) - **Network Forensics**: PCAP analysis, flow data, proxy logs - **Timeline Analysis**: Correlating events across sources - **Anti-Forensics Detection**: Identifying evidence tampering ## Code Examples ### Evidence Collection Manager ```python import hashlib import os import json from typing import Dict, List, Optional from dataclasses import dataclass, field from datetime import datetime from enum import Enum class EvidenceType(Enum): DISK_IMAGE = "disk_image" MEMORY_DUMP = "memory_dump" NETWORK_CAPTURE = "network_capture" LOG_FILE = "log_file" REGISTRY_HIVE = "registry_hive" FILE = "file" EMAIL = "email" class EvidenceIntegrity(Enum): VERIFIED = "verified" COMPROMISED = "compromised" UNKNOWN = "unknown" @dataclass class EvidenceItem: evidence_id: str case_number: str evidence_type: EvidenceType description: str source_path: str acquisition_path: str acquired_by: str acquired_at: datetime hash_md5: str hash_sha256: str size_bytes: int integrity_status: EvidenceIntegrity chain_of_custody: List[Dict] tags: List[str] = field(default_factory=list) notes: str = "" deleted: bool = False class EvidenceManager: def __init__(self, evidence_directory: str = "/forensics/evidence"): self.evidence_dir = evidence_directory os.makedirs(evidence_dir, exist_ok=True) self.evidence_items: Dict[str, EvidenceItem] = {} def calculate_hashes(self, file_path: str) -> Dict[str, str]: md5_hash = hashlib.md5() sha256_hash = hashlib.sha256() with open(file_path, 'rb') as f: while chunk := f.read(8192): md5_hash.update(chunk) sha256_hash.update(chunk) return { "md5": md5_hash.hexdigest(), "sha256": sha256_hash.hexdigest() } def acquire_file(self, case_number: str, source_path: str, description: str, acquired_by: str, evidence_type: EvidenceType = EvidenceType.FILE) -> EvidenceItem: file_size = os.path.getsize(source_path) hashes = self.calculate_hashes(source_path) evidence_id = f"EV-{case_number}-{datetime.now().strftime('%Y%m%d%H%M%S')}" filename = os.path.basename(source_path) acquisition_path = os.path.join(self.evidence_dir, f"{evidence_id}_{filename}") with open(source_path, 'rb') as src: with open(acquisition_path, 'wb') as dst: src.seek(0) while chunk := src.read(8192): dst.write(chunk) acquired_hashes = self.calculate_hashes(acquisition_path) if acquired_hashes["sha256"] != hashes["sha256"]: raise ValueError("Hash mismatch - evidence may be corrupted") evidence = EvidenceItem( evidence_id=evidence_id, case_number=case_number, evidence_type=evidence_type, description=description, source_path=source_path, acquisition_path=acquisition_path, acquired_by=acquired_by, acquired_at=datetime.now(), hash_md5=acquired_hashes["md5"], hash_sha256=acquired_hashes["sha256"], size_bytes=file_size, integrity_status=EvidenceIntegrity.VERIFIED, chain_of_custody=[{ "timestamp": datetime.now().isoformat(), "action": "acquired", "person": acquired_by, "location": acquisition_path }] ) self.evidence_items[evidence_id] = evidence return evidence def verify_integrity(self, evidence_id: str) -> bool: if evidence_id not in self.evidence_items: return False evidence = self.evidence_items[evidence_id] if not os.path.exists(evidence.acquisition_path): evidence.integrity_status = EvidenceIntegrity.COMPROMISED return False current_hashes = self.calculate_hashes(evidence.acquisition_path) if current_hashes["sha256"] == evidence.hash_sha256: evidence.integrity_status = EvidenceIntegrity.VERIFIED self._add_custody_entry(evidence, "integrity_verified") return True else: evidence.integrity_status = EvidenceIntegrity.COMPROMISED return False def chain_of_custody_log(self, evidence_id: str, action: str, person: str, notes: str = "") -> bool: if evidence_id in self.evidence_items: self._add_custody_entry( self.evidence_items[evidence_id], action, person, notes ) return True return False def _add_custody_entry(self, evidence: EvidenceItem, action: str, person: str = "", notes: str = ""): evidence.chain_of_custody.append({ "timestamp": datetime.now().isoformat(), "action": action, "person": person, "notes": notes, "hash_sha256": evidence.hash_sha256 }) def export_case_manifest(self, case_number: str) -> Dict: case_evidence = [ e for e in self.evidence_items.values() if e.case_number == case_number ] return { "case_number": case_number, "exported_at": datetime.now().isoformat(), "total_items": len(case_evidence), "total_size_bytes": sum(e.size_bytes for e in case_evidence), "evidence": [ { "evidence_id": e.evidence_id, "type": e.evidence_type.value, "description": e.description, "acquired_at": e.acquired_at.isoformat(), "hash_sha256": e.hash_sha256, "integrity": e.integrity_status.value, "chain_of_custody": e.chain_of_custody } for e in case_evidence ] } ``` ### Memory Forensics Analyzer ```python import re from typing import Dict, List, Optional from dataclasses import dataclass from datetime import datetime @dataclass class ProcessInfo: pid: int name: str path: str command_line: str parent_pid: int user: str created_time: datetime connections: List[Dict] loaded_modules: List[str] @dataclass class NetworkConnection: local_addr: str local_port: int remote_addr: str remote_port: int protocol: str state: str pid: int class MemoryAnalyzer: def __init__(self): self.processes: Dict[int, ProcessInfo] = {} self.connections: List[NetworkConnection] = [] self.suspicious_patterns: List[Dict] = [] def parse_process_list(self, process_output: str): lines = process_output.strip().split('\n') for line in lines[1:]: parts = line.split() if len(parts) >= 8: pid = int(parts[1]) self.processes[pid] = ProcessInfo( pid=pid, name=parts[0], path=parts[2], command_line=' '.join(parts[7:]), parent_pid=int(parts[3]) if parts[3].isdigit() else 0, user=parts[5], created_time=datetime.now(), connections=[], loaded_modules=[] ) def parse_netstat(self, netstat_output: str): lines = netstat_output.strip().split('\n') for line in lines[4:]: parts = line.split() if len(parts) >= 7: try: conn = NetworkConnection( local_addr=parts[1].rsplit(':', 1)[0], local_port=int(parts[1].rsplit(':', 1)[1]), remote_addr=parts[2].rsplit(':', 1)[0], remote_port=int(parts[2].rsplit(':', 1)[1]), protocol=parts[0], state=parts[5], pid=int(parts[6]) if len(parts) > 6 and parts[6].isdigit() else 0 ) self.connections.append(conn) if conn.pid in self.processes: self.processes[conn.pid].connections.append({ "local": f"{conn.local_addr}:{conn.local_port}", "remote": f"{conn.remote_addr}:{conn.remote_port}", "state": conn.state }) except (IndexError, ValueError): continue def detect_suspicious_connections(self) -> List[Dict]: suspicious = [] known_malicious_ips = { "1.1.1.1", "2.2.2.2", "3.3.3.3" } for conn in self.connections: if conn.remote_addr in known_malicious_ips: suspicious.append({ "type": "known_malicious_ip", "pid": conn.pid, "process": self.processes.get(conn.pid, ProcessInfo( pid=conn.pid, name="unknown", path="", command_line="", parent_pid=0, user="", created_time=datetime.now(), connections=[], loaded_modules=[] )).name, "connection": f"{conn.remote_addr}:{conn.remote_port}", "severity": "CRITICAL" }) for pid, proc in self.processes.items(): suspicious_ports = [445, 135, 139] for conn in proc.connections: port = int(conn["remote"].rsplit(':', 1)[1]) if port in suspicious_ports: suspicious.append({ "type": "lateral_movement_port", "pid": pid, "process": proc.name, "connection": conn["remote"], "severity": "HIGH" }) return suspicious def check_persistence_mechanisms(self) -> List[Dict]: persistence = [] registry_autoruns = [ r"HKLM\Software\Microsoft\Windows\CurrentVersion\Run", r"HKCU\Software\Microsoft\Windows\CurrentVersion\Run", r"HKLM\Software\Microsoft\Windows\CurrentVersion\RunOnce", ] for reg_key in registry_autoruns: persistence.append({ "type": "registry_autorun", "location": reg_key, "suspicious": True }) startup_folders = [ r"C:\Users\%USERNAME%\AppData\Roaming\Microsoft\Windows\Start Menu\Programs\Startup", r"C:\ProgramData\Microsoft\Windows\Start Menu\Programs\Startup" ] for folder in startup_folders: persistence.append({ "type": "startup_folder", "location": folder, "suspicious": False }) scheduled_tasks = [ "\\Microsoft\\Windows\\DefaultSetting", "\\Microsoft\\Windows\\Customer Experience Improvement Program" ] for task in scheduled_tasks: persistence.append({ "type": "scheduled_task", "location": task, "suspicious": False }) return persistence def detect_malware_indicators(self) -> List[Dict]: indicators = [] suspicious_names = ["malware", "virus", "trojan", "ransom", "crypt"] for pid, proc in self.processes.items(): for name in suspicious_names: if name in proc.name.lower(): indicators.append({ "type": "suspicious_process_name", "pid": pid, "name": proc.name, "path": proc.path, "confidence": "HIGH" }) suspicious_paths = [ r"\Temp\", r"\AppData\Local\Temp", r"\Windows\Temp", r"\Users\Public", r"\Perflogs" ] for pid, proc in self.processes.items(): for path in suspicious_paths: