diff --git a/tests/sys/netlink/test_rtnl_iface.py b/tests/sys/netlink/test_rtnl_iface.py index 8660051be8e2..7cff592b1676 100644 --- a/tests/sys/netlink/test_rtnl_iface.py +++ b/tests/sys/netlink/test_rtnl_iface.py @@ -1,256 +1,256 @@ import errno import socket import pytest from atf_python.sys.net.netlink import IflattrType from atf_python.sys.net.netlink import IflinkInfo from atf_python.sys.net.netlink import IfLinkInfoDataVlan from atf_python.sys.net.netlink import NetlinkIflaMessage from atf_python.sys.net.netlink import NetlinkTestTemplate from atf_python.sys.net.netlink import NlAttrNested from atf_python.sys.net.netlink import NlAttrStr from atf_python.sys.net.netlink import NlAttrStrn from atf_python.sys.net.netlink import NlAttrU16 from atf_python.sys.net.netlink import NlAttrU32 from atf_python.sys.net.netlink import NlConst from atf_python.sys.net.netlink import NlmBaseFlags from atf_python.sys.net.netlink import NlmNewFlags from atf_python.sys.net.netlink import NlMsgType from atf_python.sys.net.netlink import NlRtMsgType from atf_python.sys.net.vnet import SingleVnetTestTemplate -class TestRtNlIface(SingleVnetTestTemplate, NetlinkTestTemplate): +class TestRtNlIface(NetlinkTestTemplate, SingleVnetTestTemplate): def setup_method(self, method): super().setup_method(method) self.setup_netlink(NlConst.NETLINK_ROUTE) def get_interface_byname(self, ifname): msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_GETLINK.value) msg.nl_hdr.nlmsg_flags = ( NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value ) msg.add_nla(NlAttrStr(IflattrType.IFLA_IFNAME, ifname)) self.write_message(msg) while True: rx_msg = self.read_message() if msg.nl_hdr.nlmsg_seq == rx_msg.nl_hdr.nlmsg_seq: if rx_msg.is_type(NlMsgType.NLMSG_ERROR): if rx_msg.error_code != 0: raise ValueError("unable to get interface {}".format(ifname)) elif rx_msg.is_type(NlRtMsgType.RTM_NEWLINK): return rx_msg else: raise ValueError("bad message") def test_get_iface_byname_error(self): """Tests error on fetching non-existing interface name""" msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_GETLINK.value) msg.nl_hdr.nlmsg_flags = ( NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value ) msg.add_nla(NlAttrStr(IflattrType.IFLA_IFNAME, "lo10")) rx_msg = self.get_reply(msg) assert rx_msg.is_type(NlMsgType.NLMSG_ERROR) assert rx_msg.error_code == errno.ENODEV def test_get_iface_byindex_error(self): """Tests error on fetching non-existing interface index""" msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_GETLINK.value) msg.nl_hdr.nlmsg_flags = ( NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value ) msg.base_hdr.ifi_index = 2147483647 rx_msg = self.get_reply(msg) assert rx_msg.is_type(NlMsgType.NLMSG_ERROR) assert rx_msg.error_code == errno.ENODEV @pytest.mark.require_user("root") def test_create_iface_plain(self): """Tests loopback creation w/o any parameters""" flags = NlmNewFlags.NLM_F_EXCL.value | NlmNewFlags.NLM_F_CREATE.value msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_NEWLINK.value) msg.nl_hdr.nlmsg_flags = ( flags | NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value ) msg.add_nla(NlAttrStr(IflattrType.IFLA_IFNAME, "lo10")) msg.add_nla( NlAttrNested( IflattrType.IFLA_LINKINFO, [ NlAttrStrn(IflinkInfo.IFLA_INFO_KIND, "lo"), ], ) ) rx_msg = self.get_reply(msg) assert rx_msg.is_type(NlMsgType.NLMSG_ERROR) assert rx_msg.error_code == 0 self.get_interface_byname("lo10") @pytest.mark.require_user("root") def test_create_iface_attrs(self): """Tests interface creation with additional properties""" flags = NlmNewFlags.NLM_F_EXCL.value | NlmNewFlags.NLM_F_CREATE.value msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_NEWLINK.value) msg.nl_hdr.nlmsg_flags = ( flags | NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value ) msg.add_nla(NlAttrStr(IflattrType.IFLA_IFNAME, "lo10")) msg.add_nla( NlAttrNested( IflattrType.IFLA_LINKINFO, [ NlAttrStrn(IflinkInfo.IFLA_INFO_KIND, "lo"), ], ) ) # Custom attributes msg.add_nla(NlAttrStr(IflattrType.IFLA_IFALIAS, "test description")) msg.add_nla(NlAttrU32(IflattrType.IFLA_MTU, 1024)) rx_msg = self.get_reply(msg) assert rx_msg.is_type(NlMsgType.NLMSG_ERROR) assert rx_msg.error_code == 0 iface_msg = self.get_interface_byname("lo10") assert iface_msg.get_nla(IflattrType.IFLA_IFALIAS).text == "test description" assert iface_msg.get_nla(IflattrType.IFLA_MTU).u32 == 1024 @pytest.mark.require_user("root") def test_modify_iface_attrs(self): """Tests interface modifications""" flags = NlmNewFlags.NLM_F_EXCL.value | NlmNewFlags.NLM_F_CREATE.value msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_NEWLINK.value) msg.nl_hdr.nlmsg_flags = ( flags | NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value ) msg.add_nla(NlAttrStr(IflattrType.IFLA_IFNAME, "lo10")) msg.add_nla( NlAttrNested( IflattrType.IFLA_LINKINFO, [ NlAttrStrn(IflinkInfo.IFLA_INFO_KIND, "lo"), ], ) ) rx_msg = self.get_reply(msg) assert rx_msg.is_type(NlMsgType.NLMSG_ERROR) assert rx_msg.error_code == 0 msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_NEWLINK.value) msg.nl_hdr.nlmsg_flags = ( NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value ) msg.add_nla(NlAttrStr(IflattrType.IFLA_IFNAME, "lo10")) # Custom attributes msg.add_nla(NlAttrStr(IflattrType.IFLA_IFALIAS, "test description")) msg.add_nla(NlAttrU32(IflattrType.IFLA_MTU, 1024)) rx_msg = self.get_reply(msg) assert rx_msg.is_type(NlMsgType.NLMSG_ERROR) assert rx_msg.error_code == 0 iface_msg = self.get_interface_byname("lo10") assert iface_msg.get_nla(IflattrType.IFLA_IFALIAS).text == "test description" assert iface_msg.get_nla(IflattrType.IFLA_MTU).u32 == 1024 @pytest.mark.require_user("root") def test_delete_iface(self): """Tests interface modifications""" flags = NlmNewFlags.NLM_F_EXCL.value | NlmNewFlags.NLM_F_CREATE.value msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_NEWLINK.value) msg.nl_hdr.nlmsg_flags = ( flags | NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value ) msg.add_nla(NlAttrStr(IflattrType.IFLA_IFNAME, "lo10")) msg.add_nla( NlAttrNested( IflattrType.IFLA_LINKINFO, [ NlAttrStrn(IflinkInfo.IFLA_INFO_KIND, "lo"), ], ) ) rx_msg = self.get_reply(msg) assert rx_msg.is_type(NlMsgType.NLMSG_ERROR) assert rx_msg.error_code == 0 iface_msg = self.get_interface_byname("lo10") iface_idx = iface_msg.base_hdr.ifi_index msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_DELLINK.value) msg.nl_hdr.nlmsg_flags = ( NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value ) msg.base_hdr.ifi_index = iface_idx # msg.add_nla(NlAttrStr(IflattrType.IFLA_IFNAME, "lo10")) rx_msg = self.get_reply(msg) assert rx_msg.is_type(NlMsgType.NLMSG_ERROR) assert rx_msg.error_code == 0 msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_GETLINK.value) msg.nl_hdr.nlmsg_flags = ( NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value ) msg.base_hdr.ifi_index = 2147483647 rx_msg = self.get_reply(msg) assert rx_msg.is_type(NlMsgType.NLMSG_ERROR) assert rx_msg.error_code == errno.ENODEV # # * # * {len=76, type=RTM_NEWLINK, flags=NLM_F_REQUEST|NLM_F_ACK|NLM_F_EXCL|NLM_F_CREATE, seq=1662892737, pid=0}, # * {ifi_family=AF_UNSPEC, ifi_type=ARPHRD_NETROM, ifi_index=0, ifi_flags=0, ifi_change=0}, # * {{nla_len=8, nla_type=IFLA_LINK}, 2}, # * {{nla_len=12, nla_type=IFLA_IFNAME}, "xvlan22"}, # * {{nla_len=24, nla_type=IFLA_LINKINFO}, # * {{nla_len=8, nla_type=IFLA_INFO_KIND}, "vlan"...}, # * {{nla_len=12, nla_type=IFLA_INFO_DATA}, "\x06\x00\x01\x00\x16\x00\x00\x00"} # */ @pytest.mark.skip(reason="vlan support needs more work") @pytest.mark.require_user("root") def test_create_vlan_plain(self): """Creates 802.1Q VLAN interface in vlanXX and ifX fashion""" os_ifname = self.vnet.iface_alias_map["if1"].name ifindex = socket.if_nametoindex(os_ifname) flags = NlmNewFlags.NLM_F_EXCL.value | NlmNewFlags.NLM_F_CREATE.value msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_NEWLINK.value) msg.nl_hdr.nlmsg_flags = ( flags | NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value ) msg.add_nla(NlAttrU32(IflattrType.IFLA_LINK, ifindex)) msg.add_nla(NlAttrStr(IflattrType.IFLA_IFNAME, "vlan22")) msg.add_nla( NlAttrNested( IflattrType.IFLA_LINKINFO, [ NlAttrStrn(IflinkInfo.IFLA_INFO_KIND, "vlan"), NlAttrNested( IflinkInfo.IFLA_INFO_DATA, [ NlAttrU16(IfLinkInfoDataVlan.IFLA_VLAN_ID, 22), ], ), ], ) ) rx_msg = self.get_reply(msg) assert rx_msg.is_type(NlMsgType.NLMSG_ERROR) assert rx_msg.error_code == 0 self.get_interface_byname("vlan22") # ToolsHelper.print_net_debug() diff --git a/tests/sys/netlink/test_rtnl_ifaddr.py b/tests/sys/netlink/test_rtnl_ifaddr.py index 5574644f7b2f..bd1c55e268e7 100644 --- a/tests/sys/netlink/test_rtnl_ifaddr.py +++ b/tests/sys/netlink/test_rtnl_ifaddr.py @@ -1,144 +1,144 @@ import ipaddress import socket import struct from atf_python.sys.net.netlink import IfattrType from atf_python.sys.net.netlink import NetlinkIfaMessage from atf_python.sys.net.netlink import NetlinkTestTemplate from atf_python.sys.net.netlink import NlConst from atf_python.sys.net.netlink import NlHelper from atf_python.sys.net.netlink import NlmBaseFlags from atf_python.sys.net.netlink import NlMsgType from atf_python.sys.net.netlink import NlRtMsgType from atf_python.sys.net.netlink import Nlsock from atf_python.sys.net.netlink import RtScope from atf_python.sys.net.vnet import SingleVnetTestTemplate -class TestRtNlIfaddr(SingleVnetTestTemplate, NetlinkTestTemplate): +class TestRtNlIfaddr(NetlinkTestTemplate, SingleVnetTestTemplate): def setup_method(self, method): method_name = method.__name__ if "4" in method_name: self.IPV4_PREFIXES = ["192.0.2.1/24"] if "6" in method_name: self.IPV6_PREFIXES = ["2001:db8::1/64"] super().setup_method(method) self.setup_netlink(NlConst.NETLINK_ROUTE) def test_46_nofilter(self): """Tests that listing outputs both IPv4/IPv6 and interfaces""" msg = NetlinkIfaMessage(self.helper, NlRtMsgType.RTM_GETADDR.value) msg.nl_hdr.nlmsg_flags = ( NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value ) self.write_message(msg) ret = [] for rx_msg in self.read_msg_list(msg.nl_hdr.nlmsg_seq, NlRtMsgType.RTM_NEWADDR): ifname = socket.if_indextoname(rx_msg.base_hdr.ifa_index) family = rx_msg.base_hdr.ifa_family ret.append((ifname, family, rx_msg)) ifname = "lo0" assert len([r for r in ret if r[0] == ifname]) > 0 ifname = self.vnet.iface_alias_map["if1"].name assert len([r for r in ret if r[0] == ifname and r[1] == socket.AF_INET]) == 1 assert len([r for r in ret if r[0] == ifname and r[1] == socket.AF_INET6]) == 2 def test_46_filter_iface(self): """Tests that listing outputs both IPv4/IPv6 for the specific interface""" epair_ifname = self.vnet.iface_alias_map["if1"].name msg = NetlinkIfaMessage(self.helper, NlRtMsgType.RTM_GETADDR.value) msg.nl_hdr.nlmsg_flags = ( NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value ) msg.base_hdr.ifa_index = socket.if_nametoindex(epair_ifname) self.write_message(msg) ret = [] for rx_msg in self.read_msg_list(msg.nl_hdr.nlmsg_seq, NlRtMsgType.RTM_NEWADDR): ifname = socket.if_indextoname(rx_msg.base_hdr.ifa_index) family = rx_msg.base_hdr.ifa_family ret.append((ifname, family, rx_msg)) ifname = epair_ifname assert len([r for r in ret if r[0] == ifname and r[1] == socket.AF_INET]) == 1 assert len([r for r in ret if r[0] == ifname and r[1] == socket.AF_INET6]) == 2 assert len(ret) == 3 def filter_iface_family(self, family, num_items): """Tests that listing outputs IPv4 for the specific interface""" epair_ifname = self.vnet.iface_alias_map["if1"].name msg = NetlinkIfaMessage(self.helper, NlRtMsgType.RTM_GETADDR.value) msg.nl_hdr.nlmsg_flags = ( NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value ) msg.base_hdr.ifa_family = family msg.base_hdr.ifa_index = socket.if_nametoindex(epair_ifname) self.write_message(msg) ret = [] for rx_msg in self.read_msg_list(msg.nl_hdr.nlmsg_seq, NlRtMsgType.RTM_NEWADDR): assert family == rx_msg.base_hdr.ifa_family assert epair_ifname == socket.if_indextoname(rx_msg.base_hdr.ifa_index) ret.append(rx_msg) assert len(ret) == num_items return ret def test_4_broadcast(self): """Tests header/attr output for listing IPv4 ifas on broadcast iface""" ret = self.filter_iface_family(socket.AF_INET, 1) # Should be 192.0.2.1/24 msg = ret[0] # Family and ifindex has been checked already assert msg.base_hdr.ifa_prefixlen == 24 # Ignore IFA_FLAGS for now assert msg.base_hdr.ifa_scope == RtScope.RT_SCOPE_UNIVERSE.value assert msg.get_nla(IfattrType.IFA_ADDRESS).addr == "192.0.2.1" assert msg.get_nla(IfattrType.IFA_LOCAL).addr == "192.0.2.1" assert msg.get_nla(IfattrType.IFA_BROADCAST).addr == "192.0.2.255" epair_ifname = self.vnet.iface_alias_map["if1"].name assert msg.get_nla(IfattrType.IFA_LABEL).text == epair_ifname def test_6_broadcast(self): """Tests header/attr output for listing IPv6 ifas on broadcast iface""" ret = self.filter_iface_family(socket.AF_INET6, 2) # Should be 192.0.2.1/24 if ret[0].base_hdr.ifa_scope == RtScope.RT_SCOPE_UNIVERSE.value: (gmsg, lmsg) = ret else: (lmsg, gmsg) = ret # Start with global ( 2001:db8::1/64 ) msg = gmsg # Family and ifindex has been checked already assert msg.base_hdr.ifa_prefixlen == 64 # Ignore IFA_FLAGS for now assert msg.base_hdr.ifa_scope == RtScope.RT_SCOPE_UNIVERSE.value assert msg.get_nla(IfattrType.IFA_ADDRESS).addr == "2001:db8::1" assert msg.get_nla(IfattrType.IFA_LOCAL) is None assert msg.get_nla(IfattrType.IFA_BROADCAST) is None epair_ifname = self.vnet.iface_alias_map["if1"].name assert msg.get_nla(IfattrType.IFA_LABEL).text == epair_ifname # Local: fe80::/64 msg = lmsg assert msg.base_hdr.ifa_prefixlen == 64 # Ignore IFA_FLAGS for now assert msg.base_hdr.ifa_scope == RtScope.RT_SCOPE_LINK.value addr = ipaddress.ip_address(msg.get_nla(IfattrType.IFA_ADDRESS).addr) assert addr.is_link_local # Verify that ifindex is not emmbedded assert struct.unpack("!H", addr.packed[2:4])[0] == 0 assert msg.get_nla(IfattrType.IFA_LOCAL) is None assert msg.get_nla(IfattrType.IFA_BROADCAST) is None epair_ifname = self.vnet.iface_alias_map["if1"].name assert msg.get_nla(IfattrType.IFA_LABEL).text == epair_ifname