pythongh-70145: Add support for channels in Bluetooth HCI protocol (pythonGH-132481)

Author: serhiy-storchaka

Doc/library/socket.rst

@@ -156,8 +156,10 @@ created.  Socket addresses are represented as follows:
 
   - :const:`BTPROTO_HCI` accepts a format that depends on your OS.
 
-    - On Linux it accepts a tuple ``(device_id,)`` where ``device_id``
-      is an integer specifying the number of the Bluetooth device.
+    - On Linux it accepts a tuple ``(device_id, [channel])`` where ``device_id``
+      is an integer specifying the number of the Bluetooth device,
+      and ``channel`` is an optional integer specifying the HCI channel
+      (:const:`HCI_CHANNEL_RAW` by default).
     - On FreeBSD, NetBSD and DragonFly BSD it accepts ``bdaddr``
       where ``bdaddr`` is the Bluetooth address as a string.
 
@@ -167,6 +169,9 @@ created.  Socket addresses are represented as follows:
     .. versionchanged:: 3.13.3
        FreeBSD support added.
 
+    .. versionchanged:: next
+       Added ``channel`` field.
+
   - :const:`BTPROTO_SCO` accepts ``bdaddr`` where ``bdaddr`` is
     the Bluetooth address as a string or a :class:`bytes` object.
     (ex. ``'12:23:34:45:56:67'`` or ``b'12:23:34:45:56:67'``)
@@ -677,6 +682,18 @@ Constants
    available on Linux and FreeBSD. :const:`!HCI_TIME_STAMP` and
    :const:`!HCI_DATA_DIR` are only available on Linux.
 
+.. data:: HCI_CHANNEL_RAW
+          HCI_CHANNEL_USER
+          HCI_CHANNEL_MONITOR
+          HCI_CHANNEL_CONTROL
+          HCI_CHANNEL_LOGGING
+
+   Possible values for ``channel`` field in the :const:`BTPROTO_HCI` address.
+
+   .. availability:: Linux
+
+   .. versionadded:: next
+
 .. data:: AF_QIPCRTR
 
    Constant for Qualcomm's IPC router protocol, used to communicate with

Lib/test/test_socket.py

@@ -2622,6 +2622,13 @@ def testBluetoothConstants(self):
             socket.BTPROTO_L2CAP
             socket.BTPROTO_SCO
 
+        if sys.platform == "linux":
+            socket.HCI_CHANNEL_RAW
+            socket.HCI_CHANNEL_USER
+            socket.HCI_CHANNEL_MONITOR
+            socket.HCI_CHANNEL_CONTROL
+            socket.HCI_CHANNEL_LOGGING
+
     def testCreateRfcommSocket(self):
         with socket.socket(socket.AF_BLUETOOTH, socket.SOCK_STREAM, socket.BTPROTO_RFCOMM) as s:
             pass
@@ -2721,13 +2728,14 @@ def testBadRfcommAddr(self):
 
     @unittest.skipUnless(hasattr(socket, 'BTPROTO_HCI'), 'Bluetooth HCI sockets required for this test')
     def testBindHciSocket(self):
-        with socket.socket(socket.AF_BLUETOOTH, socket.SOCK_RAW, socket.BTPROTO_HCI) as s:
-            if sys.platform.startswith(('netbsd', 'dragonfly', 'freebsd')):
+        if sys.platform.startswith(('netbsd', 'dragonfly', 'freebsd')):
+            with socket.socket(socket.AF_BLUETOOTH, socket.SOCK_RAW, socket.BTPROTO_HCI) as s:
                 s.bind(socket.BDADDR_ANY)
                 addr = s.getsockname()
                 self.assertEqual(addr, socket.BDADDR_ANY)
-            else:
-                dev = 0
+        else:
+            dev = 0
+            with socket.socket(socket.AF_BLUETOOTH, socket.SOCK_RAW, socket.BTPROTO_HCI) as s:
                 try:
                     s.bind((dev,))
                 except OSError as err:
@@ -2737,6 +2745,26 @@ def testBindHciSocket(self):
                 addr = s.getsockname()
                 self.assertEqual(addr, dev)
 
