diff options
Diffstat (limited to 'src/plugins/nat/test')
| -rw-r--r-- | src/plugins/nat/test/test_nat.py | 114 |
1 files changed, 64 insertions, 50 deletions
diff --git a/src/plugins/nat/test/test_nat.py b/src/plugins/nat/test/test_nat.py index 6dee818d4bb..e996373342a 100644 --- a/src/plugins/nat/test/test_nat.py +++ b/src/plugins/nat/test/test_nat.py @@ -448,7 +448,7 @@ class MethodHolder(VppTestCase): return pkts def verify_capture_out(self, capture, nat_ip=None, same_port=False, - dst_ip=None, is_ip6=False): + dst_ip=None, is_ip6=False, ignore_port=False): """ Verify captured packets on outside network @@ -474,25 +474,32 @@ class MethodHolder(VppTestCase): if dst_ip is not None: self.assertEqual(packet[IP46].dst, dst_ip) if packet.haslayer(TCP): - if same_port: - self.assertEqual(packet[TCP].sport, self.tcp_port_in) - else: - self.assertNotEqual( - packet[TCP].sport, self.tcp_port_in) + if not ignore_port: + if same_port: + self.assertEqual( + packet[TCP].sport, self.tcp_port_in) + else: + self.assertNotEqual( + packet[TCP].sport, self.tcp_port_in) self.tcp_port_out = packet[TCP].sport self.assert_packet_checksums_valid(packet) elif packet.haslayer(UDP): - if same_port: - self.assertEqual(packet[UDP].sport, self.udp_port_in) - else: - self.assertNotEqual( - packet[UDP].sport, self.udp_port_in) + if not ignore_port: + if same_port: + self.assertEqual( + packet[UDP].sport, self.udp_port_in) + else: + self.assertNotEqual( + packet[UDP].sport, self.udp_port_in) self.udp_port_out = packet[UDP].sport else: - if same_port: - self.assertEqual(packet[ICMP46].id, self.icmp_id_in) - else: - self.assertNotEqual(packet[ICMP46].id, self.icmp_id_in) + if not ignore_port: + if same_port: + self.assertEqual( + packet[ICMP46].id, self.icmp_id_in) + else: + self.assertNotEqual( + packet[ICMP46].id, self.icmp_id_in) self.icmp_id_out = packet[ICMP46].id self.assert_packet_checksums_valid(packet) except: @@ -1105,7 +1112,8 @@ class MethodHolder(VppTestCase): else: raise Exception("Unsupported protocol") - def frag_in_order(self, proto=IP_PROTOS.tcp, dont_translate=False): + def frag_in_order(self, proto=IP_PROTOS.tcp, dont_translate=False, + ignore_port=False): layer = self.proto2layer(proto) if proto == IP_PROTOS.tcp: @@ -1132,14 +1140,16 @@ class MethodHolder(VppTestCase): if proto != IP_PROTOS.icmp: if not dont_translate: self.assertEqual(p[layer].dport, 20) - self.assertNotEqual(p[layer].sport, self.port_in) + if not ignore_port: + self.assertNotEqual(p[layer].sport, self.port_in) else: self.assertEqual(p[layer].sport, self.port_in) else: - if not dont_translate: - self.assertNotEqual(p[layer].id, self.port_in) - else: - self.assertEqual(p[layer].id, self.port_in) + if not ignore_port: + if not dont_translate: + self.assertNotEqual(p[layer].id, self.port_in) + else: + self.assertEqual(p[layer].id, self.port_in) self.assertEqual(data, p[Raw].load) # out2in @@ -1220,7 +1230,7 @@ class MethodHolder(VppTestCase): self.assertEqual(p[layer].id, self.port_in) self.assertEqual(data, p[Raw].load) - def reass_hairpinning(self, proto=IP_PROTOS.tcp): + def reass_hairpinning(self, proto=IP_PROTOS.tcp, ignore_port=False): layer = self.proto2layer(proto) if proto == IP_PROTOS.tcp: @@ -1243,13 +1253,16 @@ class MethodHolder(VppTestCase): self.nat_addr, self.server.ip4) if proto != IP_PROTOS.icmp: - self.assertNotEqual(p[layer].sport, self.host_in_port) + if not ignore_port: + self.assertNotEqual(p[layer].sport, self.host_in_port) self.assertEqual(p[layer].dport, self.server_in_port) else: - self.assertNotEqual(p[layer].id, self.host_in_port) + if not ignore_port: + self.assertNotEqual(p[layer].id, self.host_in_port) self.assertEqual(data, p[Raw].load) - def frag_out_of_order(self, proto=IP_PROTOS.tcp, dont_translate=False): + def frag_out_of_order(self, proto=IP_PROTOS.tcp, dont_translate=False, + ignore_port=False): layer = self.proto2layer(proto) if proto == IP_PROTOS.tcp: @@ -1278,14 +1291,16 @@ class MethodHolder(VppTestCase): if proto != IP_PROTOS.icmp: if not dont_translate: self.assertEqual(p[layer].dport, 20) - self.assertNotEqual(p[layer].sport, self.port_in) + if not ignore_port: + self.assertNotEqual(p[layer].sport, self.port_in) else: self.assertEqual(p[layer].sport, self.port_in) else: - if not dont_translate: - self.assertNotEqual(p[layer].id, self.port_in) - else: - self.assertEqual(p[layer].id, self.port_in) + if not ignore_port: + if not dont_translate: + self.assertNotEqual(p[layer].id, self.port_in) + else: + self.assertEqual(p[layer].id, self.port_in) self.assertEqual(data, p[Raw].load) # out2in @@ -4437,9 +4452,9 @@ class TestNAT44EndpointDependent(MethodHolder): self.vapi.nat44_interface_add_del_feature( sw_if_index=self.pg1.sw_if_index, is_add=1) - self.frag_in_order(proto=IP_PROTOS.tcp) - self.frag_in_order(proto=IP_PROTOS.udp) - self.frag_in_order(proto=IP_PROTOS.icmp) + self.frag_in_order(proto=IP_PROTOS.tcp, ignore_port=True) + self.frag_in_order(proto=IP_PROTOS.udp, ignore_port=True) + self.frag_in_order(proto=IP_PROTOS.icmp, ignore_port=True) def test_frag_in_order_dont_translate(self): """ NAT44 don't translate fragments arriving in order """ @@ -4463,9 +4478,9 @@ class TestNAT44EndpointDependent(MethodHolder): self.vapi.nat44_interface_add_del_feature( sw_if_index=self.pg1.sw_if_index, is_add=1) - self.frag_out_of_order(proto=IP_PROTOS.tcp) - self.frag_out_of_order(proto=IP_PROTOS.udp) - self.frag_out_of_order(proto=IP_PROTOS.icmp) + self.frag_out_of_order(proto=IP_PROTOS.tcp, ignore_port=True) + self.frag_out_of_order(proto=IP_PROTOS.udp, ignore_port=True) + self.frag_out_of_order(proto=IP_PROTOS.icmp, ignore_port=True) def test_frag_out_of_order_dont_translate(self): """ NAT44 don't translate fragments arriving out of order """ @@ -4593,9 +4608,9 @@ class TestNAT44EndpointDependent(MethodHolder): proto=IP_PROTOS.udp) self.nat44_add_static_mapping(self.server.ip4, self.nat_addr) - self.reass_hairpinning(proto=IP_PROTOS.tcp) - self.reass_hairpinning(proto=IP_PROTOS.udp) - self.reass_hairpinning(proto=IP_PROTOS.icmp) + self.reass_hairpinning(proto=IP_PROTOS.tcp, ignore_port=True) + self.reass_hairpinning(proto=IP_PROTOS.udp, ignore_port=True) + self.reass_hairpinning(proto=IP_PROTOS.icmp, ignore_port=True) def test_clear_sessions(self): """ NAT44 ED session clearing test """ @@ -4617,7 +4632,7 @@ class TestNAT44EndpointDependent(MethodHolder): self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) - self.verify_capture_out(capture) + self.verify_capture_out(capture, ignore_port=True) sessions = self.statistics.get_counter('/nat44/total-sessions') self.assertTrue(sessions[0][0] > 0) @@ -4664,7 +4679,7 @@ class TestNAT44EndpointDependent(MethodHolder): self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) - self.verify_capture_out(capture) + self.verify_capture_out(capture, ignore_port=True) err = self.statistics.get_err_counter( '/err/nat44-ed-in2out-slowpath/TCP packets') @@ -4752,7 +4767,7 @@ class TestNAT44EndpointDependent(MethodHolder): self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) - self.verify_capture_out(capture) + self.verify_capture_out(capture, ignore_port=True) err_new = self.statistics.get_err_counter( '/err/nat44-ed-in2out-slowpath/out of ports') @@ -4806,7 +4821,7 @@ class TestNAT44EndpointDependent(MethodHolder): self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg8.get_capture(len(pkts)) - self.verify_capture_out(capture) + self.verify_capture_out(capture, ignore_port=True) err = self.statistics.get_err_counter( '/err/nat44-ed-in2out-slowpath/TCP packets') @@ -5555,13 +5570,13 @@ class TestNAT44EndpointDependent(MethodHolder): self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) - self.verify_capture_out(capture) + self.verify_capture_out(capture, ignore_port=True) pkts = self.create_stream_in(self.pg0, self.pg1) self.pg0.add_stream(pkts) self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) - self.verify_capture_out(capture) + self.verify_capture_out(capture, ignore_port=True) # from external network back to local network host pkts = self.create_stream_out(self.pg1) @@ -5585,7 +5600,7 @@ class TestNAT44EndpointDependent(MethodHolder): self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) - self.verify_capture_out(capture) + self.verify_capture_out(capture, ignore_port=True) pkts = self.create_stream_out(self.pg1) self.pg1.add_stream(pkts) @@ -6595,7 +6610,7 @@ class TestNAT44EndpointDependent(MethodHolder): self.pg_enable_capture(self.pg_interfaces) self.pg_start() capture = self.pg1.get_capture(len(pkts)) - self.verify_capture_out(capture) + self.verify_capture_out(capture, ignore_port=True) # out2in pkts = self.create_stream_out(self.pg1) @@ -6627,7 +6642,7 @@ class TestNAT44EndpointDependent(MethodHolder): 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) + self.verify_capture_out(capture, ignore_port=True) # send out2in again, with sessions created it should work now pkts_out2in = self.create_stream_out(self.pg1) @@ -6657,7 +6672,7 @@ class TestNAT44EndpointDependent(MethodHolder): # 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) + self.verify_capture_out(capture, ignore_port=True) # send out2in again. ACL state exists so it should work now. # TCP packets with the syn flag set also need the ack flag @@ -6762,7 +6777,6 @@ class TestNAT44EndpointDependent(MethodHolder): ip = p[IP] tcp = p[TCP] self.assertEqual(ip.src, self.nat_addr) - self.assertNotEqual(tcp.sport, 2345) self.assert_packet_checksums_valid(p) port = tcp.sport except: |