Files
autarch/web/routes/android_forensics.py

368 lines
13 KiB
Python

"""Autarch Android Forensics routes — forensic acquisition and IOC scanning."""
from datetime import datetime
from flask import Blueprint, render_template, request, jsonify
from web.auth import login_required
android_forensics_bp = Blueprint('android_forensics', __name__, url_prefix='/android-forensics')
def _mgr():
from core.android_forensics import get_android_forensics_manager
return get_android_forensics_manager()
def _body():
return request.get_json(silent=True) or {}
def _serial():
data = _body()
serial = data.get('serial') or request.form.get('serial') or request.args.get('serial', '')
return str(serial).strip() if serial else ''
# ── Main Page ────────────────────────────────────────────────────────
@android_forensics_bp.route('/')
@login_required
def index():
from core.hardware import get_hardware_manager
hw = get_hardware_manager()
status = hw.get_status()
ioc_stats = _mgr().get_ioc_stats()
return render_template('android_forensics.html', status=status, ioc_stats=ioc_stats)
# ── Acquisition Routes ───────────────────────────────────────────────
@android_forensics_bp.route('/acquire/<module>', methods=['POST'])
@login_required
def acquire(module):
serial = _serial()
if not serial:
return jsonify({'error': 'No serial provided'})
func_map = {
'packages': _mgr().acquire_packages,
'processes': _mgr().acquire_processes,
'getprop': _mgr().acquire_getprop,
'settings': _mgr().acquire_settings,
'services': _mgr().acquire_services,
'dumpsys': _mgr().acquire_dumpsys,
'logcat': _mgr().acquire_logcat,
'tmp_files': _mgr().acquire_tmp_files,
}
func = func_map.get(module)
if not func:
return jsonify({'error': f'Unknown module: {module}'})
return jsonify(func(serial))
@android_forensics_bp.route('/acquire/full', methods=['POST'])
@login_required
def acquire_full():
serial = _serial()
if not serial:
return jsonify({'error': 'No serial provided'})
return jsonify(_mgr().full_acquisition(serial))
# ── IOC Scan Routes ──────────────────────────────────────────────────
@android_forensics_bp.route('/scan/apps', methods=['POST'])
@login_required
def scan_apps():
serial = _serial()
if not serial:
return jsonify({'error': 'No serial provided'})
return jsonify(_mgr().scan_packages(serial))
@android_forensics_bp.route('/scan/certs', methods=['POST'])
@login_required
def scan_certs():
serial = _serial()
if not serial:
return jsonify({'error': 'No serial provided'})
return jsonify(_mgr().scan_certificates(serial))
@android_forensics_bp.route('/scan/network', methods=['POST'])
@login_required
def scan_network():
serial = _serial()
if not serial:
return jsonify({'error': 'No serial provided'})
return jsonify(_mgr().scan_network_indicators(serial))
@android_forensics_bp.route('/scan/files', methods=['POST'])
@login_required
def scan_files():
serial = _serial()
if not serial:
return jsonify({'error': 'No serial provided'})
return jsonify(_mgr().scan_file_indicators(serial))
@android_forensics_bp.route('/scan/processes', methods=['POST'])
@login_required
def scan_processes():
serial = _serial()
if not serial:
return jsonify({'error': 'No serial provided'})
return jsonify(_mgr().scan_process_indicators(serial))
@android_forensics_bp.route('/scan/full', methods=['POST'])
@login_required
def scan_full():
serial = _serial()
if not serial:
return jsonify({'error': 'No serial provided'})
return jsonify(_mgr().full_acquisition(serial))
# ── Report Routes ────────────────────────────────────────────────────
@android_forensics_bp.route('/report/list', methods=['POST'])
@login_required
def report_list():
serial = _serial()
if not serial:
return jsonify({'error': 'No serial provided'})
return jsonify({'ok': True, 'acquisitions': _mgr().get_acquisition_list(serial)})
@android_forensics_bp.route('/report/view', methods=['POST'])
@login_required
def report_view():
data = _body()
serial = data.get('serial', '').strip()
acq_id = data.get('acq_id', '').strip()
if not serial or not acq_id:
return jsonify({'error': 'serial and acq_id required'})
return jsonify(_mgr().export_report(serial, acq_id))
# ── Database Routes ──────────────────────────────────────────────────
@android_forensics_bp.route('/db/stats', methods=['GET'])
@login_required
def db_stats():
return jsonify(_mgr().get_ioc_stats())
@android_forensics_bp.route('/db/reload', methods=['POST'])
@login_required
def db_reload():
db = _mgr().reload_ioc_database()
return jsonify({
'ok': True,
'counts': {k: len(v) for k, v in db.items() if isinstance(v, (dict, list))},
})
@android_forensics_bp.route('/ioc/export', methods=['GET'])
@login_required
def ioc_export():
"""Compact IOC export for companion app caching."""
db = _mgr().load_ioc_database()
return jsonify({
'packages': list(db.get('packages', {}).keys()),
'domains': list(db.get('domains', {}).keys()),
'hashes': list(db.get('hashes', {}).keys()),
'ips': list(db.get('ips', [])),
'generated': datetime.now().isoformat(),
'counts': {
'packages': len(db.get('packages', {})),
'domains': len(db.get('domains', {})),
'hashes': len(db.get('hashes', {})),
'ips': len(db.get('ips', [])),
},
})
# ── Direct (WebUSB) Mode ─────────────────────────────────────────────
# Op keys match the JS URL→op transformation:
# url.replace('/android-forensics/', '').replace(/\//g, '_')
# So /acquire/packages → acquire_packages, /scan/apps → scan_apps, etc.
_DIRECT_COMMANDS = {
'acquire_packages': {
'packages': 'pm list packages -f',
'packages_sys': 'pm list packages -s',
},
'acquire_processes': {
'processes': 'ps -A',
},
'acquire_getprop': {
'getprop': 'getprop',
},
'acquire_settings': {
'settings_global': 'settings list global',
'settings_secure': 'settings list secure',
'settings_system': 'settings list system',
},
'acquire_services': {
'services': 'service list',
},
'acquire_logcat': {
'logcat': 'logcat -d -t 1000',
},
'acquire_tmp_files': {
'tmp_files': 'ls -la /data/local/tmp/ 2>/dev/null',
},
'scan_apps': {
'packages': 'pm list packages -f',
'packages_sys': 'pm list packages -s',
},
'scan_network': {
'tcp_connections': 'cat /proc/net/tcp 2>/dev/null',
'dns1': 'getprop net.dns1',
'dns2': 'getprop net.dns2',
},
'scan_processes': {
'processes': 'ps -A',
},
'scan_files': {
# file artifact scanning requires server-side IOC DB; not supported in direct mode
},
}
@android_forensics_bp.route('/cmd', methods=['POST'])
@login_required
def direct_cmd():
"""Return ADB shell commands for Direct (WebUSB) mode."""
data = _body()
op = data.get('op', '').replace('/', '_').replace('-', '_')
cmds = _DIRECT_COMMANDS.get(op)
if cmds:
return jsonify({'commands': cmds, 'supported': True})
return jsonify({'commands': {}, 'supported': False,
'reason': 'This operation requires Server mode.'})
@android_forensics_bp.route('/parse', methods=['POST'])
@login_required
def direct_parse():
"""Parse raw ADB output from Direct (WebUSB) mode."""
import re
data = _body()
op = data.get('op', '').replace('/', '_').replace('-', '_')
raw = data.get('raw', {})
mgr = _mgr()
def _parse_packages(pkg_output, sys_output=''):
sys_pkgs = set()
for line in (sys_output or '').splitlines():
line = line.strip()
if line.startswith('package:'):
sys_pkgs.add(line[8:].strip())
pkgs = []
for line in (pkg_output or '').splitlines():
line = line.strip()
if not line.startswith('package:'):
continue
line = line[8:]
eq_pos = line.rfind('=')
if eq_pos == -1:
pkg_name, apk_path = line, ''
else:
apk_path, pkg_name = line[:eq_pos], line[eq_pos + 1:]
pkg_name = pkg_name.strip()
if pkg_name:
pkgs.append({'package': pkg_name, 'path': apk_path.strip(),
'system': pkg_name in sys_pkgs})
return pkgs
try:
if op in ('acquire_packages',):
pkgs = _parse_packages(raw.get('packages', ''), raw.get('packages_sys', ''))
return jsonify({'ok': True, 'packages': pkgs, 'total': len(pkgs)})
elif op == 'scan_apps':
pkgs = _parse_packages(raw.get('packages', ''), raw.get('packages_sys', ''))
return jsonify(mgr.scan_packages(None, packages_data=pkgs))
elif op == 'acquire_processes':
lines = [l for l in (raw.get('processes', '') or '').splitlines() if l.strip()]
return jsonify({'ok': True, 'output': raw.get('processes', ''),
'count': max(0, len(lines) - 1)})
elif op == 'acquire_getprop':
props = {}
for line in (raw.get('getprop', '') or '').splitlines():
m = re.match(r'\[(.+)\]:\s*\[(.*)\]\s*$', line)
if m:
props[m.group(1)] = m.group(2)
return jsonify({'ok': True, 'props': props})
elif op == 'acquire_settings':
result = {}
for ns in ('global', 'secure', 'system'):
settings = {}
for line in (raw.get(f'settings_{ns}', '') or '').splitlines():
if '=' in line:
k, _, v = line.partition('=')
settings[k.strip()] = v.strip()
result[ns] = settings
return jsonify({'ok': True, 'settings': result})
elif op in ('acquire_services', 'acquire_logcat', 'acquire_tmp_files'):
key = op.replace('acquire_', '')
out = raw.get(key, '')
return jsonify({'ok': True, 'output': out, 'lines': len((out or '').splitlines())})
elif op == 'scan_network':
db = mgr.load_ioc_database()
findings = []
ioc_ips = set(db.get('ips', []))
# Check DNS properties
for dns_key in ('dns1', 'dns2'):
val = (raw.get(dns_key, '') or '').strip().lower()
if val and val in db.get('domains', {}):
match = db['domains'][val]
findings.append({'type': 'dns_property', 'key': dns_key, 'value': val,
'threat_name': match['name'], 'source': match['source'],
'severity': 'high'})
# Check TCP connections
if ioc_ips:
for line in (raw.get('tcp_connections', '') or '').splitlines()[1:]:
parts = line.strip().split()
if len(parts) < 3:
continue
try:
remote_hex = parts[2]
ip_hex, port_hex = remote_hex.split(':')
ip = '.'.join(str(int(ip_hex[i:i+2], 16)) for i in (6, 4, 2, 0))
port = int(port_hex, 16)
if ip not in ('0.0.0.0', '127.0.0.1') and ip in ioc_ips:
findings.append({'type': 'tcp_connection', 'remote_ip': ip,
'remote_port': port, 'severity': 'critical',
'threat_name': 'Known malicious IP',
'source': 'network_scan'})
except Exception:
pass
return jsonify({'ok': True, 'findings': findings, 'clean': len(findings) == 0})
elif op == 'scan_processes':
db = mgr.load_ioc_database()
proc_output = (raw.get('processes', '') or '').lower()
findings = []
for proc_name, match in db.get('processes', {}).items():
if proc_name and proc_name in proc_output:
findings.append({'type': 'process_match', 'process': proc_name,
'threat_name': match['name'], 'source': match['source'],
'severity': 'critical'})
return jsonify({'ok': True, 'findings': findings, 'clean': len(findings) == 0})
else:
key = op.replace('acquire_', '')
out = raw.get(key, '') if isinstance(raw, dict) else ''
return jsonify({'ok': True, 'output': out})
except Exception as exc:
return jsonify({'error': str(exc)})