368 lines
13 KiB
Python
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)})
|