from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
from time import sleep
from util import ip4_range
-from vpp_mac import mactobinary
+from vpp_papi import mac_pton
from syslog_rfc5424_parser import SyslogMessage, ParseError
from syslog_rfc5424_parser.constants import SyslogFacility, SyslogSeverity
from vpp_papi_provider import SYSLOG_SEVERITY
from io import BytesIO
+from vpp_papi import VppEnum
+from scapy.all import bind_layers, Packet, ByteEnumField, ShortField, \
+ IPField, IntField, LongField, XByteField, FlagsField, FieldLenField, \
+ PacketListField
+
+
+# NAT HA protocol event data
+class Event(Packet):
+ name = "Event"
+ fields_desc = [ByteEnumField("event_type", None,
+ {1: "add", 2: "del", 3: "refresh"}),
+ ByteEnumField("protocol", None,
+ {0: "udp", 1: "tcp", 2: "icmp"}),
+ ShortField("flags", 0),
+ IPField("in_addr", None),
+ IPField("out_addr", None),
+ ShortField("in_port", None),
+ ShortField("out_port", None),
+ IPField("eh_addr", None),
+ IPField("ehn_addr", None),
+ ShortField("eh_port", None),
+ ShortField("ehn_port", None),
+ IntField("fib_index", None),
+ IntField("total_pkts", 0),
+ LongField("total_bytes", 0)]
+
+ def extract_padding(self, s):
+ return "", s
+
+
+# NAT HA protocol header
+class HANATStateSync(Packet):
+ name = "HA NAT state sync"
+ fields_desc = [XByteField("version", 1),
+ FlagsField("flags", 0, 8, ['ACK']),
+ FieldLenField("count", None, count_of="events"),
+ IntField("sequence_number", 1),
+ IntField("thread_index", 0),
+ PacketListField("events", [], Event,
+ count_from=lambda pkt:pkt.count)]
class MethodHolder(VppTestCase):
is_add=0)
for intf in [self.pg7, self.pg8]:
- neighbors = self.vapi.ip_neighbor_dump(intf.sw_if_index)
- for n in neighbors:
- self.vapi.ip_neighbor_add_del(intf.sw_if_index,
- n.mac_address,
- n.ip_address,
- is_add=0)
+ self.vapi.ip_neighbor_add_del(
+ intf.sw_if_index,
+ intf.remote_mac,
+ intf.remote_ip4,
+ flags=(VppEnum.vl_api_ip_neighbor_flags_t.
+ IP_API_NEIGHBOR_FLAG_STATIC),
+ is_add=0)
if self.pg7.has_ip4_config:
self.pg7.unconfig_ip4()
self.vapi.syslog_set_filter(SYSLOG_SEVERITY.EMERG)
+ self.vapi.nat_ha_set_listener('0.0.0.0', 0)
+ self.vapi.nat_ha_set_failover('0.0.0.0', 0)
+
interfaces = self.vapi.nat44_interface_dump()
for intf in interfaces:
if intf.is_inside > 1:
""" Verify that there is no NAT44 user """
users = self.vapi.nat44_user_dump()
self.assertEqual(len(users), 0)
+ users = self.statistics.get_counter('/nat44/total-users')
+ self.assertEqual(users[0][0], 0)
+ sessions = self.statistics.get_counter('/nat44/total-sessions')
+ self.assertEqual(sessions[0][0], 0)
def verify_ipfix_max_entries_per_user(self, data, limit, src_addr):
"""
message = data.decode('utf-8')
try:
message = SyslogMessage.parse(message)
+ except ParseError as e:
+ self.logger.error(e)
+ raise
+ else:
self.assertEqual(message.severity, SyslogSeverity.info)
self.assertEqual(message.appname, 'NAT')
self.assertEqual(message.msgid, 'APMADD' if is_add else 'APMDEL')
self.assertEqual(sd_params.get('PROTO'), "%d" % IP_PROTOS.tcp)
self.assertTrue(sd_params.get('SSUBIX') is not None)
self.assertEqual(sd_params.get('SVLAN'), '0')
- except ParseError as e:
- self.logger.error(e)
def verify_syslog_sess(self, data, is_add=True, is_ip6=False):
message = data.decode('utf-8')
try:
message = SyslogMessage.parse(message)
+ except ParseError as e:
+ self.logger.error(e)
+ raise
+ else:
self.assertEqual(message.severity, SyslogSeverity.info)
self.assertEqual(message.appname, 'NAT')
self.assertEqual(message.msgid, 'SADD' if is_add else 'SDEL')
self.assertEqual(sd_params.get('XDADDR'), self.pg1.remote_ip4)
self.assertEqual(sd_params.get('XDPORT'),
"%d" % self.tcp_external_port)
- except ParseError as e:
- self.logger.error(e)
def verify_mss_value(self, pkt, mss):
"""
cls.ipfix_src_port = 4739
cls.ipfix_domain_id = 1
cls.tcp_external_port = 80
+ cls.udp_external_port = 69
cls.create_pg_interfaces(range(10))
cls.interfaces = list(cls.pg_interfaces[0:4])
'/err/nat44-out2in/good out2in packets processed')
self.assertEqual(err - totaln, 3)
+ users = self.statistics.get_counter('/nat44/total-users')
+ self.assertEqual(users[0][0], 1)
+ sessions = self.statistics.get_counter('/nat44/total-sessions')
+ self.assertEqual(sessions[0][0], 3)
+
def test_dynamic_icmp_errors_in2out_ttl_1(self):
""" NAT44 handling of client packets with TTL=1 """
self.assertTrue(session.is_static)
self.assertEqual(session.inside_ip_address[0:4],
self.pg6.remote_ip4n)
- self.assertEqual(map(ord, session.outside_ip_address[0:4]),
- map(int, static_nat_ip.split('.')))
+ self.assertEqual(session.outside_ip_address,
+ socket.inet_pton(socket.AF_INET, static_nat_ip))
self.assertTrue(session.inside_port in
[self.tcp_port_in, self.udp_port_in,
self.icmp_id_in])
data = ipfix.decode_data_set(p.getlayer(Set))
self.verify_ipfix_addr_exhausted(data)
- @unittest.skipUnless(running_extended_tests(), "part of extended tests")
+ @unittest.skipUnless(running_extended_tests, "part of extended tests")
def test_ipfix_max_sessions(self):
""" IPFIX logging maximum session entries exceeded """
self.nat44_add_address(self.nat_addr)
def test_dynamic_ipless_interfaces(self):
""" NAT44 interfaces without configured IP address """
- self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
- mactobinary(self.pg7.remote_mac),
- self.pg7.remote_ip4n,
- is_static=1)
- self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index,
- mactobinary(self.pg8.remote_mac),
- self.pg8.remote_ip4n,
- is_static=1)
+ self.vapi.ip_neighbor_add_del(
+ self.pg7.sw_if_index,
+ self.pg7.remote_mac,
+ self.pg7.remote_ip4,
+ flags=(VppEnum.vl_api_ip_neighbor_flags_t.
+ IP_API_NEIGHBOR_FLAG_STATIC))
+ self.vapi.ip_neighbor_add_del(
+ self.pg8.sw_if_index,
+ self.pg8.remote_mac,
+ self.pg8.remote_ip4,
+ flags=(VppEnum.vl_api_ip_neighbor_flags_t.
+ IP_API_NEIGHBOR_FLAG_STATIC))
self.vapi.ip_add_del_route(dst_address=self.pg7.remote_ip4n,
dst_address_length=32,
def test_static_ipless_interfaces(self):
""" NAT44 interfaces without configured IP address - 1:1 NAT """
- self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
- mactobinary(self.pg7.remote_mac),
- self.pg7.remote_ip4n,
- is_static=1)
- self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index,
- mactobinary(self.pg8.remote_mac),
- self.pg8.remote_ip4n,
- is_static=1)
+ self.vapi.ip_neighbor_add_del(
+ self.pg7.sw_if_index,
+ self.pg7.remote_mac,
+ self.pg7.remote_ip4,
+ flags=(VppEnum.vl_api_ip_neighbor_flags_t.
+ IP_API_NEIGHBOR_FLAG_STATIC))
+ self.vapi.ip_neighbor_add_del(
+ self.pg8.sw_if_index,
+ self.pg8.remote_mac,
+ self.pg8.remote_ip4,
+ flags=(VppEnum.vl_api_ip_neighbor_flags_t.
+ IP_API_NEIGHBOR_FLAG_STATIC))
self.vapi.ip_add_del_route(dst_address=self.pg7.remote_ip4n,
dst_address_length=32,
self.udp_port_out = 30607
self.icmp_id_out = 30608
- self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
- mactobinary(self.pg7.remote_mac),
- self.pg7.remote_ip4n,
- is_static=1)
- self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index,
- mactobinary(self.pg8.remote_mac),
- self.pg8.remote_ip4n,
- is_static=1)
+ self.vapi.ip_neighbor_add_del(
+ self.pg7.sw_if_index,
+ self.pg7.remote_mac,
+ self.pg7.remote_ip4,
+ flags=(VppEnum.vl_api_ip_neighbor_flags_t.
+ IP_API_NEIGHBOR_FLAG_STATIC))
+ self.vapi.ip_neighbor_add_del(
+ self.pg8.sw_if_index,
+ self.pg8.remote_mac,
+ self.pg8.remote_ip4,
+ flags=(VppEnum.vl_api_ip_neighbor_flags_t.
+ IP_API_NEIGHBOR_FLAG_STATIC))
self.vapi.ip_add_del_route(dst_address=self.pg7.remote_ip4n,
dst_address_length=32,
self.pg1.resolve_arp()
self.pg2.resolve_arp()
- @unittest.skipUnless(running_extended_tests(), "part of extended tests")
+ @unittest.skipUnless(running_extended_tests, "part of extended tests")
def test_session_timeout(self):
""" NAT44 session timeouts """
self.nat44_add_address(self.nat_addr)
# Negotiated MSS value smaller than configured - unchanged
self.verify_mss_value(capture[0], 1400)
+ @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ def test_ha_send(self):
+ """ Send HA session synchronization events (active) """
+ self.nat44_add_address(self.nat_addr)
+ 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.nat_ha_set_listener(self.pg3.local_ip4, port=12345)
+ self.vapi.nat_ha_set_failover(self.pg3.remote_ip4, port=12346)
+ bind_layers(UDP, HANATStateSync, sport=12345)
+
+ # create sessions
+ 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)
+ # active send HA events
+ self.vapi.nat_ha_flush()
+ stats = self.statistics.get_counter('/nat44/ha/add-event-send')
+ self.assertEqual(stats[0][0], 3)
+ capture = self.pg3.get_capture(1)
+ p = capture[0]
+ self.assert_packet_checksums_valid(p)
+ try:
+ ip = p[IP]
+ udp = p[UDP]
+ hanat = p[HANATStateSync]
+ except IndexError:
+ self.logger.error(ppp("Invalid packet:", p))
+ raise
+ else:
+ self.assertEqual(ip.src, self.pg3.local_ip4)
+ self.assertEqual(ip.dst, self.pg3.remote_ip4)
+ self.assertEqual(udp.sport, 12345)
+ self.assertEqual(udp.dport, 12346)
+ self.assertEqual(hanat.version, 1)
+ self.assertEqual(hanat.thread_index, 0)
+ self.assertEqual(hanat.count, 3)
+ seq = hanat.sequence_number
+ for event in hanat.events:
+ self.assertEqual(event.event_type, 1)
+ self.assertEqual(event.in_addr, self.pg0.remote_ip4)
+ self.assertEqual(event.out_addr, self.nat_addr)
+ self.assertEqual(event.fib_index, 0)
+
+ # ACK received events
+ ack = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
+ IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
+ UDP(sport=12346, dport=12345) /
+ HANATStateSync(sequence_number=seq, flags='ACK'))
+ self.pg3.add_stream(ack)
+ self.pg_start()
+ stats = self.statistics.get_counter('/nat44/ha/ack-recv')
+ self.assertEqual(stats[0][0], 1)
+
+ # delete one session
+ self.pg_enable_capture(self.pg_interfaces)
+ self.vapi.nat44_del_session(self.pg0.remote_ip4n, self.tcp_port_in,
+ IP_PROTOS.tcp)
+ self.vapi.nat_ha_flush()
+ stats = self.statistics.get_counter('/nat44/ha/del-event-send')
+ self.assertEqual(stats[0][0], 1)
+ capture = self.pg3.get_capture(1)
+ p = capture[0]
+ try:
+ hanat = p[HANATStateSync]
+ except IndexError:
+ self.logger.error(ppp("Invalid packet:", p))
+ raise
+ else:
+ self.assertGreater(hanat.sequence_number, seq)
+
+ # do not send ACK, active retry send HA event again
+ self.pg_enable_capture(self.pg_interfaces)
+ sleep(12)
+ stats = self.statistics.get_counter('/nat44/ha/retry-count')
+ self.assertEqual(stats[0][0], 3)
+ stats = self.statistics.get_counter('/nat44/ha/missed-count')
+ self.assertEqual(stats[0][0], 1)
+ capture = self.pg3.get_capture(3)
+ for packet in capture:
+ self.assertEqual(packet, p)
+
+ # session counters refresh
+ pkts = self.create_stream_out(self.pg1)
+ self.pg1.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ self.pg0.get_capture(2)
+ self.vapi.nat_ha_flush()
+ stats = self.statistics.get_counter('/nat44/ha/refresh-event-send')
+ self.assertEqual(stats[0][0], 2)
+ capture = self.pg3.get_capture(1)
+ p = capture[0]
+ self.assert_packet_checksums_valid(p)
+ try:
+ ip = p[IP]
+ udp = p[UDP]
+ hanat = p[HANATStateSync]
+ except IndexError:
+ self.logger.error(ppp("Invalid packet:", p))
+ raise
+ else:
+ self.assertEqual(ip.src, self.pg3.local_ip4)
+ self.assertEqual(ip.dst, self.pg3.remote_ip4)
+ self.assertEqual(udp.sport, 12345)
+ self.assertEqual(udp.dport, 12346)
+ self.assertEqual(hanat.version, 1)
+ self.assertEqual(hanat.count, 2)
+ seq = hanat.sequence_number
+ for event in hanat.events:
+ self.assertEqual(event.event_type, 3)
+ self.assertEqual(event.out_addr, self.nat_addr)
+ self.assertEqual(event.fib_index, 0)
+ self.assertEqual(event.total_pkts, 2)
+ self.assertGreater(event.total_bytes, 0)
+
+ ack = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
+ IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
+ UDP(sport=12346, dport=12345) /
+ HANATStateSync(sequence_number=seq, flags='ACK'))
+ self.pg3.add_stream(ack)
+ self.pg_start()
+ stats = self.statistics.get_counter('/nat44/ha/ack-recv')
+ self.assertEqual(stats[0][0], 2)
+
+ def test_ha_recv(self):
+ """ Receive HA session synchronization events (passive) """
+ self.nat44_add_address(self.nat_addr)
+ 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.nat_ha_set_listener(self.pg3.local_ip4, port=12345)
+ bind_layers(UDP, HANATStateSync, sport=12345)
+
+ self.tcp_port_out = random.randint(1025, 65535)
+ self.udp_port_out = random.randint(1025, 65535)
+
+ # send HA session add events to failover/passive
+ p = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
+ IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
+ UDP(sport=12346, dport=12345) /
+ HANATStateSync(sequence_number=1, events=[
+ Event(event_type='add', protocol='tcp',
+ in_addr=self.pg0.remote_ip4, out_addr=self.nat_addr,
+ in_port=self.tcp_port_in, out_port=self.tcp_port_out,
+ eh_addr=self.pg1.remote_ip4,
+ ehn_addr=self.pg1.remote_ip4,
+ eh_port=self.tcp_external_port,
+ ehn_port=self.tcp_external_port, fib_index=0),
+ Event(event_type='add', protocol='udp',
+ in_addr=self.pg0.remote_ip4, out_addr=self.nat_addr,
+ in_port=self.udp_port_in, out_port=self.udp_port_out,
+ eh_addr=self.pg1.remote_ip4,
+ ehn_addr=self.pg1.remote_ip4,
+ eh_port=self.udp_external_port,
+ ehn_port=self.udp_external_port, fib_index=0)]))
+
+ self.pg3.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ # receive ACK
+ capture = self.pg3.get_capture(1)
+ p = capture[0]
+ try:
+ hanat = p[HANATStateSync]
+ except IndexError:
+ self.logger.error(ppp("Invalid packet:", p))
+ raise
+ else:
+ self.assertEqual(hanat.sequence_number, 1)
+ self.assertEqual(hanat.flags, 'ACK')
+ self.assertEqual(hanat.version, 1)
+ self.assertEqual(hanat.thread_index, 0)
+ stats = self.statistics.get_counter('/nat44/ha/ack-send')
+ self.assertEqual(stats[0][0], 1)
+ stats = self.statistics.get_counter('/nat44/ha/add-event-recv')
+ self.assertEqual(stats[0][0], 2)
+ users = self.statistics.get_counter('/nat44/total-users')
+ self.assertEqual(users[0][0], 1)
+ sessions = self.statistics.get_counter('/nat44/total-sessions')
+ self.assertEqual(sessions[0][0], 2)
+ users = self.vapi.nat44_user_dump()
+ self.assertEqual(len(users), 1)
+ self.assertEqual(users[0].ip_address, self.pg0.remote_ip4n)
+ # there should be 2 sessions created by HA
+ sessions = self.vapi.nat44_user_session_dump(users[0].ip_address,
+ users[0].vrf_id)
+ self.assertEqual(len(sessions), 2)
+ for session in sessions:
+ self.assertEqual(session.inside_ip_address, self.pg0.remote_ip4n)
+ self.assertEqual(session.outside_ip_address, self.nat_addr_n)
+ self.assertIn(session.inside_port,
+ [self.tcp_port_in, self.udp_port_in])
+ self.assertIn(session.outside_port,
+ [self.tcp_port_out, self.udp_port_out])
+ self.assertIn(session.protocol, [IP_PROTOS.tcp, IP_PROTOS.udp])
+
+ # send HA session delete event to failover/passive
+ p = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
+ IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
+ UDP(sport=12346, dport=12345) /
+ HANATStateSync(sequence_number=2, events=[
+ Event(event_type='del', protocol='udp',
+ in_addr=self.pg0.remote_ip4, out_addr=self.nat_addr,
+ in_port=self.udp_port_in, out_port=self.udp_port_out,
+ eh_addr=self.pg1.remote_ip4,
+ ehn_addr=self.pg1.remote_ip4,
+ eh_port=self.udp_external_port,
+ ehn_port=self.udp_external_port, fib_index=0)]))
+
+ self.pg3.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ # receive ACK
+ capture = self.pg3.get_capture(1)
+ p = capture[0]
+ try:
+ hanat = p[HANATStateSync]
+ except IndexError:
+ self.logger.error(ppp("Invalid packet:", p))
+ raise
+ else:
+ self.assertEqual(hanat.sequence_number, 2)
+ self.assertEqual(hanat.flags, 'ACK')
+ self.assertEqual(hanat.version, 1)
+ users = self.vapi.nat44_user_dump()
+ self.assertEqual(len(users), 1)
+ self.assertEqual(users[0].ip_address, self.pg0.remote_ip4n)
+ # now we should have only 1 session, 1 deleted by HA
+ sessions = self.vapi.nat44_user_session_dump(users[0].ip_address,
+ users[0].vrf_id)
+ self.assertEqual(len(sessions), 1)
+ stats = self.statistics.get_counter('/nat44/ha/del-event-recv')
+ self.assertEqual(stats[0][0], 1)
+
+ stats = self.statistics.get_counter('/err/nat-ha/pkts-processed')
+ self.assertEqual(stats, 2)
+
+ # send HA session refresh event to failover/passive
+ p = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
+ IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
+ UDP(sport=12346, dport=12345) /
+ HANATStateSync(sequence_number=3, events=[
+ Event(event_type='refresh', protocol='tcp',
+ in_addr=self.pg0.remote_ip4, out_addr=self.nat_addr,
+ in_port=self.tcp_port_in, out_port=self.tcp_port_out,
+ eh_addr=self.pg1.remote_ip4,
+ ehn_addr=self.pg1.remote_ip4,
+ eh_port=self.tcp_external_port,
+ ehn_port=self.tcp_external_port, fib_index=0,
+ total_bytes=1024, total_pkts=2)]))
+ self.pg3.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ # receive ACK
+ capture = self.pg3.get_capture(1)
+ p = capture[0]
+ try:
+ hanat = p[HANATStateSync]
+ except IndexError:
+ self.logger.error(ppp("Invalid packet:", p))
+ raise
+ else:
+ self.assertEqual(hanat.sequence_number, 3)
+ self.assertEqual(hanat.flags, 'ACK')
+ self.assertEqual(hanat.version, 1)
+ users = self.vapi.nat44_user_dump()
+ self.assertEqual(len(users), 1)
+ self.assertEqual(users[0].ip_address, self.pg0.remote_ip4n)
+ sessions = self.vapi.nat44_user_session_dump(users[0].ip_address,
+ users[0].vrf_id)
+ self.assertEqual(len(sessions), 1)
+ session = sessions[0]
+ self.assertEqual(session.total_bytes, 1024)
+ self.assertEqual(session.total_pkts, 2)
+ stats = self.statistics.get_counter('/nat44/ha/refresh-event-recv')
+ self.assertEqual(stats[0][0], 1)
+
+ stats = self.statistics.get_counter('/err/nat-ha/pkts-processed')
+ self.assertEqual(stats, 3)
+
+ # send packet to test session created by HA
+ p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+ IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
+ TCP(sport=self.tcp_external_port, dport=self.tcp_port_out))
+ self.pg1.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(1)
+ p = capture[0]
+ try:
+ ip = p[IP]
+ tcp = p[TCP]
+ except IndexError:
+ self.logger.error(ppp("Invalid packet:", p))
+ raise
+ else:
+ self.assertEqual(ip.src, self.pg1.remote_ip4)
+ self.assertEqual(ip.dst, self.pg0.remote_ip4)
+ self.assertEqual(tcp.sport, self.tcp_external_port)
+ self.assertEqual(tcp.dport, self.tcp_port_in)
+
def tearDown(self):
super(TestNAT44, self).tearDown()
if not self.vpp_dead:
self.logger.info(self.vapi.cli("show nat timeouts"))
self.logger.info(
self.vapi.cli("show nat addr-port-assignment-alg"))
+ self.logger.info(self.vapi.cli("show nat ha"))
self.clear_nat44()
self.vapi.cli("clear logging")
'/err/nat44-ed-out2in/good out2in packets processed')
self.assertEqual(err - totaln, 2)
+ users = self.statistics.get_counter('/nat44/total-users')
+ self.assertEqual(users[0][0], 1)
+ sessions = self.statistics.get_counter('/nat44/total-sessions')
+ self.assertEqual(sessions[0][0], 3)
+
def test_forwarding(self):
""" NAT44 forwarding test """
sessions = self.vapi.nat44_user_session_dump(server.ip4n, 0)
self.assertEqual(len(sessions), 0)
- @unittest.skipUnless(running_extended_tests(), "part of extended tests")
+ @unittest.skipUnless(running_extended_tests, "part of extended tests")
def test_static_lb_multi_clients(self):
""" NAT44 local service load balancing - multiple clients"""
local_port = 8080
server1 = self.pg0.remote_hosts[0]
server2 = self.pg0.remote_hosts[1]
+ server3 = self.pg0.remote_hosts[2]
locals = [{'addr': server1.ip4n,
'port': local_port,
server2_n += 1
self.assertGreater(server1_n, server2_n)
+ # add new back-end
+ self.vapi.nat44_lb_static_mapping_add_del_local(external_addr_n,
+ external_port,
+ server3.ip4n,
+ local_port,
+ IP_PROTOS.tcp,
+ 20)
+ server1_n = 0
+ server2_n = 0
+ server3_n = 0
+ clients = ip4_range(self.pg1.remote_ip4, 60, 110)
+ pkts = []
+ for client in clients:
+ p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
+ IP(src=client, dst=self.nat_addr) /
+ TCP(sport=12346, dport=external_port))
+ pkts.append(p)
+ self.assertGreater(len(pkts), 0)
+ self.pg1.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ for p in capture:
+ if p[IP].dst == server1.ip4:
+ server1_n += 1
+ elif p[IP].dst == server2.ip4:
+ server2_n += 1
+ else:
+ server3_n += 1
+ self.assertGreater(server1_n, 0)
+ self.assertGreater(server2_n, 0)
+ self.assertGreater(server3_n, 0)
+
+ # remove one back-end
+ self.vapi.nat44_lb_static_mapping_add_del_local(external_addr_n,
+ external_port,
+ server2.ip4n,
+ local_port,
+ IP_PROTOS.tcp,
+ 10,
+ is_add=0)
+ server1_n = 0
+ server2_n = 0
+ server3_n = 0
+ self.pg1.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ for p in capture:
+ if p[IP].dst == server1.ip4:
+ server1_n += 1
+ elif p[IP].dst == server2.ip4:
+ server2_n += 1
+ else:
+ server3_n += 1
+ self.assertGreater(server1_n, 0)
+ self.assertEqual(server2_n, 0)
+ self.assertGreater(server3_n, 0)
+
def test_static_lb_2(self):
""" NAT44 local service load balancing (asymmetrical rule) """
external_addr_n = socket.inet_pton(socket.AF_INET, self.nat_addr)
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
- @unittest.skipUnless(running_extended_tests(), "part of extended tests")
+ @unittest.skipUnless(running_extended_tests, "part of extended tests")
def test_session_timeout(self):
""" NAT44 session timeouts """
self.nat44_add_address(self.nat_addr)
nsessions = nsessions + user.nsessions
self.assertLess(nsessions, 2 * max_sessions)
- @unittest.skipUnless(running_extended_tests(), "part of extended tests")
+ @unittest.skipUnless(running_extended_tests, "part of extended tests")
def test_session_rst_timeout(self):
""" NAT44 session RST timeouts """
self.nat44_add_address(self.nat_addr)
self.assertEqual(users[0].ip_address, self.pg0.remote_ip4n)
self.assertEqual(users[0].nsessions, 1)
- @unittest.skipUnless(running_extended_tests(), "part of extended tests")
+ @unittest.skipUnless(running_extended_tests, "part of extended tests")
def test_session_limit_per_user(self):
""" Maximum sessions per user limit """
self.nat44_add_address(self.nat_addr)
self.src_ip6_pfx_len = 96
self.vapi.map_add_domain(self.dst_ip6_pfx_n, self.dst_ip6_pfx_len,
self.src_ip6_pfx_n, self.src_ip6_pfx_len,
- '\x00\x00\x00\x00', 0, is_translation=1,
- is_rfc6052=1)
+ '\x00\x00\x00\x00', 0)
@unittest.skip('Temporary disabled')
def test_464xlat_ce(self):
self.logger.error("TCP session termination failed")
raise
- @unittest.skipUnless(running_extended_tests(), "part of extended tests")
+ @unittest.skipUnless(running_extended_tests, "part of extended tests")
def test_session_timeout(self):
""" Deterministic NAT session timeouts """
self.vapi.nat_det_add_del_map(self.pg0.remote_ip4n,
dms = self.vapi.nat_det_map_dump()
self.assertEqual(0, dms[0].ses_num)
- @unittest.skipUnless(running_extended_tests(), "part of extended tests")
+ @unittest.skipUnless(running_extended_tests, "part of extended tests")
def test_session_limit_per_user(self):
""" Deterministic NAT maximum sessions per user limit """
self.vapi.nat_det_add_del_map(self.pg0.remote_ip4n,
self.assertEqual(bibe.i_port, in_port)
self.assertEqual(bibe.o_port, out_port)
self.assertEqual(static_bib_num, 1)
+ bibs = self.statistics.get_counter('/nat64/total-bibs')
+ self.assertEqual(bibs[0][0], 1)
self.vapi.nat64_add_del_static_bib(in_addr,
out_addr,
if bibe.is_static:
static_bib_num += 1
self.assertEqual(static_bib_num, 0)
+ bibs = self.statistics.get_counter('/nat64/total-bibs')
+ self.assertEqual(bibs[0][0], 0)
def test_set_timeouts(self):
""" Set NAT64 timeouts """
'/err/nat64-out2in/good out2in packets processed')
self.assertEqual(err - totaln, 3)
+ bibs = self.statistics.get_counter('/nat64/total-bibs')
+ self.assertEqual(bibs[0][0], 3)
+ sessions = self.statistics.get_counter('/nat64/total-sessions')
+ self.assertEqual(sessions[0][0], 3)
+
# in2out
pkts = self.create_stream_in_ip6(self.pg0, self.pg1)
self.pg0.add_stream(pkts)
self.assertEqual(ses_num_end - ses_num_start, 3)
- @unittest.skipUnless(running_extended_tests(), "part of extended tests")
+ @unittest.skipUnless(running_extended_tests, "part of extended tests")
def test_session_timeout(self):
""" NAT64 session timeout """
self.icmp_id_in = 1234
addresses = self.vapi.nat64_pool_addr_dump()
self.assertEqual(0, len(adresses))
- @unittest.skipUnless(running_extended_tests(), "part of extended tests")
+ @unittest.skipUnless(running_extended_tests, "part of extended tests")
def test_ipfix_max_bibs_sessions(self):
""" IPFIX logging maximum session and BIB entries exceeded """
max_bibs = 1280
vrf_id=prefix.vrf_id,
is_add=0)
+ bibs = self.statistics.get_counter('/nat64/total-bibs')
+ self.assertEqual(bibs[0][0], 0)
+ sessions = self.statistics.get_counter('/nat64/total-sessions')
+ self.assertEqual(sessions[0][0], 0)
+
def tearDown(self):
super(TestNAT64, self).tearDown()
if not self.vpp_dead:
message = data.decode('utf-8')
try:
message = SyslogMessage.parse(message)
+ except ParseError as e:
+ self.logger.error(e)
+ else:
self.assertEqual(message.severity, SyslogSeverity.info)
self.assertEqual(message.appname, 'NAT')
self.assertEqual(message.msgid, 'APMADD')
self.assertEqual(sd_params.get('PROTO'), "%d" % proto)
self.assertTrue(sd_params.get('SSUBIX') is not None)
self.assertEqual(sd_params.get('SV6ENC'), sv6enc)
- except ParseError as e:
- self.logger.error(e)
def test_dslite(self):
""" Test DS-Lite """
self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
self.assertTrue(capture.haslayer(ICMPv6EchoReply))
+ b4s = self.statistics.get_counter('/dslite/total-b4s')
+ self.assertEqual(b4s[0][0], 2)
+ sessions = self.statistics.get_counter('/dslite/total-sessions')
+ self.assertEqual(sessions[0][0], 3)
+
def tearDown(self):
super(TestDSlite, self).tearDown()
if not self.vpp_dead:
Code Review / vpp.git / blobdiff