| #!/usr/bin/env python3 |
| |
| import socket |
| import struct |
| import unittest |
| import scapy.compat |
| from time import sleep |
| from config import config |
| from framework import VppTestCase |
| from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder |
| from scapy.layers.inet import IP, TCP, UDP, ICMP |
| from scapy.layers.inet import IPerror, UDPerror |
| from scapy.layers.l2 import Ether |
| from util import ppp |
| |
| |
| class TestDET44(VppTestCase): |
| """Deterministic NAT Test Cases""" |
| |
| @classmethod |
| def setUpClass(cls): |
| super(TestDET44, cls).setUpClass() |
| cls.vapi.cli("set log class det44 level debug") |
| |
| cls.tcp_port_in = 6303 |
| cls.tcp_external_port = 6303 |
| cls.udp_port_in = 6304 |
| cls.udp_external_port = 6304 |
| cls.icmp_id_in = 6305 |
| cls.nat_addr = "10.0.0.3" |
| |
| cls.create_pg_interfaces(range(3)) |
| cls.interfaces = list(cls.pg_interfaces) |
| |
| for i in cls.interfaces: |
| i.admin_up() |
| i.config_ip4() |
| i.resolve_arp() |
| |
| cls.pg0.generate_remote_hosts(2) |
| cls.pg0.configure_ipv4_neighbors() |
| |
| @classmethod |
| def tearDownClass(cls): |
| super(TestDET44, cls).tearDownClass() |
| |
| def setUp(self): |
| super(TestDET44, self).setUp() |
| self.vapi.det44_plugin_enable_disable(enable=1) |
| |
| def tearDown(self): |
| super(TestDET44, self).tearDown() |
| if not self.vpp_dead: |
| self.vapi.det44_plugin_enable_disable(enable=0) |
| |
| def show_commands_at_teardown(self): |
| self.logger.info(self.vapi.cli("show det44 interfaces")) |
| self.logger.info(self.vapi.cli("show det44 timeouts")) |
| self.logger.info(self.vapi.cli("show det44 mappings")) |
| self.logger.info(self.vapi.cli("show det44 sessions")) |
| |
| def verify_capture_in(self, capture, in_if): |
| """ |
| Verify captured packets on inside network |
| |
| :param capture: Captured packets |
| :param in_if: Inside interface |
| """ |
| fired = False |
| for packet in capture: |
| try: |
| self.assert_packet_checksums_valid(packet) |
| self.assertEqual(packet[IP].dst, in_if.remote_ip4) |
| if packet.haslayer(TCP): |
| self.assertEqual(packet[TCP].dport, self.tcp_port_in) |
| elif packet.haslayer(UDP): |
| self.assertEqual(packet[UDP].dport, self.udp_port_in) |
| else: |
| self.assertEqual(packet[ICMP].id, self.icmp_id_in) |
| except: |
| fired = True |
| self.logger.error( |
| ppp("Unexpected or invalid packet (inside network):", packet) |
| ) |
| if fired: |
| raise |
| |
| def verify_ipfix_max_entries_per_user(self, data, limit, src_addr): |
| """ |
| Verify IPFIX maximum entries per user exceeded event |
| |
| :param data: Decoded IPFIX data records |
| :param limit: Number of maximum entries per user |
| :param src_addr: IPv4 source address |
| """ |
| self.assertEqual(1, len(data)) |
| record = data[0] |
| # natEvent |
| self.assertEqual(scapy.compat.orb(record[230]), 13) |
| # natQuotaExceededEvent |
| self.assertEqual(struct.pack("I", 3), record[466]) |
| # maxEntriesPerUser |
| self.assertEqual(struct.pack("I", limit), record[473]) |
| # sourceIPv4Address |
| self.assertEqual(socket.inet_pton(socket.AF_INET, src_addr), record[8]) |
| |
| def initiate_tcp_session(self, in_if, out_if): |
| """ |
| Initiates TCP session 3 WAY HAND SHAKE |
| |
| :param in_if: Inside interface |
| :param out_if: Outside interface |
| """ |
| |
| # SYN packet in->out |
| p = ( |
| Ether(src=in_if.remote_mac, dst=in_if.local_mac) |
| / IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) |
| / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port, flags="S") |
| ) |
| in_if.add_stream(p) |
| self.pg_enable_capture(self.pg_interfaces) |
| self.pg_start() |
| capture = out_if.get_capture(1) |
| p = capture[0] |
| self.tcp_port_out = p[TCP].sport |
| |
| # SYN + ACK packet out->in |
| p = ( |
| Ether(src=out_if.remote_mac, dst=out_if.local_mac) |
| / IP(src=out_if.remote_ip4, dst=self.nat_addr) |
| / TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, flags="SA") |
| ) |
| out_if.add_stream(p) |
| self.pg_enable_capture(self.pg_interfaces) |
| self.pg_start() |
| in_if.get_capture(1) |
| |
| # ACK packet in->out |
| p = ( |
| Ether(src=in_if.remote_mac, dst=in_if.local_mac) |
| / IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) |
| / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port, flags="A") |
| ) |
| in_if.add_stream(p) |
| self.pg_enable_capture(self.pg_interfaces) |
| self.pg_start() |
| out_if.get_capture(1) |
| |
| def create_stream_in(self, in_if, out_if, ttl=64): |
| """ |
| Create packet stream for inside network |
| |
| :param in_if: Inside interface |
| :param out_if: Outside interface |
| :param ttl: TTL of generated packets |
| """ |
| pkts = [] |
| # TCP |
| p = ( |
| Ether(dst=in_if.local_mac, src=in_if.remote_mac) |
| / IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) |
| / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port) |
| ) |
| pkts.append(p) |
| |
| # UDP |
| p = ( |
| Ether(dst=in_if.local_mac, src=in_if.remote_mac) |
| / IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) |
| / UDP(sport=self.udp_port_in, dport=self.udp_external_port) |
| ) |
| pkts.append(p) |
| |
| # ICMP |
| p = ( |
| Ether(dst=in_if.local_mac, src=in_if.remote_mac) |
| / IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) |
| / ICMP(id=self.icmp_id_in, type="echo-request") |
| ) |
| pkts.append(p) |
| |
| return pkts |
| |
| def create_stream_out(self, out_if, dst_ip=None, ttl=64): |
| """ |
| 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 |
| """ |
| if dst_ip is None: |
| dst_ip = self.nat_addr |
| 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=self.tcp_external_port) |
| ) |
| 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=self.udp_external_port) |
| ) |
| 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_external_id, type="echo-reply") |
| ) |
| pkts.append(p) |
| |
| return pkts |
| |
| def verify_capture_out(self, capture, nat_ip=None): |
| """ |
| Verify captured packets on outside network |
| |
| :param capture: Captured packets |
| :param nat_ip: Translated IP address (Default use global NAT address) |
| :param same_port: Source port number is not translated (Default False) |
| """ |
| if nat_ip is None: |
| nat_ip = self.nat_addr |
| for packet in capture: |
| try: |
| self.assertEqual(packet[IP].src, nat_ip) |
| if packet.haslayer(TCP): |
| self.tcp_port_out = packet[TCP].sport |
| elif packet.haslayer(UDP): |
| self.udp_port_out = packet[UDP].sport |
| else: |
| self.icmp_external_id = packet[ICMP].id |
| except: |
| self.logger.error( |
| ppp("Unexpected or invalid packet (outside network):", packet) |
| ) |
| raise |
| |
| def test_deterministic_mode(self): |
| """NAT plugin run deterministic mode""" |
| in_addr = "172.16.255.0" |
| out_addr = "172.17.255.50" |
| in_addr_t = "172.16.255.20" |
| in_plen = 24 |
| out_plen = 32 |
| |
| self.vapi.det44_add_del_map( |
| is_add=1, |
| in_addr=in_addr, |
| in_plen=in_plen, |
| out_addr=out_addr, |
| out_plen=out_plen, |
| ) |
| |
| rep1 = self.vapi.det44_forward(in_addr_t) |
| self.assertEqual(str(rep1.out_addr), out_addr) |
| rep2 = self.vapi.det44_reverse(rep1.out_port_hi, out_addr) |
| |
| self.assertEqual(str(rep2.in_addr), in_addr_t) |
| |
| deterministic_mappings = self.vapi.det44_map_dump() |
| self.assertEqual(len(deterministic_mappings), 1) |
| dsm = deterministic_mappings[0] |
| self.assertEqual(in_addr, str(dsm.in_addr)) |
| self.assertEqual(in_plen, dsm.in_plen) |
| self.assertEqual(out_addr, str(dsm.out_addr)) |
| self.assertEqual(out_plen, dsm.out_plen) |
| |
| def test_set_timeouts(self): |
| """Set deterministic NAT timeouts""" |
| timeouts_before = self.vapi.det44_get_timeouts() |
| |
| self.vapi.det44_set_timeouts( |
| udp=timeouts_before.udp + 10, |
| tcp_established=timeouts_before.tcp_established + 10, |
| tcp_transitory=timeouts_before.tcp_transitory + 10, |
| icmp=timeouts_before.icmp + 10, |
| ) |
| |
| timeouts_after = self.vapi.det44_get_timeouts() |
| |
| self.assertNotEqual(timeouts_before.udp, timeouts_after.udp) |
| self.assertNotEqual(timeouts_before.icmp, timeouts_after.icmp) |
| self.assertNotEqual( |
| timeouts_before.tcp_established, timeouts_after.tcp_established |
| ) |
| self.assertNotEqual( |
| timeouts_before.tcp_transitory, timeouts_after.tcp_transitory |
| ) |
| |
| def test_in(self): |
| """DET44 translation test (TCP, UDP, ICMP)""" |
| |
| nat_ip = "10.0.0.10" |
| |
| self.vapi.det44_add_del_map( |
| is_add=1, |
| in_addr=self.pg0.remote_ip4, |
| in_plen=32, |
| out_addr=socket.inet_aton(nat_ip), |
| out_plen=32, |
| ) |
| |
| self.vapi.det44_interface_add_del_feature( |
| sw_if_index=self.pg0.sw_if_index, is_add=1, is_inside=1 |
| ) |
| self.vapi.det44_interface_add_del_feature( |
| sw_if_index=self.pg1.sw_if_index, is_add=1, is_inside=0 |
| ) |
| |
| # in2out |
| 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) |
| |
| # out2in |
| pkts = self.create_stream_out(self.pg1, nat_ip) |
| 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) |
| |
| # session dump test |
| sessions = self.vapi.det44_session_dump(self.pg0.remote_ip4) |
| self.assertEqual(len(sessions), 3) |
| |
| # TCP session |
| s = sessions[0] |
| self.assertEqual(str(s.ext_addr), self.pg1.remote_ip4) |
| self.assertEqual(s.in_port, self.tcp_port_in) |
| self.assertEqual(s.out_port, self.tcp_port_out) |
| self.assertEqual(s.ext_port, self.tcp_external_port) |
| |
| # UDP session |
| s = sessions[1] |
| self.assertEqual(str(s.ext_addr), self.pg1.remote_ip4) |
| self.assertEqual(s.in_port, self.udp_port_in) |
| self.assertEqual(s.out_port, self.udp_port_out) |
| self.assertEqual(s.ext_port, self.udp_external_port) |
| |
| # ICMP session |
| s = sessions[2] |
| self.assertEqual(str(s.ext_addr), self.pg1.remote_ip4) |
| self.assertEqual(s.in_port, self.icmp_id_in) |
| self.assertEqual(s.out_port, self.icmp_external_id) |
| |
| def test_multiple_users(self): |
| """Deterministic NAT multiple users""" |
| |
| nat_ip = "10.0.0.10" |
| port_in = 80 |
| external_port = 6303 |
| |
| host0 = self.pg0.remote_hosts[0] |
| host1 = self.pg0.remote_hosts[1] |
| |
| self.vapi.det44_add_del_map( |
| is_add=1, |
| in_addr=host0.ip4, |
| in_plen=24, |
| out_addr=socket.inet_aton(nat_ip), |
| out_plen=32, |
| ) |
| self.vapi.det44_interface_add_del_feature( |
| sw_if_index=self.pg0.sw_if_index, is_add=1, is_inside=1 |
| ) |
| self.vapi.det44_interface_add_del_feature( |
| sw_if_index=self.pg1.sw_if_index, is_add=1, is_inside=0 |
| ) |
| |
| # host0 to out |
| p = ( |
| Ether(src=host0.mac, dst=self.pg0.local_mac) |
| / IP(src=host0.ip4, dst=self.pg1.remote_ip4) |
| / TCP(sport=port_in, dport=external_port) |
| ) |
| self.pg0.add_stream(p) |
| self.pg_enable_capture(self.pg_interfaces) |
| self.pg_start() |
| capture = self.pg1.get_capture(1) |
| p = capture[0] |
| try: |
| ip = p[IP] |
| tcp = p[TCP] |
| self.assertEqual(ip.src, nat_ip) |
| self.assertEqual(ip.dst, self.pg1.remote_ip4) |
| self.assertEqual(tcp.dport, external_port) |
| port_out0 = tcp.sport |
| except: |
| self.logger.error(ppp("Unexpected or invalid packet:", p)) |
| raise |
| |
| # host1 to out |
| p = ( |
| Ether(src=host1.mac, dst=self.pg0.local_mac) |
| / IP(src=host1.ip4, dst=self.pg1.remote_ip4) |
| / TCP(sport=port_in, dport=external_port) |
| ) |
| self.pg0.add_stream(p) |
| self.pg_enable_capture(self.pg_interfaces) |
| self.pg_start() |
| capture = self.pg1.get_capture(1) |
| p = capture[0] |
| try: |
| ip = p[IP] |
| tcp = p[TCP] |
| self.assertEqual(ip.src, nat_ip) |
| self.assertEqual(ip.dst, self.pg1.remote_ip4) |
| self.assertEqual(tcp.dport, external_port) |
| port_out1 = tcp.sport |
| except: |
| self.logger.error(ppp("Unexpected or invalid packet:", p)) |
| raise |
| |
| dms = self.vapi.det44_map_dump() |
| self.assertEqual(1, len(dms)) |
| self.assertEqual(2, dms[0].ses_num) |
| |
| # out to host0 |
| p = ( |
| Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) |
| / IP(src=self.pg1.remote_ip4, dst=nat_ip) |
| / TCP(sport=external_port, dport=port_out0) |
| ) |
| 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] |
| self.assertEqual(ip.src, self.pg1.remote_ip4) |
| self.assertEqual(ip.dst, host0.ip4) |
| self.assertEqual(tcp.dport, port_in) |
| self.assertEqual(tcp.sport, external_port) |
| except: |
| self.logger.error(ppp("Unexpected or invalid packet:", p)) |
| raise |
| |
| # out to host1 |
| p = ( |
| Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) |
| / IP(src=self.pg1.remote_ip4, dst=nat_ip) |
| / TCP(sport=external_port, dport=port_out1) |
| ) |
| 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] |
| self.assertEqual(ip.src, self.pg1.remote_ip4) |
| self.assertEqual(ip.dst, host1.ip4) |
| self.assertEqual(tcp.dport, port_in) |
| self.assertEqual(tcp.sport, external_port) |
| except: |
| self.logger.error(ppp("Unexpected or invalid packet", p)) |
| raise |
| |
| # session close api test |
| self.vapi.det44_close_session_out( |
| socket.inet_aton(nat_ip), port_out1, self.pg1.remote_ip4, external_port |
| ) |
| dms = self.vapi.det44_map_dump() |
| self.assertEqual(dms[0].ses_num, 1) |
| |
| self.vapi.det44_close_session_in( |
| host0.ip4, port_in, self.pg1.remote_ip4, external_port |
| ) |
| dms = self.vapi.det44_map_dump() |
| self.assertEqual(dms[0].ses_num, 0) |
| |
| def test_tcp_session_close_detection_in(self): |
| """DET44 TCP session close from inside network""" |
| self.vapi.det44_add_del_map( |
| is_add=1, |
| in_addr=self.pg0.remote_ip4, |
| in_plen=32, |
| out_addr=socket.inet_aton(self.nat_addr), |
| out_plen=32, |
| ) |
| self.vapi.det44_interface_add_del_feature( |
| sw_if_index=self.pg0.sw_if_index, is_add=1, is_inside=1 |
| ) |
| self.vapi.det44_interface_add_del_feature( |
| sw_if_index=self.pg1.sw_if_index, is_add=1, is_inside=0 |
| ) |
| |
| self.initiate_tcp_session(self.pg0, self.pg1) |
| |
| # close the session from inside |
| try: |
| # FIN packet in -> out |
| p = ( |
| Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) |
| / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) |
| / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port, flags="F") |
| ) |
| self.pg0.add_stream(p) |
| self.pg_enable_capture(self.pg_interfaces) |
| self.pg_start() |
| self.pg1.get_capture(1) |
| |
| pkts = [] |
| |
| # ACK packet out -> in |
| p = ( |
| Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) |
| / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) |
| / TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, flags="A") |
| ) |
| pkts.append(p) |
| |
| # FIN packet out -> in |
| p = ( |
| Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) |
| / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) |
| / TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, flags="F") |
| ) |
| pkts.append(p) |
| |
| self.pg1.add_stream(pkts) |
| self.pg_enable_capture(self.pg_interfaces) |
| self.pg_start() |
| self.pg0.get_capture(2) |
| |
| # ACK packet in -> out |
| p = ( |
| Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) |
| / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) |
| / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port, flags="A") |
| ) |
| self.pg0.add_stream(p) |
| self.pg_enable_capture(self.pg_interfaces) |
| self.pg_start() |
| self.pg1.get_capture(1) |
| |
| # Check if deterministic NAT44 closed the session |
| dms = self.vapi.det44_map_dump() |
| self.assertEqual(0, dms[0].ses_num) |
| except: |
| self.logger.error("TCP session termination failed") |
| raise |
| |
| def test_tcp_session_close_detection_out(self): |
| """Deterministic NAT TCP session close from outside network""" |
| self.vapi.det44_add_del_map( |
| is_add=1, |
| in_addr=self.pg0.remote_ip4, |
| in_plen=32, |
| out_addr=socket.inet_aton(self.nat_addr), |
| out_plen=32, |
| ) |
| self.vapi.det44_interface_add_del_feature( |
| sw_if_index=self.pg0.sw_if_index, is_add=1, is_inside=1 |
| ) |
| self.vapi.det44_interface_add_del_feature( |
| sw_if_index=self.pg1.sw_if_index, is_add=1, is_inside=0 |
| ) |
| |
| self.initiate_tcp_session(self.pg0, self.pg1) |
| |
| # close the session from outside |
| try: |
| # FIN packet out -> in |
| p = ( |
| Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) |
| / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) |
| / TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, flags="F") |
| ) |
| self.pg1.add_stream(p) |
| self.pg_enable_capture(self.pg_interfaces) |
| self.pg_start() |
| self.pg0.get_capture(1) |
| |
| pkts = [] |
| |
| # ACK packet in -> out |
| p = ( |
| Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) |
| / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) |
| / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port, flags="A") |
| ) |
| pkts.append(p) |
| |
| # ACK packet in -> out |
| p = ( |
| Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) |
| / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) |
| / TCP(sport=self.tcp_port_in, dport=self.tcp_external_port, flags="F") |
| ) |
| pkts.append(p) |
| |
| self.pg0.add_stream(pkts) |
| self.pg_enable_capture(self.pg_interfaces) |
| self.pg_start() |
| self.pg1.get_capture(2) |
| |
| # ACK packet out -> in |
| p = ( |
| Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) |
| / IP(src=self.pg1.remote_ip4, dst=self.nat_addr) |
| / TCP(sport=self.tcp_external_port, dport=self.tcp_port_out, flags="A") |
| ) |
| self.pg1.add_stream(p) |
| self.pg_enable_capture(self.pg_interfaces) |
| self.pg_start() |
| self.pg0.get_capture(1) |
| |
| # Check if deterministic NAT44 closed the session |
| dms = self.vapi.det44_map_dump() |
| self.assertEqual(0, dms[0].ses_num) |
| except: |
| self.logger.error("TCP session termination failed") |
| raise |
| |
| def test_session_timeout(self): |
| """Deterministic NAT session timeouts""" |
| self.vapi.det44_add_del_map( |
| is_add=1, |
| in_addr=self.pg0.remote_ip4, |
| in_plen=32, |
| out_addr=socket.inet_aton(self.nat_addr), |
| out_plen=32, |
| ) |
| self.vapi.det44_interface_add_del_feature( |
| sw_if_index=self.pg0.sw_if_index, is_add=1, is_inside=1 |
| ) |
| self.vapi.det44_interface_add_del_feature( |
| sw_if_index=self.pg1.sw_if_index, is_add=1, is_inside=0 |
| ) |
| |
| self.initiate_tcp_session(self.pg0, self.pg1) |
| self.vapi.det44_set_timeouts(udp=5, tcp_established=5, tcp_transitory=5, icmp=5) |
| 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(len(pkts)) |
| self.virtual_sleep(15) |
| |
| dms = self.vapi.det44_map_dump() |
| self.assertEqual(0, dms[0].ses_num) |
| |
| # TODO: ipfix needs to be separated from NAT base plugin |
| @unittest.skipUnless(config.extended, "part of extended tests") |
| def test_session_limit_per_user(self): |
| """Deterministic NAT maximum sessions per user limit""" |
| self.vapi.det44_add_del_map( |
| is_add=1, |
| in_addr=self.pg0.remote_ip4, |
| in_plen=32, |
| out_addr=socket.inet_aton(self.nat_addr), |
| out_plen=32, |
| ) |
| self.vapi.det44_interface_add_del_feature( |
| sw_if_index=self.pg0.sw_if_index, is_add=1, is_inside=1 |
| ) |
| self.vapi.det44_interface_add_del_feature( |
| sw_if_index=self.pg1.sw_if_index, is_add=1, is_inside=0 |
| ) |
| self.vapi.set_ipfix_exporter( |
| collector_address=self.pg2.remote_ip4, |
| src_address=self.pg2.local_ip4, |
| path_mtu=512, |
| template_interval=10, |
| ) |
| self.vapi.nat_ipfix_enable_disable(domain_id=1, src_port=4739, enable=1) |
| |
| pkts = [] |
| for port in range(1025, 2025): |
| p = ( |
| Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) |
| / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) |
| / UDP(sport=port, dport=port) |
| ) |
| pkts.append(p) |
| |
| self.pg0.add_stream(pkts) |
| self.pg_enable_capture(self.pg_interfaces) |
| self.pg_start() |
| self.pg1.get_capture(len(pkts)) |
| |
| p = ( |
| Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) |
| / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) |
| / UDP(sport=3001, dport=3002) |
| ) |
| self.pg0.add_stream(p) |
| self.pg_enable_capture(self.pg_interfaces) |
| self.pg_start() |
| self.pg1.assert_nothing_captured() |
| |
| # verify ICMP error packet |
| capture = self.pg0.get_capture(1) |
| p = capture[0] |
| self.assertTrue(p.haslayer(ICMP)) |
| icmp = p[ICMP] |
| self.assertEqual(icmp.type, 3) |
| self.assertEqual(icmp.code, 1) |
| self.assertTrue(icmp.haslayer(IPerror)) |
| inner_ip = icmp[IPerror] |
| self.assertEqual(inner_ip[UDPerror].sport, 3001) |
| self.assertEqual(inner_ip[UDPerror].dport, 3002) |
| |
| dms = self.vapi.det44_map_dump() |
| |
| self.assertEqual(1000, dms[0].ses_num) |
| |
| # verify IPFIX logging |
| self.vapi.ipfix_flush() |
| capture = self.pg2.get_capture(2) |
| ipfix = IPFIXDecoder() |
| # first load template |
| for p in capture: |
| self.assertTrue(p.haslayer(IPFIX)) |
| if p.haslayer(Template): |
| ipfix.add_template(p.getlayer(Template)) |
| # verify events in data set |
| for p in capture: |
| if p.haslayer(Data): |
| data = ipfix.decode_data_set(p.getlayer(Set)) |
| self.verify_ipfix_max_entries_per_user(data, 1000, self.pg0.remote_ip4) |
| self.vapi.nat_ipfix_enable_disable(domain_id=1, src_port=4739, enable=0) |