Added preliminary SwitchOS v12 support

Added address and profile functionality

Added address and profile emulation

Updated connection and slowed frequency

Updated API call

nxbt/bluez.py

@@ -3,6 +3,8 @@ import re
 import os
 import time
 import logging
+from shutil import which
+import random
 
 import dbus
 
@@ -90,17 +92,20 @@ def find_objects(bus, service_name, interface_name):
     return paths
 
 
-def toggle_input_plugin(toggle):
+def toggle_clean_bluez(toggle):
-    """Enables or disables the BlueZ input plugin. Requires
+    """Enables or disables all BlueZ plugins,
-    root user to be run. The units and Bluetooth service will
+    BlueZ compatibility mode, and removes all extraneous
-    not be restarted if the input plugin already matches
+    SDP Services offered.
-    the toggle.
+    Requires root user to be run. The units and Bluetooth
+    service will not be restarted if the input plugin
+    already matches the toggle.
 
-    :param toggle: A boolean element indicating if the plugin
+    :param toggle: A boolean element indicating if BlueZ 
-    is enabled (True) or disabled (False)
+    should be cleaned (True) or not (False)
     :type toggle: boolean
     :raises PermissionError: If the user is not root
     :raises Exception: If the units can't be reloaded
+    :raises Exception: If sdptool, hciconfig, or bdaddr are not available.
     """
 
     service_path = "/lib/systemd/system/bluetooth.service"
@@ -114,43 +119,148 @@ def toggle_input_plugin(toggle):
         line = lines[i]
         if line.startswith("ExecStart="):
             # If we want to ensure the plugin is enabled
-            if toggle:
+            if not toggle:
                 # If input is already enabled
-                if "--noplugin=input" not in line:
+                if "--compat --noplugin=*" not in line:
                     return
-                lines[i] = re.sub(" --noplugin=input", "", line)
+                lines[i] = re.sub(" --compat --noplugin=\*", "", line)
             else:
                 # If input is already disabled
-                if "--noplugin=input" in line:
+                if "--compat --noplugin=*" in line:
                     return
                 # If not, add the flag
-                lines[i] = line + " --noplugin=input"
+                lines[i] = line + " --compat --noplugin=*"
 
     service = "\n".join(lines)
     with open(service_path, "w") as f:
         f.write(service)
 
     # Reload units
-    result = subprocess.run(
+    _run_command(["systemctl", "daemon-reload"])
-        ["systemctl", "daemon-reload"],
-        stderr=subprocess.PIPE)
-
-    cmd_err = result.stderr.decode("utf-8").replace("\n", "")
-    if cmd_err != "":
-        raise Exception(cmd_err)
 
     # Reload the bluetooth service with input disabled
-    result = subprocess.run(
+    _run_command(["systemctl", "restart", "bluetooth"])
-            ["systemctl", "restart", "bluetooth"],
-            stderr=subprocess.PIPE)
-
-    cmd_err = result.stderr.decode("utf-8").replace("\n", "")
-    if cmd_err != "":
-        raise Exception(cmd_err)
 
     # Kill a bit of time here to ensure all services have restarted
     time.sleep(0.5)
 
+    # Only clean out SDP records if we're toggling ON
+    if toggle:
+        clean_sdp_records()
+
+def clean_sdp_records():
+    """Cleans all SDP Records from BlueZ with sdptool
+
+    :raises Exception: On CLI error or sdptool missing
+    """
+    # TODO: sdptool is deprecated in BlueZ 5. This should ideally
+    # use the DBus API, however, bugs seemingly exist with the
+    # UnregisterProfile interface.
+
+    # Check if sdptool is available for use
+    if which("sdptool") is None:
+        raise Exception("sdptool is not available on this system." +
+                        "If you can, please install this tool, as " +
+                        "it is required for proper functionality.")
+
+    # Enable Read/Write to the SDP server. This is a remedy for a 
+    # compatibility mode bug introduced in later versions of BlueZ 5
+    _run_command(["chmod", "777", "/var/run/sdp"])
+
+    # Identify/List all SDP services available with sdptool
+    result = _run_command(['sdptool', 'browse', 'local']).stdout.decode('utf-8')
+    if result is None or len(result.split('\n\n')) < 1:
+        return
+    
+    # Record all service record handles
+    exceptions = ["PnP Information"]
+    service_rec_handles = []
+    for rec in result.split('\n\n'):
+        # Skip if exception is in record
+        exception_found = False
+        for exception in exceptions:
+            if exception in rec:
+                exception_found = True
+                break
+        if exception_found:
+            continue
+
+        # Read lines and add Record Handles to the list
+        for line in rec.split('\n'):
+            if "Service RecHandle" in line:
+                service_rec_handles.append(line.split(" ")[2])
+    
+    # Delete all found service records
+    if len(service_rec_handles) > 0:
+        for record_handle in service_rec_handles:
+            _run_command(['sdptool', 'del', record_handle])
+
+
+def _run_command(command):
+    """Runs a specified command on the shell of the system.
+    If the command is run unsuccessfully, an error is raised.
+    The command must be in the form of an array with each term
+    individually listed. Eg: ["which", "bash"]
+
+    :param command: A list of command terms
+    :type command: list
+    :raises Exception: On command failure or error
+    """
+    result = subprocess.run(
+        command,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE)
+
+    cmd_err = result.stderr.decode("utf-8").replace("\n", "")
+    if cmd_err != "":
+        raise Exception(cmd_err)
+    
+    return result
+
+
+def get_random_controller_mac():
+    """Generates a random Switch-compliant MAC address
+    """
+    def seg():
+        random_number = random.randint(0,255)
+        hex_number = str(hex(random_number))
+        hex_number = hex_number[2:].upper()
+        return str(hex_number)
+    
+    return f"7C:BB:8A:{seg()}:{seg()}:{seg()}"
+
+
+def replace_mac_addresses(adapter_paths, addresses):
+    """Replaces a list of adapter's Bluetooth MAC addresses
+    with Switch-compliant Controller MAC addresses. If the
+    addresses argument is specified, the adapter path's
+    MAC addresses will be reset to respective (index-wise)
+    address in the list.
+
+    :param adapter_paths: A list of Bluetooth adapter paths
+    :type adapter_paths: list
+    :param addresses: A list of Bluetooth MAC addresses,
+    defaults to False
+    :type addresses: bool, optional
+    """
+    if which("bdaddr") is None:
+        raise Exception("bdaddr is not available on this system." +
+                        "If you can, please install this tool, as " +
+                        "it is required for proper functionality.")
+    if which("hciconfig") is None:
+        raise Exception("hciconfig is not available on this system." +
+                        "If you can, please install this tool, as " +
+                        "it is required for proper functionality.")
+
+    if addresses:
+        assert len(addresses) == len(adapter_paths)
+
+    for i in range(len(adapter_paths)):
+        adapter_id = adapter_paths[i].split('/')[-1]
+        mac = addresses[i]
+        _run_command(['bdaddr', '-i', adapter_id, mac])
+        _run_command(['hciconfig', adapter_id, 'reset'])
+
 
 def find_devices_by_alias(alias):
     """Finds the Bluetooth addresses of devices
