1#!/usr/bin/env python3 2# SPDX-License-Identifier: GPL-2.0 3 4import socket 5import struct 6import time 7from lib.py import bkg, ip, ksft_exit, ksft_run, ksft_eq, ksft_ge, ksft_true, KsftSkipEx 8from lib.py import CmdExitFailure, NetNS, NetNSEnter, RtnlAddrFamily 9 10IPV4_ALL_HOSTS_MULTICAST = b'\xe0\x00\x00\x01' 11IPV4_TEST_MULTICAST = b'\xef\x01\x01\x01' 12IPV6_TEST_MULTICAST = bytes.fromhex('ff020000000000000000000000000123') 13 14 15def _users_for(rtnl: RtnlAddrFamily, family: int, grp: bytes, ifindex: int): 16 """Return mc-users for grp on ifindex, or 0 if absent.""" 17 18 addrs = rtnl.getmulticast({"ifa-family": family}, dump=True) 19 matches = [addr for addr in addrs 20 if addr['multicast'] == grp and addr['ifa-index'] == ifindex] 21 if not matches: 22 return 0 23 if 'mc-users' not in matches[0]: 24 return None 25 26 return matches[0]['mc-users'] 27 28 29def dump_mcaddr_check() -> None: 30 """ 31 Verify IPv4 multicast addresses and their user counts in RTM_GETMULTICAST. 32 """ 33 34 with NetNS() as ns: 35 with NetNSEnter(str(ns)): 36 ip("link set lo up") 37 rtnl = RtnlAddrFamily() 38 lo_idx = socket.if_nametoindex('lo') 39 addresses = rtnl.getmulticast({"ifa-family": socket.AF_INET}, dump=True) 40 41 all_host_multicasts = [ 42 addr for addr in addresses 43 if addr['multicast'] == IPV4_ALL_HOSTS_MULTICAST 44 ] 45 46 ksft_ge(len(all_host_multicasts), 1, 47 "No interface found with the IPv4 all-hosts multicast address") 48 49 mreq = IPV4_TEST_MULTICAST + socket.inet_aton('127.0.0.1') 50 before = _users_for(rtnl, socket.AF_INET, IPV4_TEST_MULTICAST, lo_idx) 51 if before is None: 52 raise KsftSkipEx("kernel does not expose IFA_MC_USERS") 53 54 s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) 55 s2 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) 56 try: 57 s1.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq) 58 s2.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq) 59 60 after_join = _users_for(rtnl, socket.AF_INET, 61 IPV4_TEST_MULTICAST, lo_idx) 62 if after_join is None: 63 raise KsftSkipEx("kernel does not expose IFA_MC_USERS") 64 ksft_eq(after_join - before, 2, 65 f"users delta != 2 after two joins " 66 f"(before={before}, after={after_join})") 67 finally: 68 s1.close() 69 s2.close() 70 71 72def dump_mcaddr6_check() -> None: 73 """ 74 Verify IPv6 multicast addresses and their user counts in RTM_GETMULTICAST. 75 """ 76 77 with NetNS() as ns: 78 with NetNSEnter(str(ns)): 79 ip("link set lo up") 80 rtnl = RtnlAddrFamily() 81 lo_idx = socket.if_nametoindex('lo') 82 before = _users_for(rtnl, socket.AF_INET6, 83 IPV6_TEST_MULTICAST, lo_idx) 84 if before is None: 85 raise KsftSkipEx("kernel does not expose IFA_MC_USERS for IPv6") 86 87 mreq = IPV6_TEST_MULTICAST + struct.pack('=I', lo_idx) 88 s1 = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM) 89 s2 = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM) 90 try: 91 s1.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_JOIN_GROUP, mreq) 92 s2.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_JOIN_GROUP, mreq) 93 94 after_join = _users_for(rtnl, socket.AF_INET6, 95 IPV6_TEST_MULTICAST, lo_idx) 96 if after_join is None: 97 raise KsftSkipEx("kernel does not expose IFA_MC_USERS for IPv6") 98 ksft_eq(after_join - before, 2, 99 f"IPv6 users delta != 2 after two joins " 100 f"(before={before}, after={after_join})") 101 finally: 102 s1.close() 103 s2.close() 104 105 106def ipv4_devconf_notify() -> None: 107 """ 108 Configure an interface and set ipv4-devconf values through netlink 109 to verify that the appropriate netlink notifications are being sent. 110 """ 111 112 with NetNS() as ns: 113 with NetNSEnter(str(ns)): 114 ifname = "dummy1" 115 ip(f"link add name {ifname} type dummy", ns=str(ns)) 116 117 with bkg("ip monitor", ns=str(ns)) as cmd_obj: 118 time.sleep(1) 119 try: 120 ip(f"link set dev {ifname} inet forwarding on") 121 ip(f"link set dev {ifname} inet proxy_arp on") 122 ip(f"link set dev {ifname} inet rp_filter 1") 123 ip(f"link set dev {ifname} inet ignore_routes_with_linkdown on") 124 except CmdExitFailure: 125 raise KsftSkipEx("iproute2 does not support IPv4 devconf attributes") 126 time.sleep(1) 127 128 ksft_true(f"inet {ifname} ignore_routes_with_linkdown on" in cmd_obj.stdout, 129 f"No 'ignore_routes_with_linkdown on' notificiation found for interface {ifname}") 130 ksft_true(f"inet {ifname} rp_filter strict" in cmd_obj.stdout, 131 f"No 'rp_filter strict' notificiation found for interface {ifname}") 132 ksft_true(f"inet {ifname} proxy_neigh on" in cmd_obj.stdout, 133 f"No 'proxy_neigh on' notificiation found for interface {ifname}") 134 ksft_true(f"inet {ifname} forwarding on" in cmd_obj.stdout, 135 f"No 'forwarding on' notificiation found for interface {ifname}") 136 137def main() -> None: 138 ksft_run([dump_mcaddr_check, dump_mcaddr6_check, ipv4_devconf_notify]) 139 ksft_exit() 140 141if __name__ == "__main__": 142 main() 143