1 from ipaddress import IPv4Network
3 from vpp_object import VppObject
4 from vpp_papi import VppEnum
5 from vpp_ip import INVALID_INDEX
6 from vpp_papi_provider import UnexpectedApiReturnValueError
9 class VppAclPlugin(VppObject):
11 def __init__(self, test, enable_intf_counters=False):
13 self.enable_intf_counters = enable_intf_counters
16 def enable_intf_counters(self):
17 return self._enable_intf_counters
19 @enable_intf_counters.setter
20 def enable_intf_counters(self, enable):
21 self.vapi.acl_stats_intf_counters_enable(enable=enable)
23 def add_vpp_config(self):
26 def remove_vpp_config(self):
29 def query_vpp_config(self):
33 return ("acl-plugin-%d" % (self._sw_if_index))
44 udp_sport_to = udp_sport_from + 5
45 udp_dport_from = 20000
46 udp_dport_to = udp_dport_from + 5000
48 tcp_sport_to = tcp_sport_from + 5
49 tcp_dport_from = 40000
50 tcp_dport_to = tcp_dport_from + 5000
53 udp_sport_to_2 = udp_sport_from_2 + 5
54 udp_dport_from_2 = 30000
55 udp_dport_to_2 = udp_dport_from_2 + 5000
56 tcp_sport_from_2 = 130
57 tcp_sport_to_2 = tcp_sport_from_2 + 5
58 tcp_dport_from_2 = 20000
59 tcp_dport_to_2 = tcp_dport_from_2 + 5000
61 icmp4_type = 8 # echo request
63 icmp6_type = 128 # echo request
73 def __init__(self, is_permit, src_prefix=IPv4Network('0.0.0.0/0'),
74 dst_prefix=IPv4Network('0.0.0.0/0'),
75 proto=0, ports=PORTS_ALL, sport_from=None, sport_to=None,
76 dport_from=None, dport_to=None):
77 self.is_permit = is_permit
78 self.src_prefix = src_prefix
79 self.dst_prefix = dst_prefix
82 # assign ports by range
84 # assign specified ports
86 self.sport_from = sport_from
88 self.sport_to = sport_to
90 self.dport_from = dport_from
92 self.dport_to = dport_to
95 new_rule = AclRule(self.is_permit, self.src_prefix, self.dst_prefix,
96 self._proto, self._ports, self.sport_from,
97 self.sport_to, self.dport_from, self.dport_to)
100 def update_ports(self):
101 if self._ports == self.PORTS_ALL:
104 self.sport_to = 65535
105 if self._proto == 1 or self._proto == 58:
107 self.dport_to = self.sport_to
108 elif self._ports == self.PORTS_RANGE:
109 if self._proto == VppEnum.vl_api_ip_proto_t.IP_API_PROTO_ICMP:
110 self.sport_from = self.icmp4_type
111 self.sport_to = self.icmp4_type
112 self.dport_from = self.icmp4_code
113 self.dport_to = self.icmp4_code
114 elif self._proto == VppEnum.vl_api_ip_proto_t.IP_API_PROTO_ICMP6:
115 self.sport_from = self.icmp6_type
116 self.sport_to = self.icmp6_type
117 self.dport_from = self.icmp6_code
118 self.dport_to = self.icmp6_code
119 elif self._proto == VppEnum.vl_api_ip_proto_t.IP_API_PROTO_TCP:
120 self.sport_from = self.tcp_sport_from
121 self.sport_to = self.tcp_sport_to
122 self.dport_from = self.tcp_dport_from
123 self.dport_to = self.tcp_dport_to
124 elif self._proto == VppEnum.vl_api_ip_proto_t.IP_API_PROTO_UDP:
125 self.sport_from = self.udp_sport_from
126 self.sport_to = self.udp_sport_to
127 self.dport_from = self.udp_dport_from
128 self.dport_to = self.udp_dport_to
129 elif self._ports == self.PORTS_RANGE_2:
130 if self._proto == VppEnum.vl_api_ip_proto_t.IP_API_PROTO_ICMP:
131 self.sport_from = self.icmp4_type_2
132 self.sport_to = self.icmp4_type_2
133 self.dport_from = self.icmp4_code_from_2
134 self.dport_to = self.icmp4_code_to_2
135 elif self._proto == VppEnum.vl_api_ip_proto_t.IP_API_PROTO_ICMP6:
136 self.sport_from = self.icmp6_type_2
137 self.sport_to = self.icmp6_type_2
138 self.dport_from = self.icmp6_code_from_2
139 self.dport_to = self.icmp6_code_to_2
140 elif self._proto == VppEnum.vl_api_ip_proto_t.IP_API_PROTO_TCP:
141 self.sport_from = self.tcp_sport_from_2
142 self.sport_to = self.tcp_sport_to_2
143 self.dport_from = self.tcp_dport_from_2
144 self.dport_to = self.tcp_dport_to_2
145 elif self._proto == VppEnum.vl_api_ip_proto_t.IP_API_PROTO_UDP:
146 self.sport_from = self.udp_sport_from_2
147 self.sport_to = self.udp_sport_to_2
148 self.dport_from = self.udp_dport_from_2
149 self.dport_to = self.udp_dport_to_2
151 self.sport_from = self._ports
152 self.sport_to = self._ports
153 self.dport_from = self._ports
154 self.dport_to = self._ports
161 def proto(self, proto):
170 def ports(self, ports):
175 return {'is_permit': self.is_permit, 'proto': self.proto,
176 'srcport_or_icmptype_first': self.sport_from,
177 'srcport_or_icmptype_last': self.sport_to,
178 'src_prefix': self.src_prefix,
179 'dstport_or_icmpcode_first': self.dport_from,
180 'dstport_or_icmpcode_last': self.dport_to,
181 'dst_prefix': self.dst_prefix}
184 class VppAcl(VppObject):
187 def __init__(self, test, rules, acl_index=INVALID_INDEX, tag=None):
189 self._acl_index = acl_index
199 return self._acl_index
203 return len(self._rules)
205 def encode_rules(self):
207 for rule in self._rules:
208 rules.append(rule.encode())
211 def add_vpp_config(self, expect_error=False):
213 reply = self._test.vapi.acl_add_replace(
214 acl_index=self._acl_index, tag=self.tag, count=self.count,
215 r=self.encode_rules())
216 self._acl_index = reply.acl_index
217 self._test.registry.register(self, self._test.logger)
219 self._test.fail("Unexpected api reply")
221 except UnexpectedApiReturnValueError:
223 self._test.fail("Unexpected api reply")
226 def modify_vpp_config(self, rules):
228 self.add_vpp_config()
230 def remove_vpp_config(self, expect_error=False):
232 self._test.vapi.acl_del(acl_index=self._acl_index)
234 self._test.fail("Unexpected api reply")
235 except UnexpectedApiReturnValueError:
237 self._test.fail("Unexpected api reply")
240 return self._test.vapi.acl_dump(acl_index=self._acl_index)
242 def query_vpp_config(self):
245 if rule.acl_index == self._acl_index:
250 return ("acl-%s-%d" % (self.tag, self._acl_index))
253 class VppEtypeWhitelist(VppObject):
254 """ VPP Etype Whitelist """
256 def __init__(self, test, sw_if_index, whitelist, n_input=0):
258 self.whitelist = whitelist
259 self.n_input = n_input
260 self._sw_if_index = sw_if_index
263 def sw_if_index(self):
264 return self._sw_if_index
268 return len(self.whitelist)
270 def add_vpp_config(self):
271 self._test.vapi.acl_interface_set_etype_whitelist(
272 sw_if_index=self._sw_if_index, count=self.count,
273 n_input=self.n_input, whitelist=self.whitelist)
274 self._test.registry.register(self, self._test.logger)
277 def remove_vpp_config(self):
278 self._test.vapi.acl_interface_set_etype_whitelist(
279 sw_if_index=self._sw_if_index, count=0, n_input=0, whitelist=[])
281 def query_vpp_config(self):
282 self._test.vapi.acl_interface_etype_whitelist_dump(
283 sw_if_index=self._sw_if_index)
287 return ("acl-etype_wl-%d" % (self._sw_if_index))
290 class VppAclInterface(VppObject):
291 """ VPP ACL Interface """
293 def __init__(self, test, sw_if_index, acls, n_input=0):
295 self._sw_if_index = sw_if_index
296 self.n_input = n_input
300 def sw_if_index(self):
301 return self._sw_if_index
305 return len(self.acls)
307 def encode_acls(self):
309 for acl in self.acls:
310 acls.append(acl.acl_index)
313 def add_vpp_config(self, expect_error=False):
315 reply = self._test.vapi.acl_interface_set_acl_list(
316 sw_if_index=self._sw_if_index, n_input=self.n_input,
317 count=self.count, acls=self.encode_acls())
318 self._test.registry.register(self, self._test.logger)
320 self._test.fail("Unexpected api reply")
322 except UnexpectedApiReturnValueError:
324 self._test.fail("Unexpected api reply")
327 def remove_vpp_config(self, expect_error=False):
329 reply = self._test.vapi.acl_interface_set_acl_list(
330 sw_if_index=self._sw_if_index, n_input=0, count=0, acls=[])
332 self._test.fail("Unexpected api reply")
333 except UnexpectedApiReturnValueError:
335 self._test.fail("Unexpected api reply")
337 def query_vpp_config(self):
338 dump = self._test.vapi.acl_interface_list_dump(
339 sw_if_index=self._sw_if_index)
340 for acl_list in dump:
341 if acl_list.count > 0:
346 return ("acl-if-list-%d" % (self._sw_if_index))
352 def __init__(self, is_permit, src_mac=0, src_mac_mask=0,
353 src_prefix=IPv4Network('0.0.0.0/0')):
354 self.is_permit = is_permit
355 self.src_mac = src_mac
356 self.src_mac_mask = src_mac_mask
357 self.src_prefix = src_prefix
360 return {'is_permit': self.is_permit, 'src_mac': self.src_mac,
361 'src_mac_mask': self.src_mac_mask,
362 'src_prefix': self.src_prefix}
365 class VppMacipAcl(VppObject):
366 """ Vpp Mac Ip ACL """
368 def __init__(self, test, rules, acl_index=INVALID_INDEX, tag=None):
370 self._acl_index = acl_index
376 return self._acl_index
384 return len(self._rules)
386 def encode_rules(self):
388 for rule in self._rules:
389 rules.append(rule.encode())
392 def add_vpp_config(self, expect_error=False):
394 reply = self._test.vapi.macip_acl_add_replace(
395 acl_index=self._acl_index, tag=self.tag, count=self.count,
396 r=self.encode_rules())
397 self._acl_index = reply.acl_index
398 self._test.registry.register(self, self._test.logger)
400 self._test.fail("Unexpected api reply")
402 except UnexpectedApiReturnValueError:
404 self._test.fail("Unexpected api reply")
407 def modify_vpp_config(self, rules):
409 self.add_vpp_config()
411 def remove_vpp_config(self, expect_error=False):
413 self._test.vapi.macip_acl_del(acl_index=self._acl_index)
415 self._test.fail("Unexpected api reply")
416 except UnexpectedApiReturnValueError:
418 self._test.fail("Unexpected api reply")
421 return self._test.vapi.macip_acl_dump(acl_index=self._acl_index)
423 def query_vpp_config(self):
426 if rule.acl_index == self._acl_index:
431 return ("macip-acl-%s-%d" % (self.tag, self._acl_index))
434 class VppMacipAclInterface(VppObject):
435 """ VPP Mac Ip ACL Interface """
437 def __init__(self, test, sw_if_index, acls):
439 self._sw_if_index = sw_if_index
443 def sw_if_index(self):
444 return self._sw_if_index
448 return len(self.acls)
450 def add_vpp_config(self):
451 for acl in self.acls:
452 self._test.vapi.macip_acl_interface_add_del(
453 is_add=True, sw_if_index=self._sw_if_index,
454 acl_index=acl.acl_index)
455 self._test.registry.register(self, self._test.logger)
457 def remove_vpp_config(self):
458 for acl in self.acls:
459 self._test.vapi.macip_acl_interface_add_del(
460 is_add=False, sw_if_index=self._sw_if_index,
461 acl_index=acl.acl_index)
464 return self._test.vapi.macip_acl_interface_list_dump(
465 sw_if_index=self._sw_if_index)
467 def query_vpp_config(self):
469 for acl_list in dump:
470 for acl_index in acl_list.acls:
471 if acl_index != INVALID_INDEX:
476 return ("macip-acl-if-list-%d" % (self._sw_if_index))