@@ -246,6 +356,36 @@ class BlueZ():
 
         return self.device.Get(ADAPTER_INTERFACE, "Address").upper()
 
+    def set_address(self, address):
+        """Sets the Bluetooth MAC address of the Bluetooth adapter.
+        The hciconfig CLI is required for setting the address.
+
+        :param address: A Bluetooth MAC address in 
+        the form of "XX:XX:XX:XX:XX:XX
+        :type address: str
+        :raises PermissionError: On run as non-root user
+        :raises Exception: On CLI errors
+        """
+        if which("bdaddr") is None:
+            raise Exception("bdaddr is not available on this system." +
+                            "If you can, please install this tool, as " +
+                            "it is required for proper functionality.")
+        _run_command(['bdaddr', '-i', self.device_id, address])
+
+    def set_class(self, device_class):
+        if which("hciconfig") is None:
+            raise Exception("hciconfig is not available on this system." +
+                            "If you can, please install this tool, as " +
+                            "it is required for proper functionality.")
+        _run_command(['hciconfig', self.device_id, 'class', device_class])
+
+    def reset_adapter(self):
+        if which("hciconfig") is None:
+            raise Exception("hciconfig is not available on this system." +
+                            "If you can, please install this tool, as " +
+                            "it is required for proper functionality.")
+        _run_command(['hciconfig', self.device_id, 'reset'])
+
     @property
     def name(self):
         """Gets the name of the Bluetooth adapter.
@@ -470,7 +610,16 @@ class BlueZ():
         :type opts: dict
         """
 
