import unittest
from io import BytesIO
-from random import randint, shuffle, choice
+from random import randint, choice
import scapy.compat
from framework import VppTestCase, VppTestRunner
from scapy.packet import Raw
from syslog_rfc5424_parser import SyslogMessage, ParseError
from syslog_rfc5424_parser.constants import SyslogSeverity
-from util import ppp, ip4_range
+from util import ppp, pr, ip4_range
from vpp_acl import AclRule, VppAcl, VppAclInterface
from vpp_ip_route import VppIpRoute, VppRoutePath
from vpp_papi import VppEnum
+from util import StatsDiff
class TestNAT44ED(VppTestCase):
for r in rl:
r.add_vpp_config()
+ cls.no_diff = StatsDiff({
+ pg.sw_if_index: {
+ '/nat44-ed/in2out/fastpath/tcp': 0,
+ '/nat44-ed/in2out/fastpath/udp': 0,
+ '/nat44-ed/in2out/fastpath/icmp': 0,
+ '/nat44-ed/in2out/fastpath/drops': 0,
+ '/nat44-ed/in2out/slowpath/tcp': 0,
+ '/nat44-ed/in2out/slowpath/udp': 0,
+ '/nat44-ed/in2out/slowpath/icmp': 0,
+ '/nat44-ed/in2out/slowpath/drops': 0,
+ '/nat44-ed/in2out/fastpath/tcp': 0,
+ '/nat44-ed/in2out/fastpath/udp': 0,
+ '/nat44-ed/in2out/fastpath/icmp': 0,
+ '/nat44-ed/in2out/fastpath/drops': 0,
+ '/nat44-ed/in2out/slowpath/tcp': 0,
+ '/nat44-ed/in2out/slowpath/udp': 0,
+ '/nat44-ed/in2out/slowpath/icmp': 0,
+ '/nat44-ed/in2out/slowpath/drops': 0,
+ }
+ for pg in cls.pg_interfaces
+ })
+
def get_err_counter(self, path):
return self.statistics.get_err_counter(path)
self.nat_add_outside_interface(self.pg1)
# in2out (initiate connection)
- p1 = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
+ p1 = [Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
- UDP(sport=21, dport=20) / payload)
+ UDP(sport=21, dport=20) / payload,
+ Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
+ IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
+ TCP(sport=21, dport=20, flags="S") / payload,
+ Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
+ IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
+ ICMP(type='echo-request', id=7777) / payload,
+ ]
- self.pg0.add_stream(p1)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(1)[0]
+ capture = self.send_and_expect(self.pg0, p1, self.pg1)
# out2in (send error message)
- p2 = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
+ p2 = [Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
ICMP(type='dest-unreach', code='port-unreachable') /
- capture[IP:])
-
- self.pg1.add_stream(p2)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
+ c[IP:]
+ for c in capture]
- capture = self.pg0.get_capture(1)[0]
+ capture = self.send_and_expect(self.pg1, p2, self.pg0)
- self.logger.info(ppp("p1 packet:", p1))
- self.logger.info(ppp("p2 packet:", p2))
- self.logger.info(ppp("capture packet:", capture))
+ for c in capture:
+ try:
+ assert c[IP].dst == self.pg0.remote_ip4
+ assert c[IPerror].src == self.pg0.remote_ip4
+ except AssertionError as a:
+ raise AssertionError(
+ f"Packet {pr(c)} not translated properly") from a
def test_icmp_echo_reply_trailer(self):
""" ICMP echo reply with ethernet trailer"""
self.vapi.nat44_forwarding_enable_disable(enable=1)
self.nat_add_address(self.nat_addr)
- self.vapi.nat44_interface_add_del_output_feature(
- sw_if_index=self.pg1.sw_if_index, is_add=1,)
+ self.vapi.nat44_ed_add_del_output_interface(
+ sw_if_index=self.pg1.sw_if_index,
+ is_add=1)
# session initiated from service host - translate
pkts = self.create_stream_in(self.pg0, self.pg1)
self.nat_add_address(self.nat_addr)
self.nat_add_outside_interface(self.pg0)
- self.vapi.nat44_interface_add_del_output_feature(
- sw_if_index=self.pg1.sw_if_index, is_add=1)
+ self.vapi.nat44_ed_add_del_output_interface(
+ sw_if_index=self.pg1.sw_if_index,
+ is_add=1)
# in2out
pkts = self.create_stream_in(self.pg0, self.pg1)
# in2out
pkts = self.create_stream_in(self.pg0, self.pg1, ttl=1)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(len(pkts))
+ capture = self.send_and_expect_some(self.pg0, pkts, self.pg0)
for p in capture:
self.assertIn(ICMP, p)
self.assertEqual(p[ICMP].type, 11) # 11 == time-exceeded
self.nat_add_address(self.nat_addr)
flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_output_feature(
+ self.vapi.nat44_ed_add_del_output_interface(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
self.vapi.nat44_interface_add_del_feature(
def test_show_max_translations(self):
""" NAT44ED API test - max translations per thread """
- nat_config = self.vapi.nat_show_config_2()
+ config = self.vapi.nat44_show_running_config()
self.assertEqual(self.max_sessions,
- nat_config.max_translations_per_thread)
+ config.sessions)
def test_lru_cleanup(self):
""" NAT44ED LRU cleanup algorithm """
self.nat_add_outside_interface(self.pg1)
# in2out and no NAT addresses added
- err_old = self.statistics.get_err_counter(
- '/err/nat44-ed-in2out-slowpath/out of ports')
-
pkts = self.create_stream_in(self.pg0, self.pg1)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- self.pg1.get_capture(0, timeout=1)
- err_new = self.statistics.get_err_counter(
- '/err/nat44-ed-in2out-slowpath/out of ports')
-
- self.assertEqual(err_new - err_old, len(pkts))
+ self.send_and_assert_no_replies(
+ self.pg0, pkts, msg="i2o pkts",
+ stats_diff=self.no_diff | {
+ "err": {
+ '/err/nat44-ed-in2out-slowpath/out of ports': len(pkts),
+ },
+ self.pg0.sw_if_index: {
+ '/nat44-ed/in2out/slowpath/drops': len(pkts),
+ },
+ }
+ )
# in2out after NAT addresses added
self.nat_add_address(self.nat_addr)
- err_old = self.statistics.get_err_counter(
- '/err/nat44-ed-in2out-slowpath/out of ports')
-
- 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, ignore_port=True)
-
- err_new = self.statistics.get_err_counter(
- '/err/nat44-ed-in2out-slowpath/out of ports')
-
- self.assertEqual(err_new, err_old)
+ tcpn, udpn, icmpn = (sum(x) for x in
+ zip(*((TCP in p, UDP in p, ICMP in p)
+ for p in pkts)))
+
+ self.send_and_expect(
+ self.pg0, pkts, self.pg1, msg="i2o pkts",
+ stats_diff=self.no_diff | {
+ "err": {
+ '/err/nat44-ed-in2out-slowpath/out of ports': 0,
+ },
+ self.pg0.sw_if_index: {
+ '/nat44-ed/in2out/slowpath/drops': 0,
+ '/nat44-ed/in2out/slowpath/tcp': tcpn,
+ '/nat44-ed/in2out/slowpath/udp': udpn,
+ '/nat44-ed/in2out/slowpath/icmp': icmpn,
+ },
+ }
+ )
def test_unknown_proto(self):
""" NAT44ED translate packet with unknown protocol """
def test_tcp_close(self):
""" NAT44ED Close TCP session from inside network - output feature """
- old_timeouts = self.vapi.nat_get_timeouts()
+ config = self.vapi.nat44_show_running_config()
+ old_timeouts = config.timeouts
new_transitory = 2
self.vapi.nat_set_timeouts(
udp=old_timeouts.udp,
new_vrf_id = 22
self.nat_add_address(self.nat_addr)
- self.vapi.nat44_interface_add_del_output_feature(
- sw_if_index=self.pg8.sw_if_index, is_add=1)
-
+ self.vapi.nat44_ed_add_del_output_interface(
+ sw_if_index=self.pg8.sw_if_index,
+ is_add=1)
try:
self.configure_ip4_interface(self.pg7, table_id=new_vrf_id)
self.configure_ip4_interface(self.pg8, table_id=new_vrf_id)
Code Review / vpp.git / blobdiff