1import errno 2import socket 3 4import pytest 5from atf_python.sys.netlink.netlink_route import IflattrType 6from atf_python.sys.netlink.netlink_route import IflafAttrType 7from atf_python.sys.netlink.netlink_route import IflinkInfo 8from atf_python.sys.netlink.netlink_route import IfLinkInfoDataVlan 9from atf_python.sys.netlink.netlink_route import NetlinkIflaMessage 10from atf_python.sys.netlink.netlink import NetlinkTestTemplate 11from atf_python.sys.netlink.attrs import NlAttrNested 12from atf_python.sys.netlink.attrs import NlAttrStr 13from atf_python.sys.netlink.attrs import NlAttrStrn 14from atf_python.sys.netlink.attrs import NlAttrU16 15from atf_python.sys.netlink.attrs import NlAttrU32 16from atf_python.sys.netlink.utils import NlConst 17from atf_python.sys.netlink.base_headers import NlmBaseFlags 18from atf_python.sys.netlink.base_headers import NlmNewFlags 19from atf_python.sys.netlink.base_headers import NlMsgType 20from atf_python.sys.netlink.netlink_route import NlRtMsgType 21from atf_python.sys.netlink.netlink_route import rtnl_ifla_attrs 22from atf_python.sys.net.vnet import SingleVnetTestTemplate 23from atf_python.sys.net.tools import ToolsHelper 24 25 26class TestRtNlIface(NetlinkTestTemplate, SingleVnetTestTemplate): 27 def setup_method(self, method): 28 super().setup_method(method) 29 self.setup_netlink(NlConst.NETLINK_ROUTE) 30 31 def get_interface_byname(self, ifname, ext_mask=0): 32 msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_GETLINK.value) 33 msg.nl_hdr.nlmsg_flags = ( 34 NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value 35 ) 36 msg.add_nla(NlAttrStr(IflattrType.IFLA_IFNAME, ifname)) 37 if ext_mask != 0: 38 msg.add_nla(NlAttrU32(IflattrType.IFLA_EXT_MASK, ext_mask)) 39 self.write_message(msg) 40 while True: 41 rx_msg = self.read_message() 42 if msg.nl_hdr.nlmsg_seq == rx_msg.nl_hdr.nlmsg_seq: 43 if rx_msg.is_type(NlMsgType.NLMSG_ERROR): 44 if rx_msg.error_code != 0: 45 raise ValueError("unable to get interface {}".format(ifname)) 46 elif rx_msg.is_type(NlRtMsgType.RTM_NEWLINK): 47 return rx_msg 48 else: 49 raise ValueError("bad message") 50 51 def test_get_iface_byname_error(self): 52 """Tests error on fetching non-existing interface name""" 53 msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_GETLINK.value) 54 msg.nl_hdr.nlmsg_flags = ( 55 NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value 56 ) 57 msg.add_nla(NlAttrStr(IflattrType.IFLA_IFNAME, "lo10")) 58 59 rx_msg = self.get_reply(msg) 60 assert rx_msg.is_type(NlMsgType.NLMSG_ERROR) 61 assert rx_msg.error_code == errno.ENODEV 62 63 def test_get_iface_byindex_error(self): 64 """Tests error on fetching non-existing interface index""" 65 msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_GETLINK.value) 66 msg.nl_hdr.nlmsg_flags = ( 67 NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value 68 ) 69 msg.base_hdr.ifi_index = 2147483647 70 71 rx_msg = self.get_reply(msg) 72 assert rx_msg.is_type(NlMsgType.NLMSG_ERROR) 73 assert rx_msg.error_code == errno.ENODEV 74 75 def test_get_iface_vf_filter_unsupported(self): 76 """Tests that the VF filter is accepted on a non-SR-IOV interface""" 77 rx_msg = self.get_interface_byname("lo0", 1) 78 assert rx_msg.get_nla(IflattrType.IFLA_NUM_VF) is None 79 fbsd = rx_msg.get_nla(IflattrType.IFLA_FREEBSD) 80 assert fbsd is not None 81 assert fbsd.get_nla(IflafAttrType.IFLAF_VF_STATUS) is None 82 83 @pytest.mark.require_user("root") 84 def test_create_iface_plain(self): 85 """Tests loopback creation w/o any parameters""" 86 flags = NlmNewFlags.NLM_F_EXCL.value | NlmNewFlags.NLM_F_CREATE.value 87 msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_NEWLINK.value) 88 msg.nl_hdr.nlmsg_flags = ( 89 flags | NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value 90 ) 91 msg.add_nla(NlAttrStr(IflattrType.IFLA_IFNAME, "lo10")) 92 msg.add_nla( 93 NlAttrNested( 94 IflattrType.IFLA_LINKINFO, 95 [ 96 NlAttrStrn(IflinkInfo.IFLA_INFO_KIND, "lo"), 97 ], 98 ) 99 ) 100 101 rx_msg = self.get_reply(msg) 102 assert rx_msg.is_type(NlMsgType.NLMSG_ERROR) 103 assert rx_msg.error_code == 0 104 105 self.get_interface_byname("lo10") 106 107 @pytest.mark.require_user("root") 108 def test_create_iface_plain_retvals(self): 109 """Tests loopback creation w/o any parameters""" 110 flags = NlmNewFlags.NLM_F_EXCL.value | NlmNewFlags.NLM_F_CREATE.value 111 msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_NEWLINK.value) 112 msg.nl_hdr.nlmsg_flags = ( 113 flags | NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value 114 ) 115 msg.add_nla(NlAttrStr(IflattrType.IFLA_IFNAME, "lo10")) 116 msg.add_nla( 117 NlAttrNested( 118 IflattrType.IFLA_LINKINFO, 119 [ 120 NlAttrStrn(IflinkInfo.IFLA_INFO_KIND, "lo"), 121 ], 122 ) 123 ) 124 125 rx_msg = self.get_reply(msg) 126 assert rx_msg.is_type(NlMsgType.NLMSG_ERROR) 127 assert rx_msg.error_code == 0 128 assert rx_msg.cookie is not None 129 nla_list, _ = rx_msg.parse_attrs(bytes(rx_msg.cookie)[4:], rtnl_ifla_attrs) 130 nla_map = {n.nla_type: n for n in nla_list} 131 assert IflattrType.IFLA_IFNAME.value in nla_map 132 assert nla_map[IflattrType.IFLA_IFNAME.value].text == "lo10" 133 assert IflattrType.IFLA_NEW_IFINDEX.value in nla_map 134 assert nla_map[IflattrType.IFLA_NEW_IFINDEX.value].u32 > 0 135 136 lo_msg = self.get_interface_byname("lo10") 137 assert ( 138 lo_msg.base_hdr.ifi_index == nla_map[IflattrType.IFLA_NEW_IFINDEX.value].u32 139 ) 140 141 @pytest.mark.require_user("root") 142 def test_create_iface_attrs(self): 143 """Tests interface creation with additional properties""" 144 flags = NlmNewFlags.NLM_F_EXCL.value | NlmNewFlags.NLM_F_CREATE.value 145 msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_NEWLINK.value) 146 msg.nl_hdr.nlmsg_flags = ( 147 flags | NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value 148 ) 149 msg.add_nla(NlAttrStr(IflattrType.IFLA_IFNAME, "lo10")) 150 msg.add_nla( 151 NlAttrNested( 152 IflattrType.IFLA_LINKINFO, 153 [ 154 NlAttrStrn(IflinkInfo.IFLA_INFO_KIND, "lo"), 155 ], 156 ) 157 ) 158 159 # Custom attributes 160 msg.add_nla(NlAttrStr(IflattrType.IFLA_IFALIAS, "test description")) 161 msg.add_nla(NlAttrU32(IflattrType.IFLA_MTU, 1024)) 162 163 rx_msg = self.get_reply(msg) 164 assert rx_msg.is_type(NlMsgType.NLMSG_ERROR) 165 assert rx_msg.error_code == 0 166 167 iface_msg = self.get_interface_byname("lo10") 168 assert iface_msg.get_nla(IflattrType.IFLA_IFALIAS).text == "test description" 169 assert iface_msg.get_nla(IflattrType.IFLA_MTU).u32 == 1024 170 171 @pytest.mark.require_user("root") 172 def test_modify_iface_attrs(self): 173 """Tests interface modifications""" 174 flags = NlmNewFlags.NLM_F_EXCL.value | NlmNewFlags.NLM_F_CREATE.value 175 msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_NEWLINK.value) 176 msg.nl_hdr.nlmsg_flags = ( 177 flags | NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value 178 ) 179 msg.add_nla(NlAttrStr(IflattrType.IFLA_IFNAME, "lo10")) 180 msg.add_nla( 181 NlAttrNested( 182 IflattrType.IFLA_LINKINFO, 183 [ 184 NlAttrStrn(IflinkInfo.IFLA_INFO_KIND, "lo"), 185 ], 186 ) 187 ) 188 189 rx_msg = self.get_reply(msg) 190 assert rx_msg.is_type(NlMsgType.NLMSG_ERROR) 191 assert rx_msg.error_code == 0 192 193 msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_NEWLINK.value) 194 msg.nl_hdr.nlmsg_flags = ( 195 NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value 196 ) 197 msg.add_nla(NlAttrStr(IflattrType.IFLA_IFNAME, "lo10")) 198 199 # Custom attributes 200 msg.add_nla(NlAttrStr(IflattrType.IFLA_IFALIAS, "test description")) 201 msg.add_nla(NlAttrU32(IflattrType.IFLA_MTU, 1024)) 202 203 rx_msg = self.get_reply(msg) 204 assert rx_msg.is_type(NlMsgType.NLMSG_ERROR) 205 assert rx_msg.error_code == 0 206 207 iface_msg = self.get_interface_byname("lo10") 208 assert iface_msg.get_nla(IflattrType.IFLA_IFALIAS).text == "test description" 209 assert iface_msg.get_nla(IflattrType.IFLA_MTU).u32 == 1024 210 211 @pytest.mark.require_user("root") 212 def test_delete_iface(self): 213 """Tests interface modifications""" 214 flags = NlmNewFlags.NLM_F_EXCL.value | NlmNewFlags.NLM_F_CREATE.value 215 msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_NEWLINK.value) 216 msg.nl_hdr.nlmsg_flags = ( 217 flags | NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value 218 ) 219 msg.add_nla(NlAttrStr(IflattrType.IFLA_IFNAME, "lo10")) 220 msg.add_nla( 221 NlAttrNested( 222 IflattrType.IFLA_LINKINFO, 223 [ 224 NlAttrStrn(IflinkInfo.IFLA_INFO_KIND, "lo"), 225 ], 226 ) 227 ) 228 229 rx_msg = self.get_reply(msg) 230 assert rx_msg.is_type(NlMsgType.NLMSG_ERROR) 231 assert rx_msg.error_code == 0 232 233 iface_msg = self.get_interface_byname("lo10") 234 iface_idx = iface_msg.base_hdr.ifi_index 235 236 msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_DELLINK.value) 237 msg.nl_hdr.nlmsg_flags = ( 238 NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value 239 ) 240 msg.base_hdr.ifi_index = iface_idx 241 # msg.add_nla(NlAttrStr(IflattrType.IFLA_IFNAME, "lo10")) 242 243 rx_msg = self.get_reply(msg) 244 assert rx_msg.is_type(NlMsgType.NLMSG_ERROR) 245 assert rx_msg.error_code == 0 246 247 msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_GETLINK.value) 248 msg.nl_hdr.nlmsg_flags = ( 249 NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value 250 ) 251 msg.base_hdr.ifi_index = 2147483647 252 253 rx_msg = self.get_reply(msg) 254 assert rx_msg.is_type(NlMsgType.NLMSG_ERROR) 255 assert rx_msg.error_code == errno.ENODEV 256 257 @pytest.mark.require_user("root") 258 def test_dump_ifaces_many(self): 259 """Tests if interface dummp is not missing interfaces""" 260 261 ifmap = {} 262 ifmap[socket.if_nametoindex("lo0")] = "lo0" 263 264 for i in range(40): 265 ifname = "lo{}".format(i + 1) 266 flags = NlmNewFlags.NLM_F_EXCL.value | NlmNewFlags.NLM_F_CREATE.value 267 msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_NEWLINK.value) 268 msg.nl_hdr.nlmsg_flags = ( 269 flags | NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value 270 ) 271 msg.add_nla(NlAttrStr(IflattrType.IFLA_IFNAME, ifname)) 272 msg.add_nla( 273 NlAttrNested( 274 IflattrType.IFLA_LINKINFO, 275 [ 276 NlAttrStrn(IflinkInfo.IFLA_INFO_KIND, "lo"), 277 ], 278 ) 279 ) 280 281 rx_msg = self.get_reply(msg) 282 assert rx_msg.is_type(NlMsgType.NLMSG_ERROR) 283 nla_list, _ = rx_msg.parse_attrs(bytes(rx_msg.cookie)[4:], rtnl_ifla_attrs) 284 nla_map = {n.nla_type: n for n in nla_list} 285 assert nla_map[IflattrType.IFLA_IFNAME.value].text == ifname 286 ifindex = nla_map[IflattrType.IFLA_NEW_IFINDEX.value].u32 287 assert ifindex > 0 288 assert ifindex not in ifmap 289 ifmap[ifindex] = ifname 290 291 # Dump all interfaces and check if the output matches ifmap 292 kernel_ifmap = {} 293 msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_GETLINK.value) 294 msg.nl_hdr.nlmsg_flags = ( 295 NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value 296 ) 297 self.write_message(msg) 298 while True: 299 rx_msg = self.read_message() 300 if msg.nl_hdr.nlmsg_seq != rx_msg.nl_hdr.nlmsg_seq: 301 raise ValueError( 302 "unexpected seq {}".format(rx_msg.nl_hdr.nlmsg_seq) 303 ) 304 if rx_msg.is_type(NlMsgType.NLMSG_ERROR): 305 raise ValueError("unexpected message {}".format(rx_msg)) 306 if rx_msg.is_type(NlMsgType.NLMSG_DONE): 307 break 308 if not rx_msg.is_type(NlRtMsgType.RTM_NEWLINK): 309 raise ValueError("unexpected message {}".format(rx_msg)) 310 311 ifindex = rx_msg.base_hdr.ifi_index 312 assert ifindex == rx_msg.base_hdr.ifi_index 313 ifname = rx_msg.get_nla(IflattrType.IFLA_IFNAME).text 314 if ifname.startswith("lo"): 315 kernel_ifmap[ifindex] = ifname 316 assert kernel_ifmap == ifmap 317 318 # 319 # * 320 # * {len=76, type=RTM_NEWLINK, flags=NLM_F_REQUEST|NLM_F_ACK|NLM_F_EXCL|NLM_F_CREATE, seq=1662892737, pid=0}, 321 # * {ifi_family=AF_UNSPEC, ifi_type=ARPHRD_NETROM, ifi_index=0, ifi_flags=0, ifi_change=0}, 322 # * {{nla_len=8, nla_type=IFLA_LINK}, 2}, 323 # * {{nla_len=12, nla_type=IFLA_IFNAME}, "xvlan22"}, 324 # * {{nla_len=24, nla_type=IFLA_LINKINFO}, 325 # * {{nla_len=8, nla_type=IFLA_INFO_KIND}, "vlan"...}, 326 # * {{nla_len=12, nla_type=IFLA_INFO_DATA}, "\x06\x00\x01\x00\x16\x00\x00\x00"} 327 # */ 328 @pytest.mark.require_user("root") 329 def test_create_vlan_plain(self): 330 """Creates 802.1Q VLAN interface in vlanXX and ifX fashion""" 331 self.require_module("if_vlan") 332 os_ifname = self.vnet.iface_alias_map["if1"].name 333 ifindex = socket.if_nametoindex(os_ifname) 334 335 flags = NlmNewFlags.NLM_F_EXCL.value | NlmNewFlags.NLM_F_CREATE.value 336 msg = NetlinkIflaMessage(self.helper, NlRtMsgType.RTM_NEWLINK.value) 337 msg.nl_hdr.nlmsg_flags = ( 338 flags | NlmBaseFlags.NLM_F_ACK.value | NlmBaseFlags.NLM_F_REQUEST.value 339 ) 340 msg.base_hdr.ifi_index = ifindex 341 342 msg.add_nla(NlAttrU32(IflattrType.IFLA_LINK, ifindex)) 343 msg.add_nla(NlAttrStr(IflattrType.IFLA_IFNAME, "vlan22")) 344 345 msg.add_nla( 346 NlAttrNested( 347 IflattrType.IFLA_LINKINFO, 348 [ 349 NlAttrStrn(IflinkInfo.IFLA_INFO_KIND, "vlan"), 350 NlAttrNested( 351 IflinkInfo.IFLA_INFO_DATA, 352 [ 353 NlAttrU16(IfLinkInfoDataVlan.IFLA_VLAN_ID, 22), 354 ], 355 ), 356 ], 357 ) 358 ) 359 360 rx_msg = self.get_reply(msg) 361 assert rx_msg.is_type(NlMsgType.NLMSG_ERROR) 362 assert rx_msg.error_code == 0 363 364 ToolsHelper.print_net_debug() 365 self.get_interface_byname("vlan22") 366 # ToolsHelper.print_net_debug() 367