diff --git a/checkbox-support/checkbox_support/ethtool/__init__.py b/checkbox-support/checkbox_support/ethtool/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/checkbox-support/checkbox_support/ethtool/tests/__init__.py b/checkbox-support/checkbox_support/ethtool/tests/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/checkbox-support/checkbox_support/ethtool/tests/test_ts_info.py b/checkbox-support/checkbox_support/ethtool/tests/test_ts_info.py new file mode 100644 index 0000000000..28a2079c38 --- /dev/null +++ b/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 +# +# 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 . + +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() diff --git a/checkbox-support/checkbox_support/ethtool/ts_info.py b/checkbox-support/checkbox_support/ethtool/ts_info.py new file mode 100644 index 0000000000..f2b5040164 --- /dev/null +++ b/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 +# +# 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 . + +""" +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 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