PTP capable interface detection without ethtool (New)

checkbox-support/checkbox_support/ethtool/tests/test_ts_info.py

@@ -0,0 +1,216 @@
+#!/usr/bin/env python3
+# This file is part of Checkbox.
+#
+# Copyright 2026 Canonical Ltd.
+# Authors:
+#   Zhongning Li <zhongning.li@canonical.com>
+#
+# Checkbox is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3,
+# as published by the Free Software Foundation.
+#
+# Checkbox is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Checkbox.  If not, see <http://www.gnu.org/licenses/>.
+
+import ctypes
+import tempfile
+from pathlib import Path
+from unittest import TestCase
+import unittest
+from unittest.mock import MagicMock, patch
+
+from checkbox_support.ethtool.ts_info import (
+    ETHTOOL_GET_TS_INFO,
+    SIOCETHTOOL,
+    ethtool_ts_info,
+    get_ts_info,
+    is_ptp_capable,
+    _is_ethernet_interface,
+)
+
+
+class TestEthtoolTsInfoStr(TestCase):
+    def test_str_pretty_prints_scalars_and_arrays(self):
+        info = ethtool_ts_info()
+        info.cmd = ETHTOOL_GET_TS_INFO
+        info.so_timestamping = 0x1F
+        info.phc_index = 2
+        info.tx_types = 7
+        info.tx_reserved[0] = 1
+        info.tx_reserved[1] = 2
+        info.tx_reserved[2] = 3
+        info.rx_filters = 9
+        info.rx_reserved[0] = 4
+        info.rx_reserved[1] = 5
+        info.rx_reserved[2] = 6
+
+        expected = (
+            "ethtool_ts_info(cmd={}, so_timestamping=31, phc_index=2, "
+            "tx_types=7, tx_reserved=[1, 2, 3], rx_filters=9, "
+            "rx_reserved=[4, 5, 6])"
+        ).format(ETHTOOL_GET_TS_INFO)
+        self.assertEqual(str(info), expected)
+
+    def test_str_default_values(self):
+        info = ethtool_ts_info()
+        expected = (
+            "ethtool_ts_info(cmd=0, so_timestamping=0, phc_index=0, "
+            "tx_types=0, tx_reserved=[0, 0, 0], rx_filters=0, "
+            "rx_reserved=[0, 0, 0])"
+        )
+        self.assertEqual(str(info), expected)
+
+
+class TestIsEthernetInterface(TestCase):
+    def _make_sys_class_net(
+        self,
+        tmp_dir,
+        has_type=True,
+        is_ether=True,
+        has_device=True,
+        is_wifi=False,
+    ):
+        iface_dir = Path(tmp_dir) / "enp1s0"
+        iface_dir.mkdir(parents=True)
+        if has_type:
+            (iface_dir / "type").write_text("1\n" if is_ether else "772\n")
+        if has_device:
+            (iface_dir / "device").mkdir()
+        if is_wifi:
+            (iface_dir / "phy80211").mkdir()
+        return iface_dir.parent
+
+    def test_physical_ethernet_interface(self):
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            base = self._make_sys_class_net(tmp_dir)
+            with patch(
+                "checkbox_support.ethtool.ts_info.Path",
+                side_effect=lambda p: (
+                    base if "/sys/class/net" in p else Path(p)
+                ),
+            ):
+                self.assertTrue(_is_ethernet_interface("enp1s0"))
+
+    def test_not_ether_type(self):
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            base = self._make_sys_class_net(tmp_dir, is_ether=False)
+            with patch(
+                "checkbox_support.ethtool.ts_info.Path",
+                side_effect=lambda p: (
+                    base if "/sys/class/net" in p else Path(p)
+                ),
+            ):
+                self.assertFalse(_is_ethernet_interface("enp1s0"))
+
+    def test_no_device_dir(self):
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            base = self._make_sys_class_net(tmp_dir, has_device=False)
+            with patch(
+                "checkbox_support.ethtool.ts_info.Path",
+                side_effect=lambda p: (
+                    base if "/sys/class/net" in p else Path(p)
+                ),
+            ):
+                self.assertFalse(_is_ethernet_interface("enp1s0"))
+
+    def test_wifi_interface(self):
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            base = self._make_sys_class_net(tmp_dir, is_wifi=True)
+            with patch(
+                "checkbox_support.ethtool.ts_info.Path",
+                side_effect=lambda p: (
+                    base if "/sys/class/net" in p else Path(p)
+                ),
+            ):
+                self.assertFalse(_is_ethernet_interface("enp1s0"))
+
+    def test_missing_interface_returns_false(self):
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            base = Path(tmp_dir)
+            with patch(
+                "checkbox_support.ethtool.ts_info.Path",
+                side_effect=lambda p: (
+                    base if "/sys/class/net" in p else Path(p)
+                ),
+            ):
+                self.assertFalse(_is_ethernet_interface("does-not-exist"))
+
+
+class TestGetTsInfo(TestCase):
+    def _fake_ioctl(self, fd, request, arg, mutate_flag=False):
+        # mock kernel response to the SIOCETHTOOL ioctl
+        self.assertEqual(request, SIOCETHTOOL)
+        info_ptr = ctypes.cast(
+            arg.ifr_ifru.ifr_data, ctypes.POINTER(ethtool_ts_info)
+        )
+        info_ptr.contents.so_timestamping = 0x1F
+        info_ptr.contents.phc_index = 3
+        info_ptr.contents.tx_types = 7
+        info_ptr.contents.rx_filters = 9
+        return 0
+
+    @patch("checkbox_support.ethtool.ts_info._is_ethernet_interface")
+    @patch("checkbox_support.ethtool.ts_info.fcntl.ioctl")
+    def test_get_ts_info_populates_struct(
+        self, ioctl_mock: MagicMock, is_ethernet_mock: MagicMock
+    ):
+        is_ethernet_mock.return_value = True
+        ioctl_mock.side_effect = self._fake_ioctl
+
+        info = get_ts_info("enp1s0")
+
+        self.assertGreaterEqual(len(ioctl_mock.call_args_list), 1)
+        self.assertEqual(info.cmd, ETHTOOL_GET_TS_INFO)
+        self.assertEqual(info.phc_index, 3)
+        self.assertEqual(info.so_timestamping, 0x1F)
+        self.assertEqual(info.tx_types, 7)
+        self.assertEqual(info.rx_filters, 9)
+
+    @patch("checkbox_support.ethtool.ts_info._is_ethernet_interface")
+    @patch("checkbox_support.ethtool.ts_info.fcntl.ioctl")
+    def test_get_ts_info_warns_on_non_ethernet(
+        self, ioctl_mock, is_ethernet_mock
+    ):
+        is_ethernet_mock.return_value = False
+        ioctl_mock.side_effect = self._fake_ioctl
+
+        with self.assertLogs(level="WARNING"):
+            get_ts_info("wlp1s0")
+
+
+class TestIsPtpCapable(TestCase):
+    @patch("checkbox_support.ethtool.ts_info.os.path.exists")
+    @patch("checkbox_support.ethtool.ts_info.get_ts_info")
+    def test_ptp_capable(self, get_ts_info_mock, exists_mock):
+        get_ts_info_mock.return_value = ethtool_ts_info(phc_index=0)
+        exists_mock.return_value = True
+
+        self.assertTrue(is_ptp_capable("enp1s0"))
+        exists_mock.assert_called_once_with("/dev/ptp0")
+
+    @patch("checkbox_support.ethtool.ts_info.os.path.exists")
+    @patch("checkbox_support.ethtool.ts_info.get_ts_info")
+    def test_no_phc_device(self, get_ts_info_mock, exists_mock):
+        get_ts_info_mock.return_value = ethtool_ts_info(phc_index=-1)
+        exists_mock.return_value = False
+
+        self.assertFalse(is_ptp_capable("enp1s0"))
+
+    @patch("checkbox_support.ethtool.ts_info.os.path.exists")
+    @patch("checkbox_support.ethtool.ts_info.get_ts_info")
+    def test_phc_index_valid_but_device_missing(
+        self, get_ts_info_mock, exists_mock
+    ):
+        get_ts_info_mock.return_value = ethtool_ts_info(phc_index=1)
+        exists_mock.return_value = False
+
+        self.assertFalse(is_ptp_capable("enp1s0"))
+
+
+if __name__ == "__main__":
+    unittest.main()