-        self.profile_manager.RegisterProfile(profile_path, uuid, opts)
+        return self.profile_manager.RegisterProfile(profile_path, uuid, opts)
+
+    def unregister_profile(self, profile):
+        """Unregisters a given SDP record from the BlueZ SDP server.
+
+        :param profile: A SDP record profile object
+        :type profile: Profile
+        """
+
+        self.profile_manager.UnregisterProfile(profile)
 
     def reset(self):
         """Restarts the Bluetooth Service

nxbt/controller/server.py

@@ -5,6 +5,7 @@ import time
 import queue
 import logging
 import traceback
+import atexit
 
 from .controller import Controller, ControllerTypes
 from ..bluez import BlueZ
@@ -23,6 +24,8 @@ class ControllerServer():
         # Cache logging level to increase performance on checks
         self.logger_level = self.logger.level
 
+        atexit.register(self._on_exit)
+
         if state:
             self.state = state
         else:
@@ -165,12 +168,14 @@ class ControllerServer():
                 # or 60Hz for Joy-Cons.
                 # Sleep timers are compensated with the elapsed command
                 # processing time.
+                PRO_CONTROLLER_FREQUENCY = 1/15
+                JOYCON_FREQUENCY = 1/15
                 if self.controller_type == ControllerTypes.PRO_CONTROLLER:
-                    if elapsed_time < 1/120:
+                    if elapsed_time < PRO_CONTROLLER_FREQUENCY:
-                        time.sleep(1/120 - elapsed_time)
+                        time.sleep(PRO_CONTROLLER_FREQUENCY - elapsed_time)
                 else:
-                    if elapsed_time < 1/60:
+                    if elapsed_time < JOYCON_FREQUENCY:
-                        time.sleep(1/60 - elapsed_time)
+                        time.sleep(JOYCON_FREQUENCY - elapsed_time)
 
     def save_connection(self, error, state=None):
 
@@ -263,8 +268,14 @@ class ControllerServer():
 
         self.bt.set_discoverable(True)
 
-        ctrl, ctrl_address = s_ctrl.accept()
+        # WARNING:
+        # A device's class must be set **AFTER** discoverability
+        # is set. If it is set before or in a similar timeframe,
+        # the class will be reset to the default value.
+        self.bt.set_class("0x02508")
+
         itr, itr_address = s_itr.accept()
+        ctrl, ctrl_address = s_ctrl.accept()
 
         # Send an empty input report to the Switch to prompt a reply
         self.protocol.process_commands(None)
@@ -277,6 +288,7 @@ class ControllerServer():
         fcntl.fcntl(itr, fcntl.F_SETFL, os.O_NONBLOCK)
 
         # Mainloop
+        received_first_message = False
         while True:
             # Attempt to get output from Switch
             try:
@@ -286,6 +298,9 @@ class ControllerServer():
             except BlockingIOError:
                 reply = None
 
+            if reply:
+                received_first_message = True
+
             self.protocol.process_commands(reply)
             msg = self.protocol.get_report()
 
@@ -305,6 +320,9 @@ class ControllerServer():
 
             # Switch responds to packets slower during pairing
             # Pairing cycle responds optimally on a 15Hz loop
+            if not received_first_message:
+                time.sleep(1)
+            else:
                 time.sleep(1/15)
 
         self.slow_input_frequency = True
@@ -377,3 +395,6 @@ class ControllerServer():
         fcntl.fcntl(itr, fcntl.F_SETFL, os.O_NONBLOCK)
 
         return itr, ctrl
+
+    def _on_exit(self):
+        self.bt.reset_address()

nxbt/nxbt.py

@@ -12,7 +12,8 @@ import dbus
 
 from .controller import ControllerServer
 from .controller import ControllerTypes
-from .bluez import find_objects, toggle_input_plugin
+from .bluez import BlueZ, find_objects, toggle_clean_bluez
+from .bluez import replace_mac_addresses
 from .bluez import find_devices_by_alias
 from .bluez import SERVICE_NAME, ADAPTER_INTERFACE
 from .logging import create_logger
@@ -175,12 +176,33 @@ class Nxbt():
         self._adapters_in_use = {}
         self._controller_adapter_lookup = {}
 
+        # Save all MAC addresses for existing adapters and replace
+        # with Controller MAC addresses
+        self.cached_adapters = self.get_available_adapters()
+        self.old_addresses = []
+        self.masked_addresses = []
+        for adapter in self.cached_adapters:
+            bt = BlueZ(adapter_path=adapter)
+
+            # Saving old address
+            self.old_addresses.append(bt.address)
+
+            # Creating/saving a Switch-compliant masked address
+            address = bt.address.split(":")
+            address[0] = "7C"
+            address[1] = "BB"
+            address[2] = "8A"
+            address = ":".join(address)
+            self.masked_addresses.append(address)
+
+            bt.bus.close()
+        # Replace the old MAC addresses with the Switch-compliant
+        # masked ones
+        replace_mac_addresses(self.cached_adapters, self.masked_addresses)
+
         # Disable the BlueZ input plugin so we can use the
         # HID control/interrupt Bluetooth ports
-        try:
+        toggle_clean_bluez(True)
-            toggle_input_plugin(False)
-        except PermissionError:
-            pass
 
         # Exit handler
         atexit.register(self._on_exit)
@@ -209,11 +231,11 @@ class Nxbt():
 
         self.resource_manager.shutdown()
 
-        # Re-enable the BlueZ input plugin, if we have permission
+        # Reset Bluetooth MAC addresses
-        try:
+        replace_mac_addresses(self.cached_adapters, self.old_addresses)
-            toggle_input_plugin(True)
+
-        except PermissionError:
+        # Re-enable the BlueZ plugins, if we have permission
-            pass
+        toggle_clean_bluez(False)
 
     def _command_manager(self, task_queue, state):
         """Used as the main multiprocessing Process that is launched
