1# SPDX-FileCopyrightText: 2026 Defensive Lab Agency
2# SPDX-FileContributor: u039b <git@0x39b.fr>
3#
4# SPDX-License-Identifier: GPL-3.0-or-later
5
6import logging
7from functools import cached_property
8from importlib import resources
9from pathlib import Path
10from tempfile import NamedTemporaryFile
11from typing import Dict, Any, Optional
12
13import frida
14from frida.core import Device
15from ppadb.client import Client as AdbClient
16from ppadb.device import Device as PPDevice
17
18from octopus.frida.server import FridaServer
19
20logger = logging.getLogger(__name__)
21
22
[docs]
23class AndroidDevice:
24 """
25 Represents an Android device and provides methods for device management.
26
27 This class encapsulates operations such as rooting, property retrieval,
28 and Frida server management for an Android device.
29
30 Attributes:
31 adb: Instance of :class:`~octopus.android.adb.ADB` used for device communication.
32 is_root: Indicates if the device is running as root.
33 requires_su: Indicates if :textmono:`su` access is required for root operations.
34 rooted: Indicates if the device is rooted or has `su` access.
35 adb_device: The connected ADB device instance.
36 """
37
38 device_tmp_dir = Path("/data/local/tmp/")
39
[docs]
40 def __init__(self, adb_device: PPDevice):
41 """
42 Initializes the AndroidDevice instance.
43
44 Connects to the device using the provided ADB instance, attempts to
45 root the device, checks root status, and verifies Frida server installation.
46
47 Args:
48 adb_device: An instance of :class:`~ppadb.device.PPDevice` for device communication.
49 """
50 self.adb_device = adb_device
51 self.is_root = False
52 self.requires_su = False
53 self.rooted = False
54 self.tcpdump_binaries_dir = resources.files("octopus") / "assets" / "tcpdump_binaries"
55 self.tcpdump_path = self.device_tmp_dir / "tcpdump"
56 self.init()
57 self.check_frida_server_installed()
58
[docs]
59 def is_ready(self) -> bool:
60 logger.info("Checking the device configuration")
61 if self.am_i_root():
62 logger.info("The device is rooted")
63 return True
64 if self.check_device_is_debuggable() and not self.am_i_root():
65 self.root()
66 self.is_root = self.am_i_root()
67 logger.info("The device is rooted")
68 return self.is_root
69 if self.check_su_is_available():
70 self.requires_su = True
71 logger.info("The device is not rooted but 'su' is available")
72 return self.requires_su
73 return False
74
[docs]
75 def init(self):
76 if not self.check_adb_is_available():
77 raise Exception("ADB is not ready, make sure ADB server is running.")
78 if not self.is_ready():
79 raise Exception("The device is not usable because it is not rooted or 'su' command is not available.")
80
[docs]
81 def check_adb_is_available(self) -> bool:
82 """Checks if the ADB connection to the device is available.
83
84 Attempts to run a simple shell command on the device to verify
85 that ADB communication is functional.
86
87 Returns:
88 True if ADB is available, False if a :exc:`RuntimeError`
89 is raised during the shell command.
90 """
91 try:
92 self.adb_device.shell("whoami")
93 except RuntimeError:
94 return False
95 return True
96
[docs]
97 def check_device_is_debuggable(self) -> bool:
98 """Checks if the Android device is debuggable.
99
100 Retrieves the ``ro.debuggable`` system property and checks
101 whether it is set to ``1``.
102
103 Returns:
104 True if the device is debuggable, False otherwise.
105 """
106 return "1" in self.get_property("ro.debuggable")
107
[docs]
108 def check_su_is_available(self) -> bool:
109 """Checks if ``su`` is available on the device.
110
111 Runs a simple command via ``su`` and checks if it succeeds.
112
113 Returns:
114 True if ``su`` is available, False otherwise.
115 """
116 return "1" in self.adb_device.shell('su -c "echo 1"').strip()
117
[docs]
118 def am_i_root(self) -> bool:
119 """Checks if the current ADB shell user is root.
120
121 Runs the :textmono:`whoami` command via ADB and checks if the
122 output contains ``root``.
123
124 Returns:
125 True if the current user is root, False otherwise.
126 """
127 return "root" in self.adb_device.shell("whoami")
128
[docs]
129 def root(self):
130 """
131 Attempts to root the Android device using ADB.
132
133 Raises:
134 RuntimeError: If root access is disabled on the device.
135
136 Uses the internal ADB service to request root access. If the device is not already running as root,
137 attempts to reconnect.
138 """
139 logger.info("Rooting the device")
140 try:
141 self.adb_device.root()
142 except RuntimeError as e:
143 if "adbd is already running as root" in str(e):
144 return
145 # elif not self.is_rooted():
146 # raise Exception("Root access is disabled on the device.")
147 else:
148 raise e # ToDo why we fall here even if "su" exists?!
149
[docs]
150 def get_device_properties(self) -> Dict[str, str]:
151 """
152 Retrieves key properties and identifiers from the Android device.
153
154 Returns:
155 A dictionary containing device properties such as
156 fingerprint, brand, device, manufacturer, model, name, serial number,
157 Android version, API level, and IMEI.
158
159 Notes:
160 Uses the :meth:`get_property` method to fetch system properties.
161 IMEI is retrieved using a shell command that parses the output of
162 :textmono:`service call iphonesubinfo`.
163 Handles missing or empty property values gracefully.
164 """
165 props = [
166 ("fingerprint", "ro.vendor.build.fingerprint"),
167 ("brand", "ro.product.brand"),
168 ("device", "ro.product.device"),
169 ("manufacturer", "ro.product.manufacturer"),
170 ("model", "ro.product.model"),
171 ("name", "ro.product.name"),
172 ("serialno", "ro.serialno"),
173 ("android_version", "ro.build.version.release"),
174 ("api_level", "ro.build.version.sdk"),
175 ]
176 device_properties = {}
177 for name, key in props:
178 device_properties[name] = self.get_property(key).strip()
179 # Get IMEI
180 imei = self.adb_shell(
181 """service call iphonesubinfo 1|awk -F "'" '{print $2}'|sed '1 d'|tr -d '.'|awk '{print}' ORS="""
182 )
183 device_properties["imei"] = imei.strip()
184 return device_properties
185
186 def _get_system_properties(self) -> Dict[str, Any]:
187 return self.get_frida_device().query_system_parameters()
188
[docs]
189 @cached_property
190 def system_properties(self) -> Dict[str, Any]:
191 return self._get_system_properties()
192
[docs]
193 @cached_property
194 def architecture(self):
195 return self._get_architecture()
196
197 def _get_architecture(self) -> str:
198 """
199 Returns the device CPU architecture.
200
201 Returns:
202 The CPU architecture string, such as 'arm64', 'x86_64', 'arm', or 'x86'.
203 Raises a RuntimeError if the architecture cannot be determined.
204 """
205 arch = self.system_properties["arch"]
206 if arch == "arm64":
207 return "arm64"
208 elif arch == "arm":
209 return "arm32"
210 elif arch == "ia32":
211 return "x86"
212 elif arch == "x64":
213 return "x86_64"
214 elif "x86" in arch:
215 return "x86"
216 else:
217 raise RuntimeError(f"Unknown architecture: {arch}")
218
[docs]
219 def get_tcpdump_version(self):
220 return f"tcpdump_{self.architecture}_android"
221
[docs]
222 def is_rooted(self) -> bool:
223 """
224 Checks if the device is rooted.
225
226 Returns:
227 True if the device is rooted or has :textmono:`su` access, False otherwise.
228 """
229 # Check user is root
230 try:
231 self.is_root = "root" in self.adb_device.shell("whoami")
232 except RuntimeError(): # ADB is not ready
233 logger.error("ADB is not ready, cannot check if device rooted.")
234 return False
235 # Check su
236 if not self.is_root:
237 ret = self.adb_device.shell('su -c "echo 1"').strip()
238 self.requires_su = "1" == ret
239 self.rooted = self.is_root or self.requires_su
240 return self.rooted
241
[docs]
242 def adb_shell(self, command) -> str:
243 """
244 Executes a shell command on the device via ADB.
245
246 Args:
247 command: The shell command to execute.
248
249 Returns:
250 The output of the shell command as a string.
251
252 Raises:
253 Exception: If the command execution fails.
254
255 Uses :textmono:`su` if root access is required and sets :textmono:`timeout=30`.
256 """
257 if self.requires_su:
258 command = f'su -c "{command}"'
259 return self.adb_device.shell(command, timeout=30)
260
[docs]
261 def adb_shell_nohup(self, command):
262 """
263 Executes a shell command on the device without waiting for output.
264
265 Args:
266 command: The shell command to execute.
267
268 Uses :textmono:`su` if root access is required. Opens the shell command with short
269 timeouts for non-blocking execution.
270 """
271
272 def dummy_handler(_):
273 pass
274
275 # The 'nohup' command ignores the hangup signal.
276 # Output is redirected to /dev/null to avoid leaving nohup.out files.
277 if self.requires_su:
278 shell_command = f'su -c "nohup {command} > /dev/null 2>&1"'
279 else:
280 shell_command = f"nohup {command} > /dev/null 2>&1"
281
282 try:
283 self.adb_device.shell(shell_command, handler=dummy_handler)
284 logger.info("Process started successfully")
285 except Exception as e:
286 logger.error(f"An error occurred while starting the process: {e}")
287
[docs]
288 def adb_push(self, local_path, device_path):
289 """
290 Pushes a file from the local system to the device.
291
292 Args:
293 local_path: Path to the local file.
294 device_path: Destination path on the device.
295
296 Raises:
297 Exception: If the push operation fails.
298
299 Uses the ADB push method to transfer files.
300 """
301 try:
302 self.adb_device.push(local_path, device_path)
303 except (Exception,) as e:
304 raise Exception(f"Failed to push {local_path} to {device_path}") from e
305
[docs]
306 def adb_pull(self, device_path, local_path):
307 """
308 Pulls a file from the device to the local system.
309
310 Args:
311 device_path: Path to the file on the device.
312 local_path: Destination path on the local system.
313
314 Raises:
315 Exception: If the pull operation fails.
316
317 Uses the ADB pull method to transfer files.
318 """
319 try:
320 self.adb_device.pull(str(device_path), str(local_path))
321 except (Exception,) as e:
322 raise Exception(f"Failed to pull {device_path} to {local_path}") from e
323
[docs]
324 def get_property(self, key: str) -> str:
325 """
326 Retrieves a system property from the device.
327
328 Args:
329 key: The property key to retrieve.
330
331 Returns:
332 The value of the system property as a string.
333
334 Uses the :textmono:`getprop` shell command.
335 """
336 value = self.adb_shell(f"getprop {key}") or ""
337 return value
338
[docs]
339 def check_frida_server_running(self) -> bool:
340 """
341 Checks if the Frida server process is running on the device.
342
343 Returns:
344 True if the Frida server is running, False otherwise.
345
346 This method uses the :textmono:`ps` command to search for the Frida server process.
347 """
348 value = self.adb_shell(f"ps -A | grep {FridaServer.executable}")
349 value = value.strip()
350 return bool(value)
351
[docs]
352 def check_frida_server_installed(self) -> bool:
353 """
354 Checks if the Frida server binary is installed on the device.
355
356 Returns:
357 True if the Frida server binary exists, False otherwise.
358
359 Uses the :textmono:`ls` command to verify the presence of the Frida server binary.
360 """
361 status = self.adb_shell(f"ls {FridaServer.executable_path}")
362 return "No such file or directory" not in status
363
[docs]
364 def get_frida_server_version(self) -> str:
365 """
366 Retrieves the version of the installed Frida server.
367
368 Returns:
369 The version string of the Frida server, or '0.0.0' if not found.
370
371 Executes the Frida server binary with the :textmono:`--version` flag.
372 """
373 output = self.adb_shell(f"{FridaServer.executable_path} --version").strip()
374 if not output or "inaccessible or not found" in output:
375 version = "unknown"
376 else:
377 version = output
378 return version
379
[docs]
380 def get_frida_device(self) -> Device:
381 raise NotImplementedError()
382
[docs]
383 def start_frida_server(self, force_stop: bool = True):
384 """
385 Starts the Frida server on the device if it is not already running.
386
387 Args:
388 force_stop: Forces stopping the Frida server before starting it.
389
390 If the server is already running, just logs an informational message.
391 Otherwise, starts the server in daemon mode.
392 """
393 if not self.check_frida_server_installed():
394 self.install_frida_server()
395 if self.get_frida_server_version() != frida.__version__:
396 self.install_frida_server()
397 if force_stop:
398 self.stop_frida_server()
399 if self.check_frida_server_running():
400 logger.info("Frida server is already running...")
401 else:
402 logger.info("Starting Frida server...")
403 self.adb_shell_nohup(f"{FridaServer.executable_path} -l 0.0.0.0 --daemonize")
404
[docs]
405 def stop_frida_server(self):
406 """
407 Stops the Frida server process on the device.
408
409 Uses the :textmono:`pkill` command to terminate the Frida server process.
410 """
411 logger.info("Stopping Frida server...")
412 self.adb_shell(f"pkill -f -l 9 {FridaServer.executable}")
413
[docs]
414 def install_frida_server(self, version: Optional[str] = None):
415 """Installs the Frida server binary on the device.
416
417 Downloads the specified version of the Frida server binary,
418 pushes it to the device, and sets the executable permission.
419 Uses a temporary file for the download.
420
421 Args:
422 version: The version of the Frida server to install.
423 Defaults to the currently installed ``frida`` Python
424 package version.
425 """
426 target_version = version or frida.__version__
427 logger.info(f"Installing frida-server {target_version} on device ({FridaServer.executable})...")
428
429 # Stop any running Frida server instance before replacing the binary
430 self.stop_frida_server()
431
432 with NamedTemporaryFile(mode="wb") as frida_server:
433 # Download the appropriate binary for this device's architecture
434 FridaServer.download_frida_server(
435 self.architecture,
436 frida_server.name,
437 "android",
438 target_version,
439 )
440 frida_server.seek(0)
441
442 # Push the binary to the device and make it executable
443 self.adb_push(frida_server.name, FridaServer.executable_path)
444 self.adb_shell(f"chmod +x {FridaServer.executable_path}")
445
446 logger.info(f"frida-server version {target_version} successfully installed.")
447
[docs]
448 def install_tcpdump(self):
449 """
450 Installs the tcpdump binary on the Android device.
451
452 Copies the appropriate tcpdump binary for the device's architecture
453 from the local assets directory to the device's temporary directory
454 and sets the executable permission.
455
456 Logs the progress and outcome of the installation.
457 """
458 logger.info(f"Installing tcpdump on device {self.tcpdump_path}.")
459 tcpdump_version = self.get_tcpdump_version()
460 tcpdump_binary = self.tcpdump_binaries_dir / tcpdump_version
461 self.adb_device.push(str(tcpdump_binary), str(self.tcpdump_path))
462 self.adb_shell(f"chmod +x {self.tcpdump_path}")
463 logger.info("tcpdump successfully installed on the device.")
464
465
[docs]
466class AndroidDeviceUsb(AndroidDevice):
467 """
468 Android device connected via USB.
469
470 Inherits from :class:`~octopus.android.device.AndroidDevice` and
471 initializes the device using a default :class:`~octopus.android.adb.ADB`
472 instance for USB communication.
473 """
474
[docs]
475 def __init__(self, device_id: Optional[str] = None, adb_host="127.0.0.1", adb_port=5037):
476 """Instantiate an AndroidDeviceUsb instance.
477
478 Connects to an ADB client and selects the appropriate device.
479 If a ``device_id`` is provided, connects to that specific device.
480 If only one device is connected, selects it automatically.
481
482 Args:
483 device_id: Optional ADB device serial number. If omitted and
484 exactly one device is connected, it is selected
485 automatically. Raises :exc:`RuntimeError` if no device
486 can be determined.
487 adb_host: Optional ADB host address, defaults to 127.0.0.1.
488 adb_port: Optional ADB port number, defaults to 5037.
489
490 Raises:
491 RuntimeError: If no device is found and ``device_id`` is not
492 provided, or if more than one device is connected without
493 specifying a ``device_id``.
494 """
495 client = AdbClient(adb_host, adb_port)
496 client.devices()
497 if device_id:
498 device = client.device(device_id)
499 elif len(client.devices()) == 1:
500 device = client.devices()[0]
501 else:
502 raise RuntimeError("No device found.")
503 super().__init__(device)
504
[docs]
505 def get_frida_device(self) -> Device:
506 return frida.get_usb_device()
507
508
[docs]
509class AndroidDeviceTcp(AndroidDevice):
510 """
511 Android device connected via TCP/IP.
512
513 Inherits from :class:`~octopus.android.device.AndroidDevice` and
514 initializes the device using a :class:`~octopus.android.adb.ADB` instance
515 configured for TCP/IP communication.
516 """
517
[docs]
518 def __init__(self, host: str, port: int = 5555, adb_host="127.0.0.1", adb_port=5037):
519 """
520 Initializes an AndroidDeviceTcp instance.
521
522 Args:
523 host: The IP address or hostname of the device.
524 port: The TCP port for ADB connection, defaults to 5555.
525 adb_host: Optional ADB host address, defaults to 127.0.0.1.
526 adb_port: Optional ADB port number, defaults to 5037.
527 """
528 client = AdbClient(adb_host, adb_port)
529 client.remote_connect(host, port)
530 device = client.device(f"{host}:{port}")
531 super().__init__(device)
532 self.host = host
533 self.port = port
534
[docs]
535 def get_frida_device(self) -> Device:
536 return frida.get_device_manager().add_remote_device(self.host)