#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-2.0 """Test suite for PSP capable drivers.""" import errno import fcntl import os import socket import struct import termios import time from lib.py import defer from lib.py import ksft_run, ksft_exit, ksft_pr from lib.py import ksft_true, ksft_eq, ksft_ne, ksft_gt, ksft_raises from lib.py import ksft_not_none from lib.py import ksft_variants, KsftNamedVariant from lib.py import KsftSkipEx, KsftFailEx from lib.py import NetDrvEpEnv, NetDrvContEnv from lib.py import Netlink, NlError, PSPFamily, RtnlFamily from lib.py import NetNSEnter from lib.py import bkg, rand_port, wait_port_listen from lib.py import ip def _get_outq(s): one = b'\0' * 4 outq = fcntl.ioctl(s.fileno(), termios.TIOCOUTQ, one) return struct.unpack("I", outq)[0] def _send_with_ack(cfg, msg): cfg.comm_sock.send(msg) response = cfg.comm_sock.recv(4) if response != b'ack\0': raise RuntimeError("Unexpected server response", response) def _remote_read_len(cfg): cfg.comm_sock.send(b'read len\0') return int(cfg.comm_sock.recv(1024)[:-1].decode('utf-8')) def _make_clr_conn(cfg, ipver=None): _send_with_ack(cfg, b'conn clr\0') remote_addr = cfg.remote_addr_v[ipver] if ipver else cfg.remote_addr s = socket.create_connection((remote_addr, cfg.comm_port), ) return s def _make_psp_conn(cfg, version=0, ipver=None): _send_with_ack(cfg, b'conn psp\0' + struct.pack('BB', version, version)) remote_addr = cfg.remote_addr_v[ipver] if ipver else cfg.remote_addr s = socket.create_connection((remote_addr, cfg.comm_port), ) return s def _close_conn(cfg, s): _send_with_ack(cfg, b'data close\0') s.close() def _close_psp_conn(cfg, s): _close_conn(cfg, s) def _spi_xchg(s, rx): s.send(struct.pack('I', rx['spi']) + rx['key']) tx = s.recv(4 + len(rx['key'])) return { 'spi': struct.unpack('I', tx[:4])[0], 'key': tx[4:] } def _send_careful(cfg, s, rounds): data = b'0123456789' * 200 for i in range(rounds): n = 0 for _ in range(10): # allow 10 retries try: n += s.send(data[n:], socket.MSG_DONTWAIT) if n == len(data): break except BlockingIOError: time.sleep(0.05) else: rlen = _remote_read_len(cfg) outq = _get_outq(s) report = f'sent: {i * len(data) + n} remote len: {rlen} outq: {outq}' raise RuntimeError(report) return len(data) * rounds def _check_data_rx(cfg, exp_len): read_len = -1 for _ in range(30): cfg.comm_sock.send(b'read len\0') read_len = int(cfg.comm_sock.recv(1024)[:-1].decode('utf-8')) if read_len == exp_len: break time.sleep(0.01) ksft_eq(read_len, exp_len) def _check_data_outq(s, exp_len, force_wait=False): outq = 0 for _ in range(10): outq = _get_outq(s) if not force_wait and outq == exp_len: break time.sleep(0.01) ksft_eq(outq, exp_len) def _get_stat(cfg, key): return cfg.pspnl.get_stats({'dev-id': cfg.psp_dev_id})[key] # # Test case boiler plate # def _init_psp_dev(cfg, use_psp_ifindex=False): if not hasattr(cfg, 'psp_dev_id'): # Figure out which local device we are testing against # For NetDrvContEnv: use psp_ifindex instead of ifindex target_ifindex = cfg.psp_ifindex if use_psp_ifindex else cfg.ifindex for dev in cfg.pspnl.dev_get({}, dump=True): if dev['ifindex'] == target_ifindex: cfg.psp_info = dev cfg.psp_dev_id = cfg.psp_info['id'] break else: raise KsftSkipEx("No PSP devices found") # Enable PSP if necessary cap = cfg.psp_info['psp-versions-cap'] ena = cfg.psp_info['psp-versions-ena'] if cap != ena: cfg.pspnl.dev_set({'id': cfg.psp_dev_id, 'psp-versions-ena': cap}) defer(cfg.pspnl.dev_set, {'id': cfg.psp_dev_id, 'psp-versions-ena': ena }) # # Test cases # def dev_list_devices(cfg): """ Dump all devices """ _init_psp_dev(cfg) devices = cfg.pspnl.dev_get({}, dump=True) found = False for dev in devices: found |= dev['id'] == cfg.psp_dev_id ksft_true(found) def dev_get_device(cfg): """ Get the device we intend to use """ _init_psp_dev(cfg) dev = cfg.pspnl.dev_get({'id': cfg.psp_dev_id}) ksft_eq(dev['id'], cfg.psp_dev_id) def dev_get_device_bad(cfg): """ Test getting device which doesn't exist """ raised = False try: cfg.pspnl.dev_get({'id': 1234567}) except NlError as e: ksft_eq(e.nl_msg.error, -errno.ENODEV) raised = True ksft_true(raised) def dev_rotate(cfg): """ Test key rotation """ _init_psp_dev(cfg) prev_rotations = _get_stat(cfg, 'key-rotations') rot = cfg.pspnl.key_rotate({"id": cfg.psp_dev_id}) ksft_eq(rot['id'], cfg.psp_dev_id) rot = cfg.pspnl.key_rotate({"id": cfg.psp_dev_id}) ksft_eq(rot['id'], cfg.psp_dev_id) cur_rotations = _get_stat(cfg, 'key-rotations') ksft_eq(cur_rotations, prev_rotations + 2) def dev_rotate_spi(cfg): """ Test key rotation and SPI check """ _init_psp_dev(cfg) top_a = top_b = 0 with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s: assoc_a = cfg.pspnl.rx_assoc({"version": 0, "dev-id": cfg.psp_dev_id, "sock-fd": s.fileno()}) top_a = assoc_a['rx-key']['spi'] >> 31 s.close() rot = cfg.pspnl.key_rotate({"id": cfg.psp_dev_id}) with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s: ksft_eq(rot['id'], cfg.psp_dev_id) assoc_b = cfg.pspnl.rx_assoc({"version": 0, "dev-id": cfg.psp_dev_id, "sock-fd": s.fileno()}) top_b = assoc_b['rx-key']['spi'] >> 31 s.close() ksft_ne(top_a, top_b) def assoc_basic(cfg): """ Test creating associations """ _init_psp_dev(cfg) with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s: assoc = cfg.pspnl.rx_assoc({"version": 0, "dev-id": cfg.psp_dev_id, "sock-fd": s.fileno()}) ksft_eq(assoc['dev-id'], cfg.psp_dev_id) ksft_gt(assoc['rx-key']['spi'], 0) ksft_eq(len(assoc['rx-key']['key']), 16) assoc = cfg.pspnl.tx_assoc({"dev-id": cfg.psp_dev_id, "version": 0, "tx-key": assoc['rx-key'], "sock-fd": s.fileno()}) ksft_eq(len(assoc), 0) s.close() def assoc_bad_dev(cfg): """ Test creating associations with bad device ID """ _init_psp_dev(cfg) with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s: with ksft_raises(NlError) as cm: cfg.pspnl.rx_assoc({"version": 0, "dev-id": cfg.psp_dev_id + 1234567, "sock-fd": s.fileno()}) ksft_eq(cm.exception.nl_msg.error, -errno.ENODEV) def assoc_sk_only_conn(cfg): """ Test creating associations based on socket """ _init_psp_dev(cfg) with _make_clr_conn(cfg) as s: assoc = cfg.pspnl.rx_assoc({"version": 0, "sock-fd": s.fileno()}) ksft_eq(assoc['dev-id'], cfg.psp_dev_id) cfg.pspnl.tx_assoc({"version": 0, "tx-key": assoc['rx-key'], "sock-fd": s.fileno()}) _close_conn(cfg, s) def assoc_sk_only_mismatch(cfg): """ Test creating associations based on socket (dev mismatch) """ _init_psp_dev(cfg) with _make_clr_conn(cfg) as s: with ksft_raises(NlError) as cm: cfg.pspnl.rx_assoc({"version": 0, "dev-id": cfg.psp_dev_id + 1234567, "sock-fd": s.fileno()}) the_exception = cm.exception ksft_eq(the_exception.nl_msg.extack['bad-attr'], ".dev-id") ksft_eq(the_exception.nl_msg.error, -errno.EINVAL) _close_conn(cfg, s) def assoc_sk_only_mismatch_tx(cfg): """ Test creating associations based on socket (dev mismatch) """ _init_psp_dev(cfg) with _make_clr_conn(cfg) as s: with ksft_raises(NlError) as cm: assoc = cfg.pspnl.rx_assoc({"version": 0, "sock-fd": s.fileno()}) cfg.pspnl.tx_assoc({"version": 0, "tx-key": assoc['rx-key'], "dev-id": cfg.psp_dev_id + 1234567, "sock-fd": s.fileno()}) the_exception = cm.exception ksft_eq(the_exception.nl_msg.extack['bad-attr'], ".dev-id") ksft_eq(the_exception.nl_msg.error, -errno.EINVAL) _close_conn(cfg, s) def assoc_sk_only_unconn(cfg): """ Test creating associations based on socket (unconnected, should fail) """ _init_psp_dev(cfg) with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s: with ksft_raises(NlError) as cm: cfg.pspnl.rx_assoc({"version": 0, "sock-fd": s.fileno()}) the_exception = cm.exception ksft_eq(the_exception.nl_msg.extack['miss-type'], "dev-id") ksft_eq(the_exception.nl_msg.error, -errno.EINVAL) def assoc_version_mismatch(cfg): """ Test creating associations where Rx and Tx PSP versions do not match """ _init_psp_dev(cfg) versions = list(cfg.psp_info['psp-versions-cap']) if len(versions) < 2: raise KsftSkipEx("Not enough PSP versions supported by the device for the test") # Translate versions to integers versions = [cfg.pspnl.consts["version"].entries[v].value for v in versions] with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s: rx = cfg.pspnl.rx_assoc({"version": versions[0], "dev-id": cfg.psp_dev_id, "sock-fd": s.fileno()}) for version in versions[1:]: with ksft_raises(NlError) as cm: cfg.pspnl.tx_assoc({"dev-id": cfg.psp_dev_id, "version": version, "tx-key": rx['rx-key'], "sock-fd": s.fileno()}) the_exception = cm.exception ksft_eq(the_exception.nl_msg.error, -errno.EINVAL) def assoc_twice(cfg): """ Test reusing Tx assoc for two sockets """ _init_psp_dev(cfg) def rx_assoc_check(s): assoc = cfg.pspnl.rx_assoc({"version": 0, "dev-id": cfg.psp_dev_id, "sock-fd": s.fileno()}) ksft_eq(assoc['dev-id'], cfg.psp_dev_id) ksft_gt(assoc['rx-key']['spi'], 0) ksft_eq(len(assoc['rx-key']['key']), 16) return assoc with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s: assoc = rx_assoc_check(s) tx = cfg.pspnl.tx_assoc({"dev-id": cfg.psp_dev_id, "version": 0, "tx-key": assoc['rx-key'], "sock-fd": s.fileno()}) ksft_eq(len(tx), 0) # Use the same Tx assoc second time with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s2: rx_assoc_check(s2) tx = cfg.pspnl.tx_assoc({"dev-id": cfg.psp_dev_id, "version": 0, "tx-key": assoc['rx-key'], "sock-fd": s2.fileno()}) ksft_eq(len(tx), 0) s.close() def _data_basic_send(cfg, version, ipver): """ Test basic data send """ _init_psp_dev(cfg) # Version 0 is required by spec, don't let it skip if version: name = cfg.pspnl.consts["version"].entries_by_val[version].name if name not in cfg.psp_info['psp-versions-cap']: with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s: with ksft_raises(NlError) as cm: cfg.pspnl.rx_assoc({"version": version, "dev-id": cfg.psp_dev_id, "sock-fd": s.fileno()}) ksft_eq(cm.exception.nl_msg.error, -errno.EOPNOTSUPP) raise KsftSkipEx("PSP version not supported", name) s = _make_psp_conn(cfg, version, ipver) rx_assoc = cfg.pspnl.rx_assoc({"version": version, "dev-id": cfg.psp_dev_id, "sock-fd": s.fileno()}) rx = rx_assoc['rx-key'] tx = _spi_xchg(s, rx) cfg.pspnl.tx_assoc({"dev-id": cfg.psp_dev_id, "version": version, "tx-key": tx, "sock-fd": s.fileno()}) data_len = _send_careful(cfg, s, 100) _check_data_rx(cfg, data_len) _close_psp_conn(cfg, s) def __bad_xfer_do(cfg, s, tx, version='hdr0-aes-gcm-128'): # Make sure we accept the ACK for the SPI before we seal with the bad assoc _check_data_outq(s, 0) cfg.pspnl.tx_assoc({"dev-id": cfg.psp_dev_id, "version": version, "tx-key": tx, "sock-fd": s.fileno()}) data_len = _send_careful(cfg, s, 20) _check_data_outq(s, data_len, force_wait=True) _check_data_rx(cfg, 0) _close_psp_conn(cfg, s) def data_send_bad_key(cfg): """ Test send data with bad key """ _init_psp_dev(cfg) s = _make_psp_conn(cfg) rx_assoc = cfg.pspnl.rx_assoc({"version": 0, "dev-id": cfg.psp_dev_id, "sock-fd": s.fileno()}) rx = rx_assoc['rx-key'] tx = _spi_xchg(s, rx) tx['key'] = (tx['key'][0] ^ 0xff).to_bytes(1, 'little') + tx['key'][1:] __bad_xfer_do(cfg, s, tx) def data_send_disconnect(cfg): """ Test socket close after sending data """ _init_psp_dev(cfg) with _make_psp_conn(cfg) as s: assoc = cfg.pspnl.rx_assoc({"version": 0, "sock-fd": s.fileno()}) tx = _spi_xchg(s, assoc['rx-key']) cfg.pspnl.tx_assoc({"version": 0, "tx-key": tx, "sock-fd": s.fileno()}) data_len = _send_careful(cfg, s, 100) _check_data_rx(cfg, data_len) s.shutdown(socket.SHUT_RDWR) s.close() def _data_mss_adjust(cfg, ipver): _init_psp_dev(cfg) # First figure out what the MSS would be without any adjustments s = _make_clr_conn(cfg, ipver) s.send(b"0123456789abcdef" * 1024) _check_data_rx(cfg, 16 * 1024) mss = s.getsockopt(socket.IPPROTO_TCP, socket.TCP_MAXSEG) _close_conn(cfg, s) s = _make_psp_conn(cfg, 0, ipver) try: rx_assoc = cfg.pspnl.rx_assoc({"version": 0, "dev-id": cfg.psp_dev_id, "sock-fd": s.fileno()}) rx = rx_assoc['rx-key'] tx = _spi_xchg(s, rx) rxmss = s.getsockopt(socket.IPPROTO_TCP, socket.TCP_MAXSEG) ksft_eq(mss, rxmss) cfg.pspnl.tx_assoc({"dev-id": cfg.psp_dev_id, "version": 0, "tx-key": tx, "sock-fd": s.fileno()}) txmss = s.getsockopt(socket.IPPROTO_TCP, socket.TCP_MAXSEG) ksft_eq(mss, txmss + 40) data_len = _send_careful(cfg, s, 100) _check_data_rx(cfg, data_len) _check_data_outq(s, 0) txmss = s.getsockopt(socket.IPPROTO_TCP, socket.TCP_MAXSEG) ksft_eq(mss, txmss + 40) finally: _close_psp_conn(cfg, s) def data_stale_key(cfg): """ Test send on a double-rotated key """ _init_psp_dev(cfg) prev_stale = _get_stat(cfg, 'stale-events') s = _make_psp_conn(cfg) try: rx_assoc = cfg.pspnl.rx_assoc({"version": 0, "dev-id": cfg.psp_dev_id, "sock-fd": s.fileno()}) rx = rx_assoc['rx-key'] tx = _spi_xchg(s, rx) cfg.pspnl.tx_assoc({"dev-id": cfg.psp_dev_id, "version": 0, "tx-key": tx, "sock-fd": s.fileno()}) data_len = _send_careful(cfg, s, 100) _check_data_rx(cfg, data_len) _check_data_outq(s, 0) cfg.pspnl.key_rotate({"id": cfg.psp_dev_id}) cfg.pspnl.key_rotate({"id": cfg.psp_dev_id}) cur_stale = _get_stat(cfg, 'stale-events') ksft_gt(cur_stale, prev_stale) s.send(b'0123456789' * 200) _check_data_outq(s, 2000, force_wait=True) finally: _close_psp_conn(cfg, s) def __nsim_psp_rereg(cfg): # The PSP dev ID will change, remember what was there before before = set([x['id'] for x in cfg.pspnl.dev_get({}, dump=True)]) cfg._ns.nsims[0].dfs_write('psp_rereg', '1') after = set([x['id'] for x in cfg.pspnl.dev_get({}, dump=True)]) new_devs = list(after - before) ksft_eq(len(new_devs), 1) cfg.psp_dev_id = list(after - before)[0] def removal_device_rx(cfg): """ Test removing a netdev / PSD with active Rx assoc """ # We could technically devlink reload real devices, too # but that kills the control socket. So test this on # netdevsim only for now cfg.require_nsim() s = _make_clr_conn(cfg) try: rx_assoc = cfg.pspnl.rx_assoc({"version": 0, "dev-id": cfg.psp_dev_id, "sock-fd": s.fileno()}) ksft_not_none(rx_assoc) __nsim_psp_rereg(cfg) finally: _close_conn(cfg, s) def removal_device_bi(cfg): """ Test removing a netdev / PSD with active Rx/Tx assoc """ # We could technically devlink reload real devices, too # but that kills the control socket. So test this on # netdevsim only for now cfg.require_nsim() s = _make_clr_conn(cfg) try: rx_assoc = cfg.pspnl.rx_assoc({"version": 0, "dev-id": cfg.psp_dev_id, "sock-fd": s.fileno()}) cfg.pspnl.tx_assoc({"dev-id": cfg.psp_dev_id, "version": 0, "tx-key": rx_assoc['rx-key'], "sock-fd": s.fileno()}) __nsim_psp_rereg(cfg) finally: _close_conn(cfg, s) def _get_psp_ver_ip_variants(): for ver in range(4): for ipv in ("4", "6"): yield KsftNamedVariant(f"v{ver}_ip{ipv}", ver, ipv) def _get_ip_variants(): for ipv in ("4", "6"): yield KsftNamedVariant(f"ip{ipv}", ipv) @ksft_variants(_get_psp_ver_ip_variants()) def data_basic_send(cfg, version, ipver): """Test basic PSP data send.""" cfg.require_ipver(ipver) _data_basic_send(cfg, version, ipver) @ksft_variants(_get_ip_variants()) def data_mss_adjust(cfg, ipver): """Test MSS adjustment with PSP.""" cfg.require_ipver(ipver) _data_mss_adjust(cfg, ipver) def _check_assoc_list(cfg, psp_dev_id, ifindex, nsid=None): """Verify assoc-list contains device with given ifindex, no duplicates.""" dev_info = cfg.pspnl.dev_get({'id': psp_dev_id}) ksft_true('assoc-list' in dev_info, "No assoc-list in dev_get() response after association") found = False for assoc in dev_info['assoc-list']: if assoc['ifindex'] != ifindex: continue if nsid is not None and assoc['nsid'] != nsid: continue ksft_eq(found, False, "Duplicate assoc entry found") found = True ksft_eq(found, True, "Associated device not found in dev_get() response") def _data_basic_send_netkit_psp_assoc(cfg, version, ipver): """ Test basic data send with netkit interface associated with PSP dev. """ _assoc_nk_guest(cfg) # Enter guest namespace (netns) to run PSP test with NetNSEnter(cfg.netns.name): cfg.pspnl = PSPFamily() sock = _make_psp_conn(cfg, version, ipver) rx_assoc = cfg.pspnl.rx_assoc({"version": version, "dev-id": cfg.psp_dev_id, "sock-fd": sock.fileno()}) rx_key = rx_assoc['rx-key'] tx_key = _spi_xchg(sock, rx_key) cfg.pspnl.tx_assoc({"dev-id": cfg.psp_dev_id, "version": version, "tx-key": tx_key, "sock-fd": sock.fileno()}) data_len = _send_careful(cfg, sock, 100) _check_data_rx(cfg, data_len) _close_psp_conn(cfg, sock) def _assoc_check_list(cfg): """Test that assoc-list is correctly populated after dev-assoc.""" _assoc_nk_guest(cfg) _check_assoc_list(cfg, cfg.psp_dev_id, cfg.nk_guest_ifindex, cfg.psp_dev_peer_nsid) def _get_psp_ver_ip6_variants(): for ver in range(4): yield KsftNamedVariant(f"v{ver}_ip6", ver, "6") @ksft_variants(_get_psp_ver_ip6_variants()) def data_basic_send_netkit_psp_assoc(cfg, version, ipver): """Test PSP data send via netkit with dev-assoc.""" cfg.require_ipver(ipver) _data_basic_send_netkit_psp_assoc(cfg, version, ipver) def _key_rotation_notify_multi_ns_netkit(cfg): """ Test key rotation notifications across multiple namespaces using netkit """ _assoc_nk_guest(cfg) # Create listener in guest namespace; socket stays bound to that ns with NetNSEnter(cfg.netns.name): peer_pspnl = PSPFamily() peer_pspnl.ntf_subscribe('use') # Create listener in main namespace main_pspnl = PSPFamily() main_pspnl.ntf_subscribe('use') # Trigger key rotation on the PSP device cfg.pspnl.key_rotate({"id": cfg.psp_dev_id}) # Poll both sockets from main thread for pspnl, label in [(main_pspnl, "main"), (peer_pspnl, "guest")]: for ntf in pspnl.poll_ntf(duration=10): if ntf['msg'].get('id') == cfg.psp_dev_id: break else: raise KsftFailEx( f"No key rotation notification received" f" in {label} namespace") def _dev_change_notify_multi_ns_netkit(cfg): """ Test dev_change notifications across multiple namespaces using netkit """ _assoc_nk_guest(cfg) # Create listener in guest namespace; socket stays bound to that ns with NetNSEnter(cfg.netns.name): peer_pspnl = PSPFamily() peer_pspnl.ntf_subscribe('mgmt') # Create listener in main namespace main_pspnl = PSPFamily() main_pspnl.ntf_subscribe('mgmt') # Trigger dev_change by calling dev_set (notification is always sent) cfg.pspnl.dev_set({'id': cfg.psp_dev_id, 'psp-versions-ena': cfg.psp_info['psp-versions-cap']}) # Poll both sockets from main thread for pspnl, label in [(main_pspnl, "main"), (peer_pspnl, "guest")]: for ntf in pspnl.poll_ntf(duration=10): if ntf['msg'].get('id') == cfg.psp_dev_id: break else: raise KsftFailEx( f"No dev_change notification received" f" in {label} namespace") def _psp_dev_get_check_netkit_psp_assoc(cfg): """ Check psp dev-get output with netkit interface associated with PSP dev """ _assoc_nk_guest(cfg) # Check 1: In default netns, verify dev-get has correct ifindex and assoc-list dev_info = cfg.pspnl.dev_get({'id': cfg.psp_dev_id}) ksft_eq(dev_info['ifindex'], cfg.psp_ifindex) _check_assoc_list(cfg, cfg.psp_dev_id, cfg.nk_guest_ifindex, cfg.psp_dev_peer_nsid) # Check 2: In guest netns, verify dev-get has assoc-list with nk_guest device with NetNSEnter(cfg.netns.name): peer_pspnl = PSPFamily() # Dump all devices in the guest namespace peer_devices = peer_pspnl.dev_get({}, dump=True) # Find the device with by-association flag peer_dev = None for dev in peer_devices: if dev.get('by-association'): peer_dev = dev break ksft_not_none(peer_dev, "No PSP device found with by-association flag in guest netns") # Verify assoc-list contains the nk_guest device ksft_true('assoc-list' in peer_dev and len(peer_dev['assoc-list']) > 0, "Guest device should have assoc-list with local devices") # Verify the assoc-list contains nk_guest ifindex with nsid=-1 (same namespace) found = False for assoc in peer_dev['assoc-list']: if assoc['ifindex'] == cfg.nk_guest_ifindex: ksft_eq(assoc['nsid'], -1, "nsid should be -1 (NETNSA_NSID_NOT_ASSIGNED) for same-namespace device") found = True break ksft_true(found, "nk_guest ifindex not found in assoc-list") def _dev_assoc_no_nsid(cfg): """ Test dev-assoc and dev-disassoc without nsid attribute """ _init_psp_dev(cfg, True) # Associate without nsid - should look up ifindex in caller's netns cfg.pspnl.dev_assoc({'id': cfg.psp_dev_id, 'ifindex': cfg.nk_host_ifindex}) defer(_try_disassoc, cfg, cfg.psp_dev_id, cfg.nk_host_ifindex) defer(delattr, cfg, 'psp_dev_id') defer(delattr, cfg, 'psp_info') # Verify assoc-list contains the device (match by ifindex only) _check_assoc_list(cfg, cfg.psp_dev_id, cfg.nk_host_ifindex) # Disassociate without nsid - should also use caller's netns cfg.pspnl.dev_disassoc({'id': cfg.psp_dev_id, 'ifindex': cfg.nk_host_ifindex}) # Verify assoc-list no longer contains the device dev_info = cfg.pspnl.dev_get({'id': cfg.psp_dev_id}) found = False if 'assoc-list' in dev_info: for assoc in dev_info['assoc-list']: if assoc['ifindex'] == cfg.nk_host_ifindex: found = True break ksft_true(not found, "Device should not be in assoc-list after disassociation") def _psp_dev_assoc_cleanup_on_netkit_del(cfg): """Test that assoc-list is cleared when associated netkit is deleted. Creates a disposable netkit pair for this test to avoid destroying the shared environment. """ _init_psp_dev(cfg, True) defer(delattr, cfg, 'psp_dev_id') defer(delattr, cfg, 'psp_info') existing = {cfg.nk_host_ifindex, cfg.nk_guest_ifindex} # Create a temporary netkit pair tmp_host_name = "tmp_nk_host" tmp_guest_name = "tmp_nk_guest" rtnl = RtnlFamily() rtnl.newlink( { "ifname": tmp_host_name, "linkinfo": { "kind": "netkit", "data": { "mode": "l2", "policy": "forward", "peer-policy": "forward", }, }, }, flags=[Netlink.NLM_F_CREATE, Netlink.NLM_F_EXCL], ) cleanup_netkit = defer(ip, f"link del {tmp_host_name}") # Find the peer by diffing against existing netkit ifindexes all_links = ip("-d link show", json=True) tmp_peer = [link for link in all_links if link.get('linkinfo', {}).get('info_kind') == 'netkit' and link['ifindex'] not in existing and link['ifname'] != tmp_host_name] ksft_eq(len(tmp_peer), 1, "Failed to find temporary netkit peer") guest_name = tmp_peer[0]['ifname'] # Rename and move guest end into the test namespace ip(f"link set dev {guest_name} name {tmp_guest_name}") ip(f"link set dev {tmp_guest_name} netns {cfg.netns.name}") tmp_guest_dev = ip(f"link show dev {tmp_guest_name}", json=True, ns=cfg.netns)[0] tmp_guest_ifindex = tmp_guest_dev['ifindex'] ip(f"link set dev {tmp_guest_name} up", ns=cfg.netns) # Associate PSP device with the temporary guest interface cfg.pspnl.dev_assoc({'id': cfg.psp_dev_id, 'ifindex': tmp_guest_ifindex, 'nsid': cfg.psp_dev_peer_nsid}) # Verify assoc-list contains the temporary device _check_assoc_list(cfg, cfg.psp_dev_id, tmp_guest_ifindex, cfg.psp_dev_peer_nsid) # Delete the temporary netkit pair (deleting one end removes both) ip(f"link del {tmp_host_name}") cleanup_netkit.cancel() # Verify assoc-list is cleared after netkit deletion dev_info = cfg.pspnl.dev_get({'id': cfg.psp_dev_id}) ksft_true('assoc-list' not in dev_info or len(dev_info['assoc-list']) == 0, "assoc-list should be empty after netkit deletion") def _try_disassoc(cfg, psp_dev_id, ifindex, nsid=None): """Best-effort disassociate, ignoring errors if already removed.""" try: params = {'id': psp_dev_id, 'ifindex': ifindex} if nsid is not None: params['nsid'] = nsid cfg.pspnl.dev_disassoc(params) except NlError: pass def _assoc_nk_guest(cfg): """Associate nk_guest with PSP device and register cleanup via defer().""" _init_psp_dev(cfg, True) cfg.pspnl.dev_assoc({'id': cfg.psp_dev_id, 'ifindex': cfg.nk_guest_ifindex, 'nsid': cfg.psp_dev_peer_nsid}) defer(_disassoc_nk_guest, cfg, cfg.psp_dev_id, cfg.nk_guest_ifindex) def _disassoc_nk_guest(cfg, psp_dev_id, nk_guest_ifindex): """Disassociate nk_guest and reset cfg PSP state.""" pspnl = PSPFamily() pspnl.dev_disassoc({'id': psp_dev_id, 'ifindex': nk_guest_ifindex, 'nsid': cfg.psp_dev_peer_nsid}) cfg.pspnl = pspnl del cfg.psp_dev_id del cfg.psp_info def _get_nsid(ns_name): """Get the nsid for a namespace.""" for entry in ip("netns list-id", json=True): if entry.get("name") == str(ns_name): return entry["nsid"] raise KsftSkipEx(f"nsid not found for namespace {ns_name}") def _setup_psp_attributes(cfg): # pylint: disable=protected-access """ Set up PSP-specific attributes on the environment. This sets attributes needed for PSP tests based on whether we're using netdevsim or a real NIC. """ if cfg._ns is not None: # netdevsim case: PSP device is the local dev (in host namespace) cfg.psp_dev = cfg._ns.nsims[0].dev cfg.psp_ifname = cfg.psp_dev['ifname'] cfg.psp_ifindex = cfg.psp_dev['ifindex'] # PSP peer device is the remote dev (in _netns, where psp_responder runs) cfg.psp_dev_peer = cfg._ns_peer.nsims[0].dev cfg.psp_dev_peer_ifname = cfg.psp_dev_peer['ifname'] cfg.psp_dev_peer_ifindex = cfg.psp_dev_peer['ifindex'] else: # Real NIC case: PSP device is the local interface cfg.psp_dev = cfg.dev cfg.psp_ifname = cfg.ifname cfg.psp_ifindex = cfg.ifindex # PSP peer device is the remote interface cfg.psp_dev_peer = cfg.remote_dev cfg.psp_dev_peer_ifname = cfg.remote_ifname cfg.psp_dev_peer_ifindex = cfg.remote_ifindex # Get nsid for the guest namespace (netns) where nk_guest is cfg.psp_dev_peer_nsid = _get_nsid(cfg.netns.name) def main() -> None: """ Ksft boiler plate main """ # Make sure LOCAL_PREFIX_V6 is set if "LOCAL_PREFIX_V6" not in os.environ: os.environ["LOCAL_PREFIX_V6"] = "2001:db8:2::" try: env = NetDrvContEnv(__file__, primary_rx_redirect=True) has_cont = True except KsftSkipEx: env = NetDrvEpEnv(__file__) has_cont = False with env as cfg: cfg.pspnl = PSPFamily() if has_cont: _setup_psp_attributes(cfg) # Set up responder and communication sock # psp_responder runs in _netns (remote namespace with psp_dev_peer) responder = cfg.remote.deploy("psp_responder") cfg.comm_port = rand_port() srv = None try: with bkg(responder + f" -p {cfg.comm_port} -i {cfg.remote_ifindex}", host=cfg.remote, exit_wait=True) as srv: wait_port_listen(cfg.comm_port, host=cfg.remote) cfg.comm_sock = socket.create_connection((cfg.remote_addr, cfg.comm_port), timeout=1) cases = [data_basic_send, data_mss_adjust] if has_cont: cases += [ _assoc_check_list, data_basic_send_netkit_psp_assoc, _key_rotation_notify_multi_ns_netkit, _dev_change_notify_multi_ns_netkit, _psp_dev_get_check_netkit_psp_assoc, _dev_assoc_no_nsid, _psp_dev_assoc_cleanup_on_netkit_del, ] ksft_run(cases=cases, globs=globals(), case_pfx={"dev_", "data_", "assoc_", "removal_"}, args=(cfg, )) cfg.comm_sock.send(b"exit\0") cfg.comm_sock.close() finally: if srv and (srv.stdout or srv.stderr): ksft_pr("") ksft_pr(f"Responder logs ({srv.ret}):") if srv and srv.stdout: ksft_pr("STDOUT:\n# " + srv.stdout.strip().replace("\n", "\n# ")) if srv and srv.stderr: ksft_pr("STDERR:\n# " + srv.stderr.strip().replace("\n", "\n# ")) ksft_exit() if __name__ == "__main__": main()