""" Autarch Android Forensics — forensic acquisition and IOC scanning for Android devices. """ DESCRIPTION = "Android forensic acquisition and spyware IOC scanning" AUTHOR = "AUTARCH" VERSION = "1.0" CATEGORY = "counter" class AutarchAndroidModule: """Interactive Android forensics CLI menu.""" def __init__(self): from core.android_forensics import get_android_forensics_manager from core.hardware import get_hardware_manager self.mgr = get_android_forensics_manager() self.hw = get_hardware_manager() self.serial = None def _pick_device(self): devices = self.hw.adb_devices() if not devices: print(" No ADB devices connected.") return None if len(devices) == 1: self.serial = devices[0]['serial'] print(f" Selected: {self.serial}") return self.serial print("\n Select device:") for i, d in enumerate(devices, 1): model = d.get('model', '') print(f" {i}) {d['serial']} {model}") try: choice = int(input(" > ").strip()) if 1 <= choice <= len(devices): self.serial = devices[choice - 1]['serial'] return self.serial except (ValueError, EOFError, KeyboardInterrupt): pass return None def _ensure_device(self): if self.serial: return self.serial return self._pick_device() def _sev_marker(self, sev): return {'critical': '[!!!]', 'high': '[!! ]', 'medium': '[! ]', 'low': '[ ]'}.get(sev, '[? ]') def show_menu(self): status = self.hw.get_status() serial_str = self.serial or 'None selected' stats = self.mgr.get_ioc_stats() counts = stats.get('counts', {}) ioc_str = (f"{counts.get('packages', 0)} pkgs, " f"{counts.get('domains', 0)} domains, " f"{counts.get('certs', 0)} certs, " f"{counts.get('hashes', 0)} hashes") print(f"\n{'='*60}") print(" Autarch Android Forensics") print(f"{'='*60}") print(f" ADB: {'Available' if status.get('adb') else 'Not found'}") print(f" Device: {serial_str}") print(f" IOC DB: {ioc_str}") print() print(" -- Acquisition --") print(" 1) Full Acquisition (all modules)") print(" 2) Package List") print(" 3) Running Processes") print(" 4) System Properties (getprop)") print(" 5) Settings") print(" 6) Services") print(" 7) Dumpsys") print(" 8) Logcat") print(" 9) Temp Files") print() print(" -- IOC Scanning --") print(" 20) Scan Apps vs IOC Database") print(" 21) Scan APK Certificates") print(" 22) Scan Network Indicators") print(" 23) Scan File Artifacts") print(" 24) Scan Running Processes") print(" 25) Full IOC Scan (all)") print() print(" -- Reports --") print(" 30) List Past Acquisitions") print(" 31) View Acquisition Report") print() print(" -- Database --") print(" 40) IOC Database Stats") print(" 41) Reload IOC Database") print() print(" [s] Select Device | [0] Back") print() # ── Acquisition ───────────────────────────────────────────────── def do_full_acquisition(self): if not self._ensure_device(): return print(f"\n Running full acquisition on {self.serial}...") print(" This may take several minutes.") result = self.mgr.full_acquisition(self.serial) summary = result.get('scan_summary', {}) modules = result.get('modules', {}) print(f"\n Acquisition complete: {result.get('acq_id', '')}") print(f" Modules collected: {len(modules)}") print(f" IOC findings: {summary.get('total_findings', 0)}") print(f" Critical: {summary.get('critical', 0)}") print(f" High: {summary.get('high', 0)}") if not summary.get('clean', True): print("\n Findings:") for f in result.get('ioc_findings', [])[:20]: marker = self._sev_marker(f.get('severity', 'high')) print(f" {marker} [{f.get('type', '?')}] {f.get('threat_name', '?')}") detail = f.get('package') or f.get('process') or f.get('path') or f.get('remote_ip', '') if detail: print(f" {detail}") else: print(" Device appears clean.") def do_acquire_module(self, module_name): if not self._ensure_device(): return func_map = { 'packages': self.mgr.acquire_packages, 'processes': self.mgr.acquire_processes, 'getprop': self.mgr.acquire_getprop, 'settings': self.mgr.acquire_settings, 'services': self.mgr.acquire_services, 'dumpsys': self.mgr.acquire_dumpsys, 'logcat': self.mgr.acquire_logcat, 'tmp_files': self.mgr.acquire_tmp_files, } func = func_map.get(module_name) if not func: print(f" Unknown module: {module_name}") return print(f"\n Acquiring {module_name} from {self.serial}...") result = func(self.serial) if not result.get('ok'): print(f" Error: {result.get('error', 'Unknown')}") return if module_name == 'packages': pkgs = result.get('packages', []) print(f" Found {len(pkgs)} packages ({sum(1 for p in pkgs if p.get('system'))} system, " f"{sum(1 for p in pkgs if not p.get('system'))} user)") for p in pkgs[:20]: marker = "[sys]" if p.get('system') else "[usr]" print(f" {marker} {p['package']}") if len(pkgs) > 20: print(f" ... and {len(pkgs) - 20} more") elif module_name == 'getprop': props = result.get('props', {}) print(f" {len(props)} properties") for k in ('ro.product.model', 'ro.build.version.release', 'ro.build.fingerprint', 'ro.secure', 'ro.debuggable', 'net.dns1', 'net.dns2'): if k in props: print(f" {k} = {props[k]}") else: out = result.get('output', '') lines = out.splitlines() print(f" {len(lines)} lines") for line in lines[:30]: print(f" {line}") if len(lines) > 30: print(f" ... {len(lines) - 30} more lines") # ── IOC Scanning ───────────────────────────────────────────────── def _print_findings(self, result, scan_name): if not result.get('ok'): print(f" Error: {result.get('error', 'Unknown')}") return findings = result.get('findings', []) note = result.get('note', '') if note: print(f" Note: {note}") if findings: print(f"\n {scan_name}: {len(findings)} finding(s)") for f in findings: marker = self._sev_marker(f.get('severity', 'high')) name = f.get('threat_name', '?') detail = f.get('package') or f.get('process') or f.get('path') or f.get('remote_ip', '') print(f" {marker} {name}") if detail: print(f" {detail}") print(f" Source: {f.get('source', '?')}") else: scanned = result.get('total_scanned', '') scanned_str = f" ({scanned} scanned)" if scanned else "" print(f" {scan_name}: Clean{scanned_str}") def do_scan_packages(self): if not self._ensure_device(): return print(f"\n Scanning packages on {self.serial}...") self._print_findings(self.mgr.scan_packages(self.serial), "Package Scan") def do_scan_certs(self): if not self._ensure_device(): return print(f"\n Scanning APK certificates (may take a while)...") self._print_findings(self.mgr.scan_certificates(self.serial), "Certificate Scan") def do_scan_network(self): if not self._ensure_device(): return print(f"\n Scanning network indicators on {self.serial}...") self._print_findings(self.mgr.scan_network_indicators(self.serial), "Network Scan") def do_scan_files(self): if not self._ensure_device(): return print(f"\n Scanning file artifacts on {self.serial}...") self._print_findings(self.mgr.scan_file_indicators(self.serial), "File Scan") def do_scan_processes(self): if not self._ensure_device(): return print(f"\n Scanning running processes on {self.serial}...") self._print_findings(self.mgr.scan_process_indicators(self.serial), "Process Scan") def do_full_scan(self): if not self._ensure_device(): return print(f"\n Running full IOC scan on {self.serial}...") all_findings = [] for scan_name, func in [ ("Packages", self.mgr.scan_packages), ("Certs", self.mgr.scan_certificates), ("Network", self.mgr.scan_network_indicators), ("Files", self.mgr.scan_file_indicators), ("Processes", self.mgr.scan_process_indicators), ]: print(f" Scanning {scan_name}...", end='', flush=True) result = func(self.serial) found = len(result.get('findings', [])) print(f" {found} findings") all_findings.extend(result.get('findings', [])) print(f"\n Total findings: {len(all_findings)}") if all_findings: for f in all_findings: marker = self._sev_marker(f.get('severity', 'high')) detail = f.get('package') or f.get('process') or f.get('path') or f.get('remote_ip', '') print(f" {marker} {f.get('threat_name', '?')} — {detail}") else: print(" Device appears clean.") # ── Reports ────────────────────────────────────────────────────── def do_list_acquisitions(self): if not self._ensure_device(): return acqs = self.mgr.get_acquisition_list(self.serial) if not acqs: print(f" No acquisitions found for {self.serial}") return print(f"\n Acquisitions for {self.serial}:") for i, a in enumerate(acqs, 1): ts = a.get('timestamp', a['acq_id']) findings = a.get('findings', 0) modules = ', '.join(a.get('modules', [])) print(f" {i:2}) {ts} findings={findings} [{modules}]") def do_view_report(self): if not self._ensure_device(): return acqs = self.mgr.get_acquisition_list(self.serial) if not acqs: print(f" No acquisitions found for {self.serial}") return print(f"\n Select acquisition:") for i, a in enumerate(acqs, 1): print(f" {i}) {a.get('timestamp', a['acq_id'])}") try: choice = int(input(" > ").strip()) if 1 <= choice <= len(acqs): result = self.mgr.export_report(self.serial, acqs[choice - 1]['acq_id']) if result.get('ok'): report = result['report'] summary = report.get('scan_summary', {}) print(f"\n Report: {report.get('acq_id', '')}") print(f" Device: {report.get('serial', '')}") print(f" Time: {report.get('timestamp', '')}") print(f" Total findings: {summary.get('total_findings', 0)}") print(f" Critical: {summary.get('critical', 0)}, High: {summary.get('high', 0)}") for f in report.get('ioc_findings', []): marker = self._sev_marker(f.get('severity', 'high')) detail = f.get('package') or f.get('process') or f.get('path') or '' print(f" {marker} {f.get('threat_name', '?')}: {detail}") else: print(f" Error: {result.get('error')}") except (ValueError, EOFError, KeyboardInterrupt): pass # ── Database ───────────────────────────────────────────────────── def do_ioc_stats(self): stats = self.mgr.get_ioc_stats() counts = stats.get('counts', {}) sources = stats.get('sources', {}) print(f"\n IOC Database Statistics:") print(f" Generated: {stats.get('generated', 'N/A')}") print(f" Packages: {counts.get('packages', 0)}") print(f" Domains: {counts.get('domains', 0)}") print(f" Certificates: {counts.get('certs', 0)}") print(f" File paths: {counts.get('file_paths', 0)}") print(f" Process names: {counts.get('processes', 0)}") print(f" Hashes: {counts.get('hashes', 0)}") print(f" IPs: {counts.get('ips', 0)}") if sources: print(f"\n Sources:") print(f" Stalkerware YAML: {'Yes' if sources.get('stalkerware_ioc_yaml') else 'No'}") print(f" Amnesty campaigns: {sources.get('amnesty_investigations', 0)}") print(f" MVT campaigns: {sources.get('mvt_campaigns', 0)}") print(f" Citizen Lab: {sources.get('citizen_lab_campaigns', 0)}") print(f" YARA rules: {sources.get('yara_rules', 0)}") def do_reload_db(self): print(" Reloading IOC database from disk...") db = self.mgr.reload_ioc_database() print(f" Done — {len(db.get('packages', {}))} packages, " f"{len(db.get('domains', {}))} domains loaded") # ── Main Loop ───────────────────────────────────────────────────── def run_interactive(self): while True: self.show_menu() try: choice = input(" Select > ").strip() except (EOFError, KeyboardInterrupt): break if choice == '0': break actions = { '1': self.do_full_acquisition, '2': lambda: self.do_acquire_module('packages'), '3': lambda: self.do_acquire_module('processes'), '4': lambda: self.do_acquire_module('getprop'), '5': lambda: self.do_acquire_module('settings'), '6': lambda: self.do_acquire_module('services'), '7': lambda: self.do_acquire_module('dumpsys'), '8': lambda: self.do_acquire_module('logcat'), '9': lambda: self.do_acquire_module('tmp_files'), '20': self.do_scan_packages, '21': self.do_scan_certs, '22': self.do_scan_network, '23': self.do_scan_files, '24': self.do_scan_processes, '25': self.do_full_scan, '30': self.do_list_acquisitions, '31': self.do_view_report, '40': self.do_ioc_stats, '41': self.do_reload_db, 's': self._pick_device, } action = actions.get(choice) if action: action() else: print(" Invalid choice.") def run(): m = AutarchAndroidModule() m.run_interactive()