1# SPDX-FileCopyrightText: 2026 Defensive Lab Agency
2# SPDX-FileContributor: u039b <git@0x39b.fr>
3#
4# SPDX-License-Identifier: GPL-3.0-or-later
5import logging
6import signal
7import threading
8import time
9
10from octopus.android.device import AndroidDeviceUsb, AndroidDeviceTcp
11from octopus.capture.manager import CaptureManager
12
13logger = logging.getLogger("octopus")
14
15
[docs]
16def common(device, output_path, no_screen_record, no_network_capture, no_instrumentation, duration, overwrite):
17 """Run the capture session for a given device.
18
19 Sets up signal handlers for graceful shutdown, initializes the
20 :class:`~octopus.capture.manager.CaptureManager`, and runs the
21 capture session for the specified duration or until interrupted.
22
23 Args:
24 device: The Android device instance to capture from.
25 output_path: Path to the directory where output will be saved.
26 no_screen_record: If True, disables screen recording.
27 no_network_capture: If True, disables network traffic capture.
28 no_instrumentation: If True, disables app instrumentation.
29 duration: Duration in seconds to run the capture. If 0 or less,
30 runs until a stop signal is received.
31 overwrite: If True, allows overwriting existing output files.
32 """
33 stop_event = threading.Event()
34
35 def _handle_stop(signum: int, frame) -> None:
36 """Handle OS stop signals by setting the stop event."""
37 stop_event.set()
38
39 # Register signal handlers for graceful shutdown on SIGINT and SIGTERM
40 signal.signal(signal.SIGINT, _handle_stop)
41 signal.signal(signal.SIGTERM, _handle_stop)
42
43 capture_manager = CaptureManager(
44 device, output_path, no_screen_record, no_network_capture, no_instrumentation, overwrite
45 )
46
47 # Abort early if the capture manager failed to initialize
48 if not capture_manager.ready:
49 return
50
51 capture_manager.start()
52
53 if duration > 0:
54 # Run for the specified duration, then signal stop
55 time.sleep(duration)
56 stop_event.set()
57 else:
58 # Wait indefinitely until a stop signal is received
59 stop_event.wait()
60
61 capture_manager.stop()
62
63
[docs]
64def usb_mode(
65 adb_host,
66 adb_port,
67 device_id,
68 output_path,
69 no_screen_record,
70 no_network_capture,
71 no_instrumentation,
72 duration,
73 overwrite,
74):
75 """Start a capture session using a USB-connected Android device.
76
77 Creates an :class:`~octopus.android.device.AndroidDeviceUsb` instance
78 and delegates to :func:`common` to run the capture session.
79
80 Args:
81 adb_host: Hostname or IP address of the ADB server.
82 adb_port: Port number of the ADB server.
83 device_id: The serial ID of the USB-connected Android device.
84 output_path: Path to the directory where output will be saved.
85 no_screen_record: If True, disables screen recording.
86 no_network_capture: If True, disables network traffic capture.
87 no_instrumentation: If True, disables app instrumentation.
88 duration: Duration in seconds to run the capture. If 0 or less,
89 runs until a stop signal is received.
90 overwrite: If True, allows overwriting existing output files.
91 """
92 try:
93 device = AndroidDeviceUsb(device_id, adb_host, adb_port)
94 common(device, output_path, no_screen_record, no_network_capture, no_instrumentation, duration, overwrite)
95 except (Exception,) as e:
96 logger.error(e)
97
98
[docs]
99def tcp_mode(
100 adb_host,
101 adb_port,
102 device_host,
103 device_port,
104 output_path,
105 no_screen_record,
106 no_network_capture,
107 no_instrumentation,
108 duration,
109 overwrite,
110):
111 """Start a capture session using a TCP-connected Android device.
112
113 Creates an :class:`~octopus.android.device.AndroidDeviceTcp` instance
114 and delegates to :func:`common` to run the capture session.
115
116 Args:
117 adb_host: Hostname or IP address of the ADB server.
118 adb_port: Port number of the ADB server.
119 device_host: Hostname or IP address of the Android device.
120 device_port: Port number used to connect to the Android device.
121 output_path: Path to the directory where output will be saved.
122 no_screen_record: If True, disables screen recording.
123 no_network_capture: If True, disables network traffic capture.
124 no_instrumentation: If True, disables app instrumentation.
125 duration: Duration in seconds to run the capture. If 0 or less,
126 runs until a stop signal is received.
127 overwrite: If True, allows overwriting existing output files.
128 """
129 try:
130 device = AndroidDeviceTcp(device_host, device_port, adb_host, adb_port)
131 common(device, output_path, no_screen_record, no_network_capture, no_instrumentation, duration, overwrite)
132 except (Exception,) as e:
133 logger.error(e)