1# SPDX-License-Identifier: GPL-2.0 2 3import ipaddress 4import os 5import sys 6import time 7import json 8from pathlib import Path 9from lib.py import KsftSkipEx, KsftXfailEx 10from lib.py import ksft_pr, ksft_setup, wait_file 11from lib.py import cmd, ethtool, ip, CmdExitFailure 12from lib.py import NetNS, NetdevSimDev, UserNetNS 13from .remote import Remote 14from . import bpftool, RtnlFamily, Netlink 15 16 17class NetDrvEnvBase: 18 """ 19 Base class for a NIC / host environments 20 21 Attributes: 22 test_dir: Path to the source directory of the test 23 net_lib_dir: Path to the net/lib directory 24 """ 25 def __init__(self, src_path): 26 self.src_path = Path(src_path) 27 self.test_dir = self.src_path.parent.resolve() 28 self.net_lib_dir = (Path(__file__).parent / "../../../../net/lib").resolve() 29 30 self.env = self._load_env_file() 31 32 # Following attrs must be set be inheriting classes 33 self.dev = None 34 self.ifname = None 35 36 def _load_env_file(self): 37 env = os.environ.copy() 38 39 src_dir = Path(self.src_path).parent.resolve() 40 if not (src_dir / "net.config").exists(): 41 return ksft_setup(env) 42 43 with open((src_dir / "net.config").as_posix(), 'r') as fp: 44 for line in fp.readlines(): 45 full_file = line 46 # Strip comments 47 pos = line.find("#") 48 if pos >= 0: 49 line = line[:pos] 50 line = line.strip() 51 if not line: 52 continue 53 pair = line.split('=', maxsplit=1) 54 if len(pair) != 2: 55 raise Exception("Can't parse configuration line:", full_file) 56 env[pair[0]] = pair[1] 57 return ksft_setup(env) 58 59 def __del__(self): 60 pass 61 62 def _print_dev_info(self): 63 """ 64 Show whether the test ran on real hardware or netdevsim. 65 Useful to confirm when results are shared on the mailing list. 66 """ 67 driver = "unknown" 68 try: 69 info = ethtool(f"-i {self.ifname}").stdout 70 for line in info.splitlines(): 71 if line.startswith("driver:"): 72 driver = line.split(':', 1)[1].strip() or driver 73 break 74 except (CmdExitFailure, FileNotFoundError): 75 pass 76 ksft_pr(f"Interface: {self.ifname}, driver: {driver}") 77 78 def __enter__(self): 79 ip(f"link set dev {self.dev['ifname']} up") 80 wait_file(f"/sys/class/net/{self.dev['ifname']}/carrier", 81 lambda x: x.strip() == "1") 82 83 return self 84 85 def __exit__(self, ex_type, ex_value, ex_tb): 86 """ 87 __exit__ gets called at the end of a "with" block. 88 """ 89 self.__del__() 90 91 92class NetDrvEnv(NetDrvEnvBase): 93 """ 94 Class for a single NIC / host env, with no remote end 95 """ 96 def __init__(self, src_path, nsim_test=None, **kwargs): 97 super().__init__(src_path) 98 99 self._ns = None 100 101 if 'NETIF' in self.env: 102 if nsim_test is True: 103 raise KsftXfailEx("Test only works on netdevsim") 104 105 self.dev = ip("-d link show dev " + self.env['NETIF'], json=True)[0] 106 else: 107 if nsim_test is False: 108 raise KsftXfailEx("Test does not work on netdevsim") 109 110 self._ns = NetdevSimDev(**kwargs) 111 self.dev = self._ns.nsims[0].dev 112 self.ifname = self.dev['ifname'] 113 self.ifindex = self.dev['ifindex'] 114 self._print_dev_info() 115 116 def __del__(self): 117 if self._ns: 118 self._ns.remove() 119 self._ns = None 120 121 122class NetDrvEpEnv(NetDrvEnvBase): 123 """ 124 Class for an environment with a local device and "remote endpoint" 125 which can be used to send traffic in. 126 127 For local testing it creates two network namespaces and a pair 128 of netdevsim devices. 129 """ 130 131 # Network prefixes used for local tests 132 nsim_v4_pfx = "192.0.2." 133 nsim_v6_pfx = "2001:db8::" 134 135 def __init__(self, src_path, nsim_test=None, queue_count=None): 136 super().__init__(src_path) 137 138 self._stats_settle_time = None 139 self._queue_count = queue_count 140 141 # Things we try to destroy 142 self.remote = None 143 # These are for local testing state 144 self._netns = None 145 self._ns = None 146 self._ns_peer = None 147 148 self.addr_v = { "4": None, "6": None } 149 self.remote_addr_v = { "4": None, "6": None } 150 151 if "NETIF" in self.env: 152 if nsim_test is True: 153 raise KsftXfailEx("Test only works on netdevsim") 154 self._check_env() 155 156 self.dev = ip("-d link show dev " + self.env['NETIF'], json=True)[0] 157 158 self.addr_v["4"] = self.env.get("LOCAL_V4") 159 self.addr_v["6"] = self.env.get("LOCAL_V6") 160 self.remote_addr_v["4"] = self.env.get("REMOTE_V4") 161 self.remote_addr_v["6"] = self.env.get("REMOTE_V6") 162 kind = self.env["REMOTE_TYPE"] 163 args = self.env["REMOTE_ARGS"] 164 else: 165 if nsim_test is False: 166 raise KsftXfailEx("Test does not work on netdevsim") 167 168 self.create_local() 169 170 self.dev = self._ns.nsims[0].dev 171 172 self.addr_v["4"] = self.nsim_v4_pfx + "1" 173 self.addr_v["6"] = self.nsim_v6_pfx + "1" 174 self.remote_addr_v["4"] = self.nsim_v4_pfx + "2" 175 self.remote_addr_v["6"] = self.nsim_v6_pfx + "2" 176 kind = "netns" 177 args = self._netns.name 178 179 self.remote = Remote(kind, args, src_path) 180 181 self.set_ipver("6" if self.addr_v["6"] else "4") 182 183 self.ifname = self.dev['ifname'] 184 self.ifindex = self.dev['ifindex'] 185 self._print_dev_info() 186 187 # resolve remote interface name 188 self.remote_ifname = self.resolve_remote_ifc() 189 self.remote_dev = ip("-d link show dev " + self.remote_ifname, 190 host=self.remote, json=True)[0] 191 self.remote_ifindex = self.remote_dev['ifindex'] 192 193 self._required_cmd = {} 194 195 def create_local(self): 196 nsim_kwargs = {} 197 if self._queue_count: 198 nsim_kwargs["queue_count"] = self._queue_count 199 200 self._netns = NetNS() 201 self._ns = NetdevSimDev(**nsim_kwargs) 202 self._ns_peer = NetdevSimDev(ns=self._netns, **nsim_kwargs) 203 204 with open("/proc/self/ns/net") as nsfd0, \ 205 open("/var/run/netns/" + self._netns.name) as nsfd1: 206 ifi0 = self._ns.nsims[0].ifindex 207 ifi1 = self._ns_peer.nsims[0].ifindex 208 NetdevSimDev.ctrl_write('link_device', 209 f'{nsfd0.fileno()}:{ifi0} {nsfd1.fileno()}:{ifi1}') 210 211 ip(f" addr add dev {self._ns.nsims[0].ifname} {self.nsim_v4_pfx}1/24") 212 ip(f"-6 addr add dev {self._ns.nsims[0].ifname} {self.nsim_v6_pfx}1/64 nodad") 213 ip(f" link set dev {self._ns.nsims[0].ifname} up") 214 215 ip(f" addr add dev {self._ns_peer.nsims[0].ifname} {self.nsim_v4_pfx}2/24", ns=self._netns) 216 ip(f"-6 addr add dev {self._ns_peer.nsims[0].ifname} {self.nsim_v6_pfx}2/64 nodad", ns=self._netns) 217 ip(f" link set dev {self._ns_peer.nsims[0].ifname} up", ns=self._netns) 218 219 def _check_env(self): 220 vars_needed = [ 221 ["LOCAL_V4", "LOCAL_V6"], 222 ["REMOTE_V4", "REMOTE_V6"], 223 ["REMOTE_TYPE"], 224 ["REMOTE_ARGS"] 225 ] 226 missing = [] 227 228 for choice in vars_needed: 229 for entry in choice: 230 if entry in self.env: 231 break 232 else: 233 missing.append(choice) 234 # Make sure v4 / v6 configs are symmetric 235 if ("LOCAL_V6" in self.env) != ("REMOTE_V6" in self.env): 236 missing.append(["LOCAL_V6", "REMOTE_V6"]) 237 if ("LOCAL_V4" in self.env) != ("REMOTE_V4" in self.env): 238 missing.append(["LOCAL_V4", "REMOTE_V4"]) 239 if missing: 240 raise Exception("Invalid environment, missing configuration:", missing, 241 "Please see tools/testing/selftests/drivers/net/README.rst") 242 243 def resolve_remote_ifc(self): 244 v4 = v6 = None 245 if self.remote_addr_v["4"]: 246 v4 = ip("addr show to " + self.remote_addr_v["4"], json=True, host=self.remote) 247 if self.remote_addr_v["6"]: 248 v6 = ip("addr show to " + self.remote_addr_v["6"], json=True, host=self.remote) 249 if v4 and v6 and v4[0]["ifname"] != v6[0]["ifname"]: 250 raise Exception("Can't resolve remote interface name, v4 and v6 don't match") 251 if (v4 and len(v4) > 1) or (v6 and len(v6) > 1): 252 raise Exception("Can't resolve remote interface name, multiple interfaces match") 253 return v6[0]["ifname"] if v6 else v4[0]["ifname"] 254 255 def __del__(self): 256 if self._ns: 257 self._ns.remove() 258 self._ns = None 259 if self._ns_peer: 260 self._ns_peer.remove() 261 self._ns_peer = None 262 if self._netns: 263 del self._netns 264 self._netns = None 265 if self.remote: 266 del self.remote 267 self.remote = None 268 269 def require_ipver(self, ipver): 270 if not self.addr_v[ipver] or not self.remote_addr_v[ipver]: 271 raise KsftSkipEx(f"Test requires IPv{ipver} connectivity") 272 273 def set_ipver(self, ipver): 274 """ 275 Modify the IP version used by the generic address fields. 276 """ 277 if ipver == getattr(self, "addr_ipver", None): 278 return 279 280 self.require_ipver(ipver) 281 282 self.addr_ipver = ipver 283 self.addr = self.addr_v[ipver] 284 self.remote_addr = self.remote_addr_v[ipver] 285 286 # Bracketed addresses, some commands need IPv6 to be inside [] 287 self.baddr = (f"[{self.addr_v['6']}]" if ipver == "6" 288 else self.addr_v["4"]) 289 self.remote_baddr = (f"[{self.remote_addr_v['6']}]" if ipver == "6" 290 else self.remote_addr_v["4"]) 291 292 def require_nsim(self, nsim_test=True): 293 """Require or exclude netdevsim for this test""" 294 if nsim_test and self._ns is None: 295 raise KsftXfailEx("Test only works on netdevsim") 296 if nsim_test is False and self._ns is not None: 297 raise KsftXfailEx("Test does not work on netdevsim") 298 299 def get_local_nsim_dev(self): 300 """Returns the local netdevsim device or None. 301 Using this method is discouraged, as it makes tests nsim-specific. 302 Standard interfaces available on all HW should ideally be used. 303 This method is intended for the few cases where nsim-specific 304 assertions need to be verified which cannot be verified otherwise. 305 """ 306 return self._ns 307 308 def _require_cmd(self, comm, key, host=None): 309 cached = self._required_cmd.get(comm, {}) 310 if cached.get(key) is None: 311 cached[key] = cmd("command -v -- " + comm, fail=False, 312 shell=True, host=host).ret == 0 313 self._required_cmd[comm] = cached 314 return cached[key] 315 316 def require_cmd(self, comm, local=True, remote=False): 317 if local: 318 if not self._require_cmd(comm, "local"): 319 raise KsftSkipEx("Test requires command: " + comm) 320 if remote: 321 if not self._require_cmd(comm, "remote", host=self.remote): 322 raise KsftSkipEx("Test requires (remote) command: " + comm) 323 324 def wait_hw_stats_settle(self): 325 """ 326 Wait for HW stats to become consistent, some devices DMA HW stats 327 periodically so events won't be reflected until next sync. 328 Good drivers will tell us via ethtool what their sync period is. 329 """ 330 if self._stats_settle_time is None: 331 data = {} 332 try: 333 data = ethtool("-c " + self.ifname, json=True)[0] 334 except CmdExitFailure as e: 335 if "Operation not supported" not in e.cmd.stderr: 336 raise 337 338 self._stats_settle_time = \ 339 1.25 * data.get('stats-block-usecs', 20000) / 1000 / 1000 340 341 time.sleep(self._stats_settle_time) 342 343 344class NetDrvContEnv(NetDrvEpEnv): 345 """ 346 Class for an environment with a netkit pair setup for forwarding traffic 347 between the physical interface and a network namespace. 348 NETIF = "eth0" 349 LOCAL_V6 = "2001:db8:1::1" 350 REMOTE_V6 = "2001:db8:1::2" 351 LOCAL_PREFIX_V6 = "2001:db8:2::0/64" 352 353 +-----------------------------+ +------------------------------+ 354 dst | INIT NS | | TEST NS | 355 2001: | +---------------+ | | | 356 db8:2::2| | NETIF | | bpf | | 357 +---|>| 2001:db8:1::1 | |redirect| +-------------------------+ | 358 | | | |-----------|--------|>| Netkit | | 359 | | +---------------+ | _peer | | nk_guest | | 360 | | +-------------+ Netkit pair | | | fe80::2/64 | | 361 | | | Netkit |.............|........|>| 2001:db8:2::2/64 | | 362 | | | nk_host | | | +-------------------------+ | 363 | | | fe80::1/64 | | | | 364 | | +-------------+ | | route: | 365 | | | | default | 366 | | route: | | via fe80::1 dev nk_guest | 367 | | 2001:db8:2::2/128 | +------------------------------+ 368 | | via fe80::2 dev nk_host | 369 | +-----------------------------+ 370 | 371 | +---------------+ 372 | | REMOTE | 373 +---| 2001:db8:1::2 | 374 +---------------+ 375 """ 376 377 def __init__(self, src_path, rxqueues=1, primary_rx_redirect=False, 378 userns=False, **kwargs): 379 self.netns = None 380 self._userns = userns 381 self.nk_host_ifname = None 382 self.nk_guest_ifname = None 383 self._tc_clsact_added = False 384 self._tc_attached = False 385 self._primary_rx_redirect_attached = False 386 self._primary_rx_redirect_clsact_added = False 387 self._bpf_prog_pref = None 388 self._bpf_prog_id = None 389 self._init_ns_attached = False 390 self._remote_route_added = False 391 self._old_fwd = None 392 self._old_accept_ra = None 393 394 super().__init__(src_path, **kwargs) 395 396 self.require_ipver("6") 397 local_prefix = self.env.get("LOCAL_PREFIX_V6") 398 if not local_prefix: 399 raise KsftSkipEx("LOCAL_PREFIX_V6 required") 400 401 net = ipaddress.IPv6Network(local_prefix, strict=False) 402 self.ipv6_prefix = str(net.network_address) 403 self.nk_host_ipv6 = f"{self.ipv6_prefix}2:1" 404 self.nk_guest_ipv6 = f"{self.ipv6_prefix}2:2" 405 406 local_v6 = ipaddress.IPv6Address(self.addr_v["6"]) 407 if local_v6 in net: 408 raise KsftSkipEx("LOCAL_V6 must not fall within LOCAL_PREFIX_V6") 409 410 rtnl = RtnlFamily() 411 rtnl.newlink( 412 { 413 "linkinfo": { 414 "kind": "netkit", 415 "data": { 416 "mode": "l2", 417 "policy": "forward", 418 "peer-policy": "forward", 419 }, 420 }, 421 "num-rx-queues": rxqueues, 422 }, 423 flags=[Netlink.NLM_F_CREATE, Netlink.NLM_F_EXCL], 424 ) 425 426 all_links = ip("-d link show", json=True) 427 netkit_links = [link for link in all_links 428 if link.get('linkinfo', {}).get('info_kind') == 'netkit' 429 and 'UP' not in link.get('flags', [])] 430 431 if len(netkit_links) != 2: 432 raise KsftSkipEx("Failed to create netkit pair") 433 434 netkit_links.sort(key=lambda x: x['ifindex']) 435 self.nk_host_ifname = netkit_links[1]['ifname'] 436 self.nk_guest_ifname = netkit_links[0]['ifname'] 437 self.nk_host_ifindex = netkit_links[1]['ifindex'] 438 self.nk_guest_ifindex = netkit_links[0]['ifindex'] 439 440 self._setup_ns() 441 self.attach_bpf() 442 if primary_rx_redirect: 443 self._attach_primary_rx_redirect_bpf() 444 445 def __del__(self): 446 if self._primary_rx_redirect_attached: 447 cmd(f"tc filter del dev {self.nk_host_ifname} ingress", fail=False) 448 self._primary_rx_redirect_attached = False 449 450 if self._primary_rx_redirect_clsact_added: 451 cmd(f"tc qdisc del dev {self.nk_host_ifname} clsact", fail=False) 452 self._primary_rx_redirect_clsact_added = False 453 454 if self._tc_attached: 455 cmd(f"tc filter del dev {self.ifname} ingress pref {self._bpf_prog_pref}") 456 self._tc_attached = False 457 458 if self._tc_clsact_added: 459 cmd(f"tc qdisc del dev {self.ifname} clsact") 460 self._tc_clsact_added = False 461 462 if self._remote_route_added: 463 cmd(f"ip -6 route del {self.nk_guest_ipv6}/128", 464 host=self.remote, fail=False) 465 self._remote_route_added = False 466 467 if self.nk_host_ifname: 468 cmd(f"ip link del dev {self.nk_host_ifname}") 469 self.nk_host_ifname = None 470 self.nk_guest_ifname = None 471 472 if self._init_ns_attached: 473 cmd("ip netns del init", fail=False) 474 self._init_ns_attached = False 475 476 if self.netns: 477 del self.netns 478 self.netns = None 479 480 if self._old_fwd is not None: 481 with open("/proc/sys/net/ipv6/conf/all/forwarding", "w", 482 encoding="utf-8") as f: 483 f.write(self._old_fwd) 484 self._old_fwd = None 485 if self._old_accept_ra is not None: 486 with open("/proc/sys/net/ipv6/conf/all/accept_ra", "w", 487 encoding="utf-8") as f: 488 f.write(self._old_accept_ra) 489 self._old_accept_ra = None 490 491 super().__del__() 492 493 def _setup_ns(self): 494 fwd_path = "/proc/sys/net/ipv6/conf/all/forwarding" 495 ra_path = "/proc/sys/net/ipv6/conf/all/accept_ra" 496 with open(fwd_path, encoding="utf-8") as f: 497 self._old_fwd = f.read().strip() 498 with open(ra_path, encoding="utf-8") as f: 499 self._old_accept_ra = f.read().strip() 500 with open(fwd_path, "w", encoding="utf-8") as f: 501 f.write("1") 502 with open(ra_path, "w", encoding="utf-8") as f: 503 f.write("2") 504 505 self.netns = UserNetNS() if self._userns else NetNS() 506 cmd("ip netns attach init 1") 507 self._init_ns_attached = True 508 ip("netns set init 0", ns=self.netns) 509 ip(f"link set dev {self.nk_guest_ifname} netns {self.netns.name}") 510 nk_guest_dev = ip(f"link show dev {self.nk_guest_ifname}", 511 json=True, ns=self.netns)[0] 512 self.nk_guest_ifindex = nk_guest_dev['ifindex'] 513 ip(f"link set dev {self.nk_host_ifname} up") 514 ip(f"-6 addr add fe80::1/64 dev {self.nk_host_ifname} nodad") 515 ip(f"-6 route add {self.nk_guest_ipv6}/128 via fe80::2 dev {self.nk_host_ifname}") 516 517 ip("link set lo up", ns=self.netns) 518 ip(f"link set dev {self.nk_guest_ifname} up", ns=self.netns) 519 ip(f"-6 addr add fe80::2/64 dev {self.nk_guest_ifname}", ns=self.netns) 520 ip(f"-6 addr add {self.nk_guest_ipv6}/64 dev {self.nk_guest_ifname} nodad", ns=self.netns) 521 ip(f"-6 route add default via fe80::1 dev {self.nk_guest_ifname}", ns=self.netns) 522 523 def _tc_ensure_clsact(self, ifname=None): 524 """Ensure a clsact qdisc exists on @ifname. 525 526 Returns True if this call added the qdisc, otherwise returns False. 527 """ 528 if ifname is None: 529 ifname = self.ifname 530 qdisc = json.loads(cmd(f"tc -j qdisc show dev {ifname}").stdout) 531 for q in qdisc: 532 if q['kind'] == 'clsact': 533 return False 534 cmd(f"tc qdisc add dev {ifname} clsact") 535 return True 536 537 def _get_bpf_prog_ids(self): 538 filters = json.loads(cmd(f"tc -j filter show dev {self.ifname} ingress").stdout) 539 for bpf in filters: 540 if 'options' not in bpf: 541 continue 542 if bpf['options']['bpf_name'].startswith('nk_forward.bpf'): 543 return (bpf['pref'], bpf['options']['prog']['id']) 544 raise Exception("Failed to get BPF prog ID") 545 546 def _find_bss_map_id(self, prog_id): 547 """Find the .bss map ID for a loaded BPF program.""" 548 prog_info = bpftool(f"prog show id {prog_id}", json=True) 549 for map_id in prog_info.get("map_ids", []): 550 map_info = bpftool(f"map show id {map_id}", json=True) 551 if map_info.get("name", "").endswith("bss"): 552 return map_id 553 raise Exception(f"Failed to find .bss map for prog {prog_id}") 554 555 def _find_bpf_obj(self, name): 556 bpf_obj = self.test_dir / name 557 if bpf_obj.exists(): 558 return bpf_obj 559 bpf_obj = self.test_dir / "hw" / name 560 if bpf_obj.exists(): 561 return bpf_obj 562 return None 563 564 def detach_bpf(self): 565 if self._tc_attached: 566 cmd(f"tc filter del dev {self.ifname} ingress pref " 567 f"{self._bpf_prog_pref}", fail=False) 568 self._tc_attached = False 569 570 def attach_bpf(self): 571 bpf_obj = self._find_bpf_obj("nk_forward.bpf.o") 572 if not bpf_obj: 573 raise KsftSkipEx("BPF prog nk_forward.bpf.o not found") 574 575 if self._tc_ensure_clsact(): 576 self._tc_clsact_added = True 577 cmd(f"tc filter add dev {self.ifname} ingress bpf obj {bpf_obj}" 578 " sec tc/ingress direct-action") 579 self._tc_attached = True 580 581 (self._bpf_prog_pref, self._bpf_prog_id) = self._get_bpf_prog_ids() 582 bss_map_id = self._find_bss_map_id(self._bpf_prog_id) 583 584 ipv6_addr = ipaddress.IPv6Address(self.ipv6_prefix) 585 ipv6_bytes = ipv6_addr.packed 586 ifindex_bytes = self.nk_host_ifindex.to_bytes(4, byteorder='little') 587 value = ipv6_bytes + ifindex_bytes 588 value_hex = ' '.join(f'{b:02x}' for b in value) 589 bpftool(f"map update id {bss_map_id} key hex 00 00 00 00 value hex {value_hex}") 590 591 def _attach_primary_rx_redirect_bpf(self): 592 """Attach BPF redirect program on the primary netkit ingress.""" 593 bpf_obj = self._find_bpf_obj("nk_primary_rx_redirect.bpf.o") 594 if not bpf_obj: 595 raise KsftSkipEx("nk_primary_rx_redirect.bpf.o not found") 596 597 if self._tc_ensure_clsact(self.nk_host_ifname): 598 self._primary_rx_redirect_clsact_added = True 599 cmd(f"tc filter add dev {self.nk_host_ifname} ingress" 600 f" bpf obj {bpf_obj} sec tc/ingress direct-action") 601 self._primary_rx_redirect_attached = True 602 603 ip(f"-6 route add {self.nk_guest_ipv6}/128 via {self.addr_v['6']}", 604 host=self.remote) 605 self._remote_route_added = True 606 607 filters = json.loads( 608 cmd(f"tc -j filter show dev {self.nk_host_ifname} ingress").stdout) 609 redirect_prog_id = None 610 for bpf in filters: 611 if 'options' not in bpf: 612 continue 613 if bpf['options']['bpf_name'].startswith('nk_primary_rx_redirect'): 614 redirect_prog_id = bpf['options']['prog']['id'] 615 break 616 if redirect_prog_id is None: 617 raise Exception("Failed to get primary RX redirect BPF prog ID") 618 619 bss_map_id = self._find_bss_map_id(redirect_prog_id) 620 phys_ifindex_bytes = self.ifindex.to_bytes(4, byteorder=sys.byteorder) 621 value_hex = ' '.join(f'{b:02x}' for b in phys_ifindex_bytes) 622 bpftool(f"map update id {bss_map_id} key hex 00 00 00 00 value hex {value_hex}") 623