diff options
Diffstat (limited to 'src/plugins/nat/test/test_nat.py')
| -rw-r--r-- | src/plugins/nat/test/test_nat.py | 103 |
1 files changed, 103 insertions, 0 deletions
diff --git a/src/plugins/nat/test/test_nat.py b/src/plugins/nat/test/test_nat.py index 46b97c05dbe..f87065598f1 100644 --- a/src/plugins/nat/test/test_nat.py +++ b/src/plugins/nat/test/test_nat.py @@ -32,6 +32,7 @@ from scapy.all import bind_layers, Packet, ByteEnumField, ShortField, \ PacketListField from ipaddress import IPv6Network from util import ppc, ppp +from socket import inet_pton, AF_INET # NAT HA protocol event data @@ -6386,6 +6387,108 @@ class TestNAT44EndpointDependent(MethodHolder): capture = self.pg0.get_capture(len(pkts)) self.verify_capture_in(capture, self.pg0) + def test_output_feature_stateful_acl(self): + """ NAT44 endpoint-dependent output feature works with stateful ACL """ + self.nat44_add_address(self.nat_addr) + self.vapi.nat44_interface_add_del_output_feature( + sw_if_index=self.pg0.sw_if_index, + flags=self.config_flags.NAT_IS_INSIDE, + is_add=1) + self.vapi.nat44_interface_add_del_output_feature( + sw_if_index=self.pg1.sw_if_index, + flags=self.config_flags.NAT_IS_OUTSIDE, + is_add=1) + + # First ensure that the NAT is working sans ACL + + # send packets out2in, no sessions yet so packets should drop + pkts_out2in = self.create_stream_out(self.pg1) + self.send_and_assert_no_replies(self.pg1, pkts_out2in) + + # send packets into inside intf, ensure received via outside intf + pkts_in2out = self.create_stream_in(self.pg0, self.pg1) + capture = self.send_and_expect(self.pg0, pkts_in2out, self.pg1, + len(pkts_in2out)) + self.verify_capture_out(capture) + + # send out2in again, with sessions created it should work now + pkts_out2in = self.create_stream_out(self.pg1) + capture = self.send_and_expect(self.pg1, pkts_out2in, self.pg0, + len(pkts_out2in)) + self.verify_capture_in(capture, self.pg0) + + # Create an ACL blocking everything + out2in_deny_rule = { + 'is_permit': 0, + 'is_ipv6': 0, + 'src_ip_addr': inet_pton(AF_INET, "0.0.0.0"), + 'src_ip_prefix_len': 0, + 'dst_ip_addr': inet_pton(AF_INET, "0.0.0.0"), + 'dst_ip_prefix_len': 0, + 'srcport_or_icmptype_first': 0, + 'srcport_or_icmptype_last': 65535, + 'dstport_or_icmpcode_first': 0, + 'dstport_or_icmpcode_last': 65535, + 'proto': 0, + } + out2in_rules = [out2in_deny_rule] + res = self.vapi.acl_add_replace(0xffffffff, out2in_rules) + self.assertEqual(res.retval, 0, "error adding out2in ACL") + out2in_acl = res.acl_index + + # apply as input acl on interface and confirm it blocks everything + self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg1.sw_if_index, + n_input=1, + acls=[out2in_acl]) + self.send_and_assert_no_replies(self.pg1, pkts_out2in) + + # create an ACL to permit/reflect everything + in2out_reflect_rule = { + 'is_permit': 2, + 'is_ipv6': 0, + 'src_ip_addr': inet_pton(AF_INET, "0.0.0.0"), + 'src_ip_prefix_len': 0, + 'dst_ip_addr': inet_pton(AF_INET, "0.0.0.0"), + 'dst_ip_prefix_len': 0, + 'srcport_or_icmptype_first': 0, + 'srcport_or_icmptype_last': 65535, + 'dstport_or_icmpcode_first': 0, + 'dstport_or_icmpcode_last': 65535, + 'proto': 0, + } + in2out_rules = [in2out_reflect_rule] + res = self.vapi.acl_add_replace(0xffffffff, in2out_rules) + self.assertEqual(res.retval, 0, "error adding in2out ACL") + in2out_acl = res.acl_index + + # apply output acl + self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg1.sw_if_index, + n_input=1, + acls=[out2in_acl, in2out_acl]) + # send in2out to generate ACL state (NAT state was created earlier) + capture = self.send_and_expect(self.pg0, pkts_in2out, self.pg1, + len(pkts_in2out)) + self.verify_capture_out(capture) + + # send out2in again. ACL state exists so it should work now. + # TCP packets with the syn flag set also need the ack flag + for p in pkts_out2in: + if p.haslayer(TCP) and p[TCP].flags & 0x02: + p[TCP].flags |= 0x10 + capture = self.send_and_expect(self.pg1, pkts_out2in, self.pg0, + len(pkts_out2in)) + self.verify_capture_in(capture, self.pg0) + self.logger.info(self.vapi.cli("show trace")) + + # Clean up + # Remove ACLs from interface + self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg1.sw_if_index, + n_input=0, + acls=[]) + # delete ACLs + self.vapi.acl_del(acl_index=out2in_acl, expected_retval=0) + self.vapi.acl_del(acl_index=in2out_acl, expected_retval=0) + def test_multiple_vrf(self): """ Multiple VRF setup """ external_addr = '1.2.3.4' |