| #!/usr/bin/env python3 |
| # SPDX-License-Identifier: GPL-2.0 |
| |
| import socket |
| import struct |
| import time |
| from lib.py import bkg, ip, ksft_exit, ksft_run, ksft_eq, ksft_ge, ksft_true, KsftSkipEx |
| from lib.py import ksft_not_in, ksft_not_none |
| from lib.py import CmdExitFailure, NetNS, NetNSEnter, RtnlAddrFamily, RtnlRouteFamily |
| from lib.py import defer |
| |
| IPV4_ALL_HOSTS_MULTICAST = b'\xe0\x00\x00\x01' |
| IPV4_TEST_MULTICAST = b'\xef\x01\x01\x01' |
| IPV6_TEST_MULTICAST = bytes.fromhex('ff020000000000000000000000000123') |
| |
| |
| def _users_for(rtnl: RtnlAddrFamily, family: int, grp: bytes, ifindex: int): |
| """Return mc-users for grp on ifindex, or 0 if absent.""" |
| |
| addrs = rtnl.getmulticast({"ifa-family": family}, dump=True) |
| matches = [addr for addr in addrs |
| if addr['multicast'] == grp and addr['ifa-index'] == ifindex] |
| if not matches: |
| return 0 |
| if 'mc-users' not in matches[0]: |
| return None |
| |
| return matches[0]['mc-users'] |
| |
| |
| def dump_mcaddr_check() -> None: |
| """ |
| Verify IPv4 multicast addresses and their user counts in RTM_GETMULTICAST. |
| """ |
| |
| with NetNS() as ns: |
| with NetNSEnter(str(ns)): |
| ip("link set lo up") |
| rtnl = RtnlAddrFamily() |
| lo_idx = socket.if_nametoindex('lo') |
| addresses = rtnl.getmulticast({"ifa-family": socket.AF_INET}, dump=True) |
| |
| all_host_multicasts = [ |
| addr for addr in addresses |
| if addr['multicast'] == IPV4_ALL_HOSTS_MULTICAST |
| ] |
| |
| ksft_ge(len(all_host_multicasts), 1, |
| "No interface found with the IPv4 all-hosts multicast address") |
| |
| mreq = IPV4_TEST_MULTICAST + socket.inet_aton('127.0.0.1') |
| before = _users_for(rtnl, socket.AF_INET, IPV4_TEST_MULTICAST, lo_idx) |
| if before is None: |
| raise KsftSkipEx("kernel does not expose IFA_MC_USERS") |
| |
| s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) |
| s2 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) |
| try: |
| s1.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq) |
| s2.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq) |
| |
| after_join = _users_for(rtnl, socket.AF_INET, |
| IPV4_TEST_MULTICAST, lo_idx) |
| if after_join is None: |
| raise KsftSkipEx("kernel does not expose IFA_MC_USERS") |
| ksft_eq(after_join - before, 2, |
| f"users delta != 2 after two joins " |
| f"(before={before}, after={after_join})") |
| finally: |
| s1.close() |
| s2.close() |
| |
| |
| def dump_mcaddr6_check() -> None: |
| """ |
| Verify IPv6 multicast addresses and their user counts in RTM_GETMULTICAST. |
| """ |
| |
| with NetNS() as ns: |
| with NetNSEnter(str(ns)): |
| ip("link set lo up") |
| rtnl = RtnlAddrFamily() |
| lo_idx = socket.if_nametoindex('lo') |
| before = _users_for(rtnl, socket.AF_INET6, |
| IPV6_TEST_MULTICAST, lo_idx) |
| if before is None: |
| raise KsftSkipEx("kernel does not expose IFA_MC_USERS for IPv6") |
| |
| mreq = IPV6_TEST_MULTICAST + struct.pack('=I', lo_idx) |
| s1 = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM) |
| s2 = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM) |
| try: |
| s1.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_JOIN_GROUP, mreq) |
| s2.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_JOIN_GROUP, mreq) |
| |
| after_join = _users_for(rtnl, socket.AF_INET6, |
| IPV6_TEST_MULTICAST, lo_idx) |
| if after_join is None: |
| raise KsftSkipEx("kernel does not expose IFA_MC_USERS for IPv6") |
| ksft_eq(after_join - before, 2, |
| f"IPv6 users delta != 2 after two joins " |
| f"(before={before}, after={after_join})") |
| finally: |
| s1.close() |
| s2.close() |
| |
| |
| def ipv4_devconf_notify() -> None: |
| """ |
| Configure an interface and set ipv4-devconf values through netlink |
| to verify that the appropriate netlink notifications are being sent. |
| """ |
| |
| with NetNS() as ns: |
| with NetNSEnter(str(ns)): |
| ifname = "dummy1" |
| ip(f"link add name {ifname} type dummy", ns=str(ns)) |
| |
| with bkg("ip monitor", ns=str(ns)) as cmd_obj: |
| time.sleep(1) |
| try: |
| ip(f"link set dev {ifname} inet forwarding on") |
| ip(f"link set dev {ifname} inet proxy_arp on") |
| ip(f"link set dev {ifname} inet rp_filter 1") |
| ip(f"link set dev {ifname} inet ignore_routes_with_linkdown on") |
| except CmdExitFailure: |
| raise KsftSkipEx("iproute2 does not support IPv4 devconf attributes") |
| time.sleep(1) |
| |
| ksft_true(f"inet {ifname} ignore_routes_with_linkdown on" in cmd_obj.stdout, |
| f"No 'ignore_routes_with_linkdown on' notificiation found for interface {ifname}") |
| ksft_true(f"inet {ifname} rp_filter strict" in cmd_obj.stdout, |
| f"No 'rp_filter strict' notificiation found for interface {ifname}") |
| ksft_true(f"inet {ifname} proxy_neigh on" in cmd_obj.stdout, |
| f"No 'proxy_neigh on' notificiation found for interface {ifname}") |
| ksft_true(f"inet {ifname} forwarding on" in cmd_obj.stdout, |
| f"No 'forwarding on' notificiation found for interface {ifname}") |
| |
| def _rtnl_route_subscribe(ns): |
| with NetNSEnter(str(ns)): |
| rtnl = RtnlRouteFamily() |
| defer(rtnl.close) |
| rtnl.ntf_subscribe("rtnlgrp-ipv6-route") |
| return rtnl |
| |
| |
| def _wait_route_ntf(rtnl, name, dst_len, dst=None, deadline=10): |
| """Return the attrs of the first matching notification, None on timeout.""" |
| |
| for msg in rtnl.poll_ntf(duration=deadline): |
| if msg['name'] != name: |
| continue |
| attrs = msg['msg'] |
| if attrs['rtm-dst-len'] != dst_len: |
| continue |
| if dst is not None and attrs.get('dst') != dst: |
| continue |
| return attrs |
| return None |
| |
| |
| def _collect_route_ntfs(rtnl, name, want, deadline=10): |
| """Gather attrs of matching notifications, keyed by (dst_len, dst).""" |
| |
| seen = {} |
| for msg in rtnl.poll_ntf(duration=deadline): |
| if msg['name'] != name: |
| continue |
| attrs = msg['msg'] |
| key = (attrs['rtm-dst-len'], attrs.get('dst')) |
| if key in want: |
| seen[key] = attrs |
| if len(seen) == len(want): |
| break |
| return seen |
| |
| |
| def _write_ipv6_sysctl(name, value): |
| with open(f"/proc/sys/net/ipv6/{name}", "w") as f: |
| f.write(f"{value}\n") |
| |
| |
| def ipv6_route_del_reason_expired() -> None: |
| """An expired route reports RTA_DEL_REASON == expired.""" |
| |
| with NetNS() as ns: |
| rtnl = _rtnl_route_subscribe(ns) |
| with NetNSEnter(str(ns)): |
| _write_ipv6_sysctl("route/gc_interval", 2) |
| ip("link add name dummy1 type dummy", ns=str(ns)) |
| ip("link set dev dummy1 up", ns=str(ns)) |
| ip("-6 route add 2001:db8:2::/64 dev dummy1 expires 2", ns=str(ns)) |
| |
| attrs = _wait_route_ntf(rtnl, 'delroute-ntf', 64, '2001:db8:2::', |
| deadline=15) |
| ksft_not_none(attrs, "no RTM_DELROUTE for the expired route") |
| if attrs is not None: |
| ksft_eq(attrs.get('del-reason'), 'expired') |
| |
| |
| def _send_ra(sock, ifindex, lifetime, rio=None, pio=None): |
| """The kernel fills in the ICMPv6 checksum on raw ICMPv6 sockets.""" |
| |
| # type, code, cksum, hop limit, flags, router lifetime, |
| # reachable time, retrans timer |
| ra = struct.pack('!BBHBBHII', 134, 0, 0, 64, 0, lifetime, 0, 0) |
| if rio is not None: |
| prefix, plen, rio_lifetime = rio |
| # RFC 4191 route information option, /64 prefix (8 bytes) |
| ra += struct.pack('!BBBBI', 24, 2, plen, 0, rio_lifetime) |
| ra += socket.inet_pton(socket.AF_INET6, prefix)[:8] |
| if pio is not None: |
| prefix, plen, valid_lft = pio |
| # RFC 4861 prefix information option, on-link only (L set, A clear) |
| ra += struct.pack('!BBBBIII', 3, 4, plen, 0x80, valid_lft, 0, 0) |
| ra += socket.inet_pton(socket.AF_INET6, prefix) |
| sock.sendto(ra, ('ff02::1', 0, 0, ifindex)) |
| |
| |
| def _ra_router_sock(ns_r, ifname): |
| with NetNSEnter(str(ns_r)): |
| sock = socket.socket(socket.AF_INET6, socket.SOCK_RAW, |
| socket.IPPROTO_ICMPV6) |
| sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_MULTICAST_HOPS, 255) |
| defer(sock.close) |
| return sock, socket.if_nametoindex(ifname) |
| |
| |
| def _ra_advertise_routes(rtnl, sock, ifindex, want, **ra_opts): |
| """ |
| Sending fails with EADDRNOTAVAIL while the router's link-local |
| address is still tentative. addrconf_dad_start() only queues |
| addrconf_dad_work(), and IFA_F_TENTATIVE is cleared when that work |
| item runs, so retry until it does. |
| """ |
| |
| seen = {} |
| for _ in range(10): |
| try: |
| _send_ra(sock, ifindex, **ra_opts) |
| except OSError: |
| time.sleep(0.2) |
| continue |
| seen.update(_collect_route_ntfs(rtnl, 'newroute-ntf', |
| want - set(seen.keys()), deadline=2)) |
| if len(seen) == len(want): |
| break |
| return seen |
| |
| |
| def ipv6_route_del_reason_ra_withdrawn() -> None: |
| """ |
| Routes withdrawn by a zero-lifetime RA (router lifetime, RFC 4861 |
| PIO, RFC 4191 RIO) report RTA_DEL_REASON == ra-withdrawn. |
| """ |
| |
| # (rtm-dst-len, dst); the default route carries no RTA_DST |
| routes = {(0, None), (64, '2001:db8:6::'), (64, '2001:db8:5::')} |
| |
| with NetNS() as ns_h, NetNS() as ns_r: |
| ip(f"link add veth0 netns {ns_h} type veth peer name veth1 netns {ns_r}") |
| with NetNSEnter(str(ns_h)): |
| _write_ipv6_sysctl("conf/veth0/accept_ra", 2) |
| _write_ipv6_sysctl("conf/veth0/forwarding", 0) |
| try: |
| _write_ipv6_sysctl("conf/veth0/accept_ra_rt_info_max_plen", 64) |
| except FileNotFoundError: |
| raise KsftSkipEx("no CONFIG_IPV6_ROUTE_INFO") |
| with NetNSEnter(str(ns_r)): |
| # skip the DAD probe so the router's link-local source only |
| # has to wait for addrconf_dad_work() to clear IFA_F_TENTATIVE |
| _write_ipv6_sysctl("conf/veth1/accept_dad", 0) |
| ip("link set dev veth0 up", ns=str(ns_h)) |
| ip("link set dev veth1 up", ns=str(ns_r)) |
| |
| rtnl = _rtnl_route_subscribe(ns_h) |
| sock, ifindex = _ra_router_sock(ns_r, "veth1") |
| |
| seen = _ra_advertise_routes(rtnl, sock, ifindex, routes, |
| lifetime=1800, |
| rio=('2001:db8:5::', 64, 600), |
| pio=('2001:db8:6::', 64, 600)) |
| ksft_eq(set(seen), routes, "not all RA routes were installed") |
| if set(seen) != routes: |
| return |
| |
| _send_ra(sock, ifindex, 0, rio=('2001:db8:5::', 64, 0), |
| pio=('2001:db8:6::', 64, 0)) |
| seen = _collect_route_ntfs(rtnl, 'delroute-ntf', routes) |
| for key in routes: |
| attrs = seen.get(key) |
| ksft_not_none(attrs, f"no RTM_DELROUTE for {key}") |
| if attrs is not None: |
| ksft_eq(attrs.get('del-reason'), 'ra-withdrawn') |
| |
| |
| def ipv6_route_del_reason_absent() -> None: |
| """ |
| A deletion path that records no cause (here a userspace request) |
| must not carry RTA_DEL_REASON at all. |
| """ |
| |
| with NetNS() as ns: |
| rtnl = _rtnl_route_subscribe(ns) |
| ip("link add name dummy1 type dummy", ns=str(ns)) |
| ip("link set dev dummy1 up", ns=str(ns)) |
| ip("-6 route add 2001:db8:1::/64 dev dummy1", ns=str(ns)) |
| ip("-6 route del 2001:db8:1::/64 dev dummy1", ns=str(ns)) |
| |
| attrs = _wait_route_ntf(rtnl, 'delroute-ntf', 64, '2001:db8:1::') |
| ksft_not_none(attrs, "no RTM_DELROUTE for 2001:db8:1::/64") |
| if attrs is not None: |
| ksft_not_in('del-reason', attrs, |
| "user deletion must not carry del-reason") |
| |
| |
| def main() -> None: |
| ksft_run([dump_mcaddr_check, dump_mcaddr6_check, ipv4_devconf_notify, |
| ipv6_route_del_reason_expired, |
| ipv6_route_del_reason_ra_withdrawn, |
| ipv6_route_del_reason_absent]) |
| ksft_exit() |
| |
| if __name__ == "__main__": |
| main() |
