summaryrefslogtreecommitdiffstats
path: root/test/test_nat.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/test_nat.py')
-rw-r--r--test/test_nat.py80
1 files changed, 76 insertions, 4 deletions
diff --git a/test/test_nat.py b/test/test_nat.py
index 1f87bffab57..aeeb5aa2bd5 100644
--- a/test/test_nat.py
+++ b/test/test_nat.py
@@ -242,33 +242,44 @@ class MethodHolder(VppTestCase):
return pkts
- def create_stream_out(self, out_if, dst_ip=None, ttl=64):
+ def create_stream_out(self, out_if, dst_ip=None, ttl=64,
+ use_inside_ports=False):
"""
Create packet stream for outside network
:param out_if: Outside interface
:param dst_ip: Destination IP address (Default use global NAT address)
:param ttl: TTL of generated packets
+ :param use_inside_ports: Use inside NAT ports as destination ports
+ instead of outside ports
"""
if dst_ip is None:
dst_ip = self.nat_addr
+ if not use_inside_ports:
+ tcp_port = self.tcp_port_out
+ udp_port = self.udp_port_out
+ icmp_id = self.icmp_id_out
+ else:
+ tcp_port = self.tcp_port_in
+ udp_port = self.udp_port_in
+ icmp_id = self.icmp_id_in
pkts = []
# TCP
p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
- TCP(dport=self.tcp_port_out, sport=20))
+ TCP(dport=tcp_port, sport=20))
pkts.append(p)
# UDP
p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
- UDP(dport=self.udp_port_out, sport=20))
+ UDP(dport=udp_port, sport=20))
pkts.append(p)
# ICMP
p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
- ICMP(id=self.icmp_id_out, type='echo-reply'))
+ ICMP(id=icmp_id, type='echo-reply'))
pkts.append(p)
return pkts
@@ -654,6 +665,7 @@ class TestNAT44(MethodHolder):
cls.icmp_id_in = 6305
cls.icmp_id_out = 6305
cls.nat_addr = '10.0.0.3'
+ cls.nat_addr_n = socket.inet_pton(socket.AF_INET, cls.nat_addr)
cls.ipfix_src_port = 4739
cls.ipfix_domain_id = 1
@@ -1044,6 +1056,66 @@ class TestNAT44(MethodHolder):
self.verify_capture_out(capture, same_port=True, packet_num=1)
self.assert_equal(capture[0][IP].proto, IP_PROTOS.icmp)
+ def test_forwarding(self):
+ """ NAT44 forwarding test """
+
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
+ self.vapi.nat44_forwarding_enable_disable(1)
+
+ real_ip = self.pg0.remote_ip4n
+ alias_ip = self.nat_addr_n
+ self.vapi.nat44_add_del_static_mapping(local_ip=real_ip,
+ external_ip=alias_ip)
+
+ try:
+ # in2out - static mapping match
+
+ pkts = self.create_stream_out(self.pg1)
+ self.pg1.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ self.verify_capture_in(capture, self.pg0)
+
+ 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, same_port=True)
+
+ # in2out - no static mapping match
+
+ host0 = self.pg0.remote_hosts[0]
+ self.pg0.remote_hosts[0] = self.pg0.remote_hosts[1]
+ try:
+ pkts = self.create_stream_out(self.pg1,
+ dst_ip=self.pg0.remote_ip4,
+ use_inside_ports=True)
+ self.pg1.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ self.verify_capture_in(capture, self.pg0)
+
+ 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, nat_ip=self.pg0.remote_ip4,
+ same_port=True)
+ finally:
+ self.pg0.remote_hosts[0] = host0
+
+ finally:
+ self.vapi.nat44_forwarding_enable_disable(0)
+ self.vapi.nat44_add_del_static_mapping(local_ip=real_ip,
+ external_ip=alias_ip,
+ is_add=0)
+
def test_static_in(self):
""" 1:1 NAT initialized from inside network """