+            with (self.subTest('channel=HCI_CHANNEL_RAW'),
+                  socket.socket(socket.AF_BLUETOOTH, socket.SOCK_RAW, socket.BTPROTO_HCI) as s):
+                channel = socket.HCI_CHANNEL_RAW
+                s.bind((dev, channel))
+                addr = s.getsockname()
+                self.assertEqual(addr, dev)
+
+            with (self.subTest('channel=HCI_CHANNEL_USER'),
+                  socket.socket(socket.AF_BLUETOOTH, socket.SOCK_RAW, socket.BTPROTO_HCI) as s):
+                channel = socket.HCI_CHANNEL_USER
+                try:
+                    s.bind((dev, channel))
+                except OSError as err:
+                    # Needs special permissions.
+                    if err.errno in (errno.EPERM, errno.EBUSY, errno.ERFKILL):
+                        self.skipTest(str(err))
+                    raise
+                addr = s.getsockname()
+                self.assertEqual(addr, (dev, channel))
+
     @unittest.skipUnless(hasattr(socket, 'BTPROTO_HCI'), 'Bluetooth HCI sockets required for this test')
     def testBadHciAddr(self):
         with socket.socket(socket.AF_BLUETOOTH, socket.SOCK_RAW, socket.BTPROTO_HCI) as s:
@@ -2760,7 +2788,7 @@ def testBadHciAddr(self):
                 with self.assertRaises(OSError):
                     s.bind(())
                 with self.assertRaises(OSError):
-                    s.bind((dev, 0))
+                    s.bind((dev, socket.HCI_CHANNEL_RAW, 0, 0))
                 with self.assertRaises(OSError):
                     s.bind(dev)
                 with self.assertRaises(OSError):

Misc/NEWS.d/next/Library/2025-04-13-19-17-14.gh-issue-70145.nJ2MKg.rst

@@ -0,0 +1,2 @@
+Add support for channels in Bluetooth HCI protocol
+(:const:`~socket.BTPROTO_HCI`).

Modules/socketmodule.c

@@ -1541,7 +1541,14 @@ makesockaddr(SOCKET_T sockfd, struct sockaddr *addr, size_t addrlen, int proto)
             struct sockaddr_hci *a = (struct sockaddr_hci *) addr;
 #if defined(HAVE_BLUETOOTH_BLUETOOTH_H)
             PyObject *ret = NULL;
-            ret = Py_BuildValue("i", _BT_HCI_MEMB(a, dev));
+            if (_BT_HCI_MEMB(a, channel) == HCI_CHANNEL_RAW) {
+                return Py_BuildValue("i", _BT_HCI_MEMB(a, dev));
+            }
+            else {
+                return Py_BuildValue("ii",
+                                     _BT_HCI_MEMB(a, dev),
+                                     _BT_HCI_MEMB(a, channel));
+            }
             return ret;
 #elif defined(__FreeBSD__)
             const char *node = _BT_HCI_MEMB(a, node);
@@ -2138,13 +2145,15 @@ getsockaddrarg(PySocketSockObject *s, PyObject *args,
             memset(addr, 0, sizeof(struct sockaddr_hci));
             _BT_HCI_MEMB(addr, family) = AF_BLUETOOTH;
 #if defined(HAVE_BLUETOOTH_BLUETOOTH_H)
-            unsigned short dev = _BT_HCI_MEMB(addr, dev);
-            if (!PyArg_ParseTuple(args, "H", &dev)) {
+            unsigned short dev;
+            unsigned short channel = HCI_CHANNEL_RAW;
+            if (!PyArg_ParseTuple(args, "H|H", &dev, &channel)) {
                 PyErr_Format(PyExc_OSError,
                              "%s(): wrong format", caller);
                 return 0;
             }
             _BT_HCI_MEMB(addr, dev) = dev;
+            _BT_HCI_MEMB(addr, channel) = channel;
 #else
             const char *straddr;
             if (!PyArg_Parse(args, "s", &straddr)) {
@@ -7874,6 +7883,13 @@ socket_exec(PyObject *m)
 #ifdef BTPROTO_HCI
     ADD_INT_MACRO(m, BTPROTO_HCI);
     ADD_INT_MACRO(m, SOL_HCI);
+#if defined(HCI_CHANNEL_RAW)
+    ADD_INT_MACRO(m, HCI_CHANNEL_RAW);
+    ADD_INT_MACRO(m, HCI_CHANNEL_USER);
+    ADD_INT_MACRO(m, HCI_CHANNEL_MONITOR);
+    ADD_INT_MACRO(m, HCI_CHANNEL_CONTROL);
+    ADD_INT_MACRO(m, HCI_CHANNEL_LOGGING);
+#endif
 #if defined(HCI_FILTER)
     ADD_INT_MACRO(m, HCI_FILTER);
 #endif