325 lines · 12.3 KB · Python
RawDownload
#!/usr/bin/env python3# SPDX-License-Identifier: GPL-2.0 import socketimport structimport timefrom lib.py import bkg, ip, ksft_exit, ksft_run, ksft_eq, ksft_ge, ksft_true, KsftSkipExfrom lib.py import ksft_not_in, ksft_not_nonefrom lib.py import CmdExitFailure, NetNS, NetNSEnter, RtnlAddrFamily, RtnlRouteFamilyfrom 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()