checkbox-support/checkbox_support/ethtool/ts_info.py

@@ -0,0 +1,177 @@
+#!/usr/bin/env python3
+# This file is part of Checkbox.
+#
+# Copyright 2026 Canonical Ltd.
+# Authors:
+#   Zhongning Li <zhongning.li@canonical.com>
+#
+# Checkbox is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3,
+# as published by the Free Software Foundation.
+#
+# Checkbox is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Checkbox.  If not, see <http://www.gnu.org/licenses/>.
+
+"""
+Source code of ethtool:
+https://git.kernel.org/pub/scm/network/ethtool/ethtool.git
+This python program re-implements this function:
+https://git.kernel.org/pub/scm/network/ethtool/ethtool.git/tree/ethtool.c#n4986
+"""
+
+import ctypes
+import fcntl
+import os
+import socket
+import logging
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+# these 2 are used to make same ioctl request as ethtool
+# https://github.com/torvalds/linux/blob/3b029c035b34bbc693405ddf759f0e9b920c27f1/include/uapi/linux/sockios.h#L102
+SIOCETHTOOL = 0x8946
+# https://github.com/torvalds/linux/blob/3b029c035b34bbc693405ddf759f0e9b920c27f1/include/uapi/linux/ethtool.h#L1940
+ETHTOOL_GET_TS_INFO = 0x00000041
+
+# yes this is actually how it's spelled
+# https://github.com/torvalds/linux/blob/3b029c035b34bbc693405ddf759f0e9b920c27f1/include/uapi/linux/if.h#L33
+IFNAMSIZ = 16
+
+
+# the main struct of interest
+# for the purpose of finding PTP capable interfaces we only need phc_index
+# we also need to specify .cmd to indicate that we want to do ETHTOOL_GET_TS_INFO
+# https://github.com/torvalds/linux/blob/3b029c035b34bbc693405ddf759f0e9b920c27f1/include/uapi/linux/ethtool.h#L1739
+class ethtool_ts_info(ctypes.Structure):
+    _fields_ = [
+        ("cmd", ctypes.c_uint32),
+        ("so_timestamping", ctypes.c_uint32),
+        ("phc_index", ctypes.c_int32),
+        ("tx_types", ctypes.c_uint32),
+        ("tx_reserved", ctypes.c_uint32 * 3),
+        ("rx_filters", ctypes.c_uint32),
+        ("rx_reserved", ctypes.c_uint32 * 3),
+    ]
+
+    def __str__(self) -> str:
+        words = []  # type: list[str]
+        for field in self._fields_:
+            name = field[0]
+            value = getattr(self, name)
+            if isinstance(value, ctypes.Array):
+                value = list(value)
+            words.append("{}={}".format(name, value))
+        return "ethtool_ts_info({})".format(", ".join(words))
+
+
+# ------ these structs are only used to make the ioctl request
+class ifmap(ctypes.Structure):
+    _fields_ = [
+        ("mem_start", ctypes.c_ulong),
+        ("mem_end", ctypes.c_ulong),
+        ("base_addr", ctypes.c_ushort),
+        ("irq", ctypes.c_ubyte),  # ubyte is unsigned char
+        ("dma", ctypes.c_ubyte),
+        ("port", ctypes.c_ubyte),
+    ]
+
+
+class ifr_ifru(ctypes.Union):
+    _fields_ = [
+        # there are more union members in the source code
+        # we are not really using them here so they are omitted
+        ("ifr_data", ctypes.c_void_p),
+        # ifmap is included for the padding
+        # because it's the largest union member
+        # it has to be here to avoid this field going out of bounds
+        ("ifr_map", ifmap),
+    ]
+
+
+# https://github.com/torvalds/linux/blob/fce2dfa773ced15f27dd27cd0b482a7473cdcf2a/include/uapi/linux/if.h#L234-L256
+class ifreq(ctypes.Structure):
+    _fields_ = [
+        ("ifr_name", ctypes.c_char * IFNAMSIZ),
+        ("ifr_ifru", ifr_ifru),
+    ]
+
+
+# ------------------------------------------------
+
+
+def _is_ethernet_interface(interface: str) -> bool:
+    """Check whether the given interface is a physical ethernet device
+
+    :param interface: interface name like enp1s1 or wlp194s0
+    :return: True if the interface is a real device
+             (not wifi, not loopback, not vpn, etc.)
+    """
+
+    try:
+        sys_class_net_interface = Path("/sys/class/net/") / interface
+
+        # check for ARPHRD_ETHER
+        if not (sys_class_net_interface / "type").read_text().strip() == "1":
+            return False
+        # skip everything not associated with a physical device
+        if not (sys_class_net_interface / "device").exists():
+            return False
+        # wifi interfaces (16.04+)
+        if (sys_class_net_interface / "phy80211").exists():
+            return False
+
+        return True
+    except OSError:
+        # assume false if failed to read any of the paths above
+        return False
+
+
+def get_ts_info(interface: str) -> ethtool_ts_info:
+    """
+    The main entry point of getting ts_info for the given interface
+
+    :param interface: interface name like enp1s1
+    :return: the ethtool_ts_info struct. The main attr we need is the phc_index
+             because it directly maps to /dev/ptpX
+    """
+    if not _is_ethernet_interface(interface):
+        logger.warning(
+            "{} is not a physical ethernet interface".format(interface)
+        )
+
+    info = ethtool_ts_info()
+    info.cmd = ETHTOOL_GET_TS_INFO
+
+    ifr = ifreq()
+    ifr.ifr_name = interface.encode()
+    ifr.ifr_ifru.ifr_data = ctypes.cast(ctypes.pointer(info), ctypes.c_void_p)
+
+    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
+        fcntl.ioctl(sock.fileno(), SIOCETHTOOL, ifr, True)
+
+    return info
+
+
+def is_ptp_capable(interface: str) -> bool:
+    """Checks if <interface> supports PTP
+
+    This is equivalent to checking the "PTP Hardware Clock" line in ethtool
+    and testing if the corresponding device node exists
+
+    :param interface: the interface to check
+    :return: if kernel says the interface supports ptp
+             AND The /dev/ptpX device exists
+    """
+    try:
+        info = get_ts_info(interface)
+        phc_index = int(info.phc_index)
+        expected_ptp_device_path = "/dev/ptp{}".format(phc_index)
+
+        return phc_index >= 0 and os.path.exists(expected_ptp_device_path)
+    except (OSError, ValueError):
+        return False