@@ -641,6 +663,7 @@ class Nxbt():
 
         bus = dbus.SystemBus()
         adapters = find_objects(bus, SERVICE_NAME, ADAPTER_INTERFACE)
+        bus.close()
 
         return adapters
 

scripts/crash_switch.py

@@ -44,7 +44,7 @@ import os
 import time
 import fcntl
 
-from nxbt import toggle_input_plugin
+from nxbt import toggle_clean_bluez
 from nxbt import BlueZ
 from nxbt import Controller
 from nxbt import PRO_CONTROLLER
@@ -142,7 +142,7 @@ if __name__ == "__main__":
     port_ctrl = 17
     port_itr = 19
 
-    toggle_input_plugin(False)
+    toggle_clean_bluez(False)
     bt = BlueZ(adapter_path="/org/bluez/hci0")
 
     controller = Controller(bt, PRO_CONTROLLER)
@@ -225,4 +225,4 @@ if __name__ == "__main__":
         raise e
 
     finally:
-        toggle_input_plugin(True)
+        toggle_clean_bluez(True)

scripts/proxy.py

@@ -35,7 +35,7 @@ import time
 import fcntl
 from time import perf_counter
 
-from nxbt import toggle_input_plugin
+from nxbt import toggle_clean_bluez
 from nxbt import BlueZ
 from nxbt import Controller
 from nxbt import JOYCON_L, JOYCON_R, PRO_CONTROLLER
@@ -116,7 +116,8 @@ def write_to_buffer(buffer, message, message_type):
 
 if __name__ == "__main__":
     # Switch Controller Bluetooth MAC Address goes here
-    jc_MAC = "98:B6:E9:B0:05:E7"
+    # jc_MAC = "98:B6:E9:B0:05:E7"
+    jc_MAC = "7C:BB:8A:FA:41:3D"
     # Specify the type of controller here
     controller_type = PRO_CONTROLLER
     if controller_type == JOYCON_L:
@@ -130,7 +131,7 @@ if __name__ == "__main__":
     port_itr = 19
     message_buffer = []
 
-    toggle_input_plugin(False)
+    toggle_clean_bluez(True)
     bt = BlueZ(adapter_path="/org/bluez/hci0")
 
     controller = Controller(bt, controller_type)
@@ -150,6 +151,12 @@ if __name__ == "__main__":
     switch_ctrl = socket.socket(family=socket.AF_BLUETOOTH,
                                 type=socket.SOCK_SEQPACKET,
                                 proto=socket.BTPROTO_L2CAP)
+    switch_test1 = socket.socket(family=socket.AF_BLUETOOTH,
+                                type=socket.SOCK_SEQPACKET,
+                                proto=socket.BTPROTO_L2CAP)
+    switch_test3 = socket.socket(family=socket.AF_BLUETOOTH,
+                                type=socket.SOCK_SEQPACKET,
+                                proto=socket.BTPROTO_L2CAP)
 
     try:
         # Remove the device before we try to re-pair
@@ -179,6 +186,23 @@ if __name__ == "__main__":
         jc_itr.connect((jc_MAC, port_itr))
         print("Got connection.")
 
