373 lines
16 KiB
Python
373 lines
16 KiB
Python
|
"""
|
||
|
|
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()
|