+        # print("Bruteforcing sockets")
+        # for i in range(1,99999,2):
+        #     try:
+        #         test_socket = socket.socket(family=socket.AF_BLUETOOTH,
+        #                                 type=socket.SOCK_SEQPACKET,
+        #                                 proto=socket.BTPROTO_L2CAP)
+        #         test_socket.settimeout(3)
+        #         # print("Connecting to", i)
+        #         test_socket.connect((jc_MAC, i))
+        #         print("!!!!!! Got connection to", i)
+        #     except Exception as e:
+        #         # print(str(e))
+        #         pass
+
+        # switch_test1.bind((bt.address, 1))
+        # switch_test3.bind((bt.address, 3))
+
         switch_ctrl.bind((bt.address, port_ctrl))
         switch_itr.bind((bt.address, port_itr))
 
@@ -285,9 +309,9 @@ if __name__ == "__main__":
     except KeyboardInterrupt:
         print("Closing sockets")
 
-        time_new = perf_counter()
+        # time_new = perf_counter()
-        print(f"Total Delta: {(time_new - time_old) * 1000}")
+        # print(f"Total Delta: {(time_new - time_old) * 1000}")
-        print(f"Timer Counter: {timer_counter}")
+        # print(f"Timer Counter: {timer_counter}")
 
         jc_ctrl.close()
         jc_itr.close()
@@ -313,7 +337,11 @@ if __name__ == "__main__":
         switch_itr.close()
         switch_ctrl.close()
 
+        # Write the buffer
+        with open("messages.txt", "w") as f:
+            f.write("\n".join(message_buffer))
+
         raise e
 
     finally:
-        toggle_input_plugin(True)
+        toggle_clean_bluez(False)

scripts/reconnect_proxy.py

@@ -16,7 +16,7 @@ import time
 import fcntl
 from time import perf_counter
 
-from nxbt import toggle_input_plugin
+from nxbt import toggle_clean_bluez
 from nxbt import BlueZ
 from nxbt import Controller
 from nxbt import JOYCON_L, JOYCON_R, PRO_CONTROLLER
@@ -112,7 +112,7 @@ if __name__ == "__main__":
     port_itr = 19
     message_buffer = []
 
-    toggle_input_plugin(False)
+    toggle_clean_bluez(True)
     bt = BlueZ(adapter_path="/org/bluez/hci0")
 
     controller = Controller(bt, controller_type)
@@ -288,4 +288,4 @@ if __name__ == "__main__":
         raise e
 
     finally:
-        toggle_input_plugin(True)
+        toggle_clean_bluez(False)

scripts/reconnect_test.py

@@ -26,7 +26,7 @@ import os
 import time
 import fcntl
 
-from nxbt import toggle_input_plugin
+from nxbt import toggle_clean_bluez
 from nxbt import BlueZ
 from nxbt import Controller
 from nxbt import PRO_CONTROLLER
@@ -125,7 +125,7 @@ if __name__ == "__main__":
     port_ctrl = 17
     port_itr = 19
 
-    toggle_input_plugin(False)
+    toggle_clean_bluez(True)
     bt = BlueZ(adapter_path="/org/bluez/hci0")
 
     controller = Controller(bt, PRO_CONTROLLER)
@@ -199,4 +199,4 @@ if __name__ == "__main__":
         raise e
 
     finally:
-        toggle_input_plugin(True)
+        toggle_clean_bluez(False)

scripts/switch_emu.py

@@ -11,7 +11,7 @@ import sys
 import os
 import time
 
-from nxbt import toggle_input_plugin
+from nxbt import toggle_clean_bluez
 from nxbt import BlueZ
 
 REQUEST_INFO = b'\xA2\x01\x02\x00\x00\x00\x00\x00\x00\x00\x00\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
@@ -126,7 +126,7 @@ if __name__ == "__main__":
     jc_itr = socket.socket(family=socket.AF_BLUETOOTH,
                            type=socket.SOCK_SEQPACKET,
                            proto=socket.BTPROTO_L2CAP)
-    toggle_input_plugin(False)
+    toggle_clean_bluez(True)
     bt = BlueZ(adapter_path="/org/bluez/hci0")
 
     try:
@@ -194,4 +194,4 @@ if __name__ == "__main__":
         raise e
 
     finally:
-        toggle_input_plugin(True)
+        toggle_clean_bluez(False)