test/test_det44.py

   1 #!/usr/bin/env python3
   2
   3 import socket
   4 import struct
   5 import unittest
   6 import scapy.compat
   7 from framework import VppTestCase, running_extended_tests
   8 from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
   9 from scapy.layers.inet import IP, TCP, UDP, ICMP
  10 from scapy.layers.inet import IPerror, UDPerror
  11 from scapy.layers.l2 import Ether
  12 from util import ppp
  13
  14
  15 class TestDET44(VppTestCase):
  16     """ Deterministic NAT Test Cases """
  17
  18     @classmethod
  19     def setUpClass(cls):
  20         super(TestDET44, cls).setUpClass()
  21         cls.vapi.cli("set log class det44 level debug")
  22
  23         cls.tcp_port_in = 6303
  24         cls.tcp_external_port = 6303
  25         cls.udp_port_in = 6304
  26         cls.udp_external_port = 6304
  27         cls.icmp_id_in = 6305
  28         cls.nat_addr = '10.0.0.3'
  29
  30         cls.create_pg_interfaces(range(3))
  31         cls.interfaces = list(cls.pg_interfaces)
  32
  33         for i in cls.interfaces:
  34             i.admin_up()
  35             i.config_ip4()
  36             i.resolve_arp()
  37
  38         cls.pg0.generate_remote_hosts(2)
  39         cls.pg0.configure_ipv4_neighbors()
  40
  41     @classmethod
  42     def tearDownClass(cls):
  43         super(TestDET44, cls).tearDownClass()
  44
  45     def setUp(self):
  46         super(TestDET44, self).setUp()
  47         self.vapi.det44_plugin_enable_disable(enable=1)
  48
  49     def tearDown(self):
  50         super(TestDET44, self).tearDown()
  51         if not self.vpp_dead:
  52             self.vapi.det44_plugin_enable_disable(enable=0)
  53
  54     def show_commands_at_teardown(self):
  55         self.logger.info(self.vapi.cli("show det44 interfaces"))
  56         self.logger.info(self.vapi.cli("show det44 timeouts"))
  57         self.logger.info(self.vapi.cli("show det44 mappings"))
  58         self.logger.info(self.vapi.cli("show det44 sessions"))
  59
  60     def verify_capture_in(self, capture, in_if):
  61         """
  62         Verify captured packets on inside network
  63
  64         :param capture: Captured packets
  65         :param in_if: Inside interface
  66         """
  67         fired = False
  68         for packet in capture:
  69             try:
  70                 self.assert_packet_checksums_valid(packet)
  71                 self.assertEqual(packet[IP].dst, in_if.remote_ip4)
  72                 if packet.haslayer(TCP):
  73                     self.assertEqual(packet[TCP].dport, self.tcp_port_in)
  74                 elif packet.haslayer(UDP):
  75                     self.assertEqual(packet[UDP].dport, self.udp_port_in)
  76                 else:
  77                     self.assertEqual(packet[ICMP].id, self.icmp_id_in)
  78             except:
  79                 fired = True
  80                 self.logger.error(ppp("Unexpected or invalid packet "
  81                                       "(inside network):", packet))
  82         if fired:
  83             raise
  84
  85     def verify_ipfix_max_entries_per_user(self, data, limit, src_addr):
  86         """
  87         Verify IPFIX maximum entries per user exceeded event
  88
  89         :param data: Decoded IPFIX data records
  90         :param limit: Number of maximum entries per user
  91         :param src_addr: IPv4 source address
  92         """
  93         self.assertEqual(1, len(data))
  94         record = data[0]
  95         # natEvent
  96         self.assertEqual(scapy.compat.orb(record[230]), 13)
  97         # natQuotaExceededEvent
  98         self.assertEqual(struct.pack("I", 3), record[466])
  99         # maxEntriesPerUser
 100         self.assertEqual(struct.pack("I", limit), record[473])
 101         # sourceIPv4Address
 102         self.assertEqual(socket.inet_pton(socket.AF_INET, src_addr), record[8])
 103
 104     def initiate_tcp_session(self, in_if, out_if):
 105         """
 106         Initiates TCP session 3 WAY HAND SHAKE
 107
 108         :param in_if: Inside interface
 109         :param out_if: Outside interface
 110         """
 111
 112         # SYN packet in->out
 113         p = (Ether(src=in_if.remote_mac, dst=in_if.local_mac) /
 114              IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) /
 115              TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
 116              flags="S"))
 117         in_if.add_stream(p)
 118         self.pg_enable_capture(self.pg_interfaces)
 119         self.pg_start()
 120         capture = out_if.get_capture(1)
 121         p = capture[0]
 122         self.tcp_port_out = p[TCP].sport
 123
 124         # SYN + ACK packet out->in
 125         p = (Ether(src=out_if.remote_mac, dst=out_if.local_mac) /
 126              IP(src=out_if.remote_ip4, dst=self.nat_addr) /
 127              TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
 128              flags="SA"))
 129         out_if.add_stream(p)
 130         self.pg_enable_capture(self.pg_interfaces)
 131         self.pg_start()
 132         in_if.get_capture(1)
 133
 134         # ACK packet in->out
 135         p = (Ether(src=in_if.remote_mac, dst=in_if.local_mac) /
 136              IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) /
 137              TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
 138              flags="A"))
 139         in_if.add_stream(p)
 140         self.pg_enable_capture(self.pg_interfaces)
 141         self.pg_start()
 142         out_if.get_capture(1)
 143
 144     def create_stream_in(self, in_if, out_if, ttl=64):
 145         """
 146         Create packet stream for inside network
 147
 148         :param in_if: Inside interface
 149         :param out_if: Outside interface
 150         :param ttl: TTL of generated packets
 151         """
 152         pkts = []
 153         # TCP
 154         p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
 155              IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
 156              TCP(sport=self.tcp_port_in, dport=self.tcp_external_port))
 157         pkts.append(p)
 158
 159         # UDP
 160         p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
 161              IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
 162              UDP(sport=self.udp_port_in, dport=self.udp_external_port))
 163         pkts.append(p)
 164
 165         # ICMP
 166         p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
 167              IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
 168              ICMP(id=self.icmp_id_in, type='echo-request'))
 169         pkts.append(p)
 170
 171         return pkts
 172
 173     def create_stream_out(self, out_if, dst_ip=None, ttl=64):
 174         """
 175         Create packet stream for outside network
 176
 177         :param out_if: Outside interface
 178         :param dst_ip: Destination IP address (Default use global NAT address)
 179         :param ttl: TTL of generated packets
 180         """
 181         if dst_ip is None:
 182             dst_ip = self.nat_addr
 183         pkts = []
 184         # TCP
 185         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
 186              IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
 187              TCP(dport=self.tcp_port_out, sport=self.tcp_external_port))
 188         pkts.append(p)
 189
 190         # UDP
 191         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
 192              IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
 193              UDP(dport=self.udp_port_out, sport=self.udp_external_port))
 194         pkts.append(p)
 195
 196         # ICMP
 197         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
 198              IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
 199              ICMP(id=self.icmp_external_id, type='echo-reply'))
 200         pkts.append(p)
 201
 202         return pkts
 203
 204     def verify_capture_out(self, capture, nat_ip=None):
 205         """
 206         Verify captured packets on outside network
 207
 208         :param capture: Captured packets
 209         :param nat_ip: Translated IP address (Default use global NAT address)
 210         :param same_port: Source port number is not translated (Default False)
 211         """
 212         if nat_ip is None:
 213             nat_ip = self.nat_addr
 214         for packet in capture:
 215             try:
 216                 self.assertEqual(packet[IP].src, nat_ip)
 217                 if packet.haslayer(TCP):
 218                     self.tcp_port_out = packet[TCP].sport
 219                 elif packet.haslayer(UDP):
 220                     self.udp_port_out = packet[UDP].sport
 221                 else:
 222                     self.icmp_external_id = packet[ICMP].id
 223             except:
 224                 self.logger.error(ppp("Unexpected or invalid packet "
 225                                       "(outside network):", packet))
 226                 raise
 227
 228     def test_deterministic_mode(self):
 229         """ NAT plugin run deterministic mode """
 230         in_addr = '172.16.255.0'
 231         out_addr = '172.17.255.50'
 232         in_addr_t = '172.16.255.20'
 233         in_plen = 24
 234         out_plen = 32
 235
 236         self.vapi.det44_add_del_map(is_add=1, in_addr=in_addr,
 237                                     in_plen=in_plen, out_addr=out_addr,
 238                                     out_plen=out_plen)
 239
 240         rep1 = self.vapi.det44_forward(in_addr_t)
 241         self.assertEqual(str(rep1.out_addr), out_addr)
 242         rep2 = self.vapi.det44_reverse(rep1.out_port_hi, out_addr)
 243
 244         self.assertEqual(str(rep2.in_addr), in_addr_t)
 245
 246         deterministic_mappings = self.vapi.det44_map_dump()
 247         self.assertEqual(len(deterministic_mappings), 1)
 248         dsm = deterministic_mappings[0]
 249         self.assertEqual(in_addr, str(dsm.in_addr))
 250         self.assertEqual(in_plen, dsm.in_plen)
 251         self.assertEqual(out_addr, str(dsm.out_addr))
 252         self.assertEqual(out_plen, dsm.out_plen)
 253
 254     def test_set_timeouts(self):
 255         """ Set deterministic NAT timeouts """
 256         timeouts_before = self.vapi.det44_get_timeouts()
 257
 258         self.vapi.det44_set_timeouts(
 259             udp=timeouts_before.udp + 10,
 260             tcp_established=timeouts_before.tcp_established + 10,
 261             tcp_transitory=timeouts_before.tcp_transitory + 10,
 262             icmp=timeouts_before.icmp + 10)
 263
 264         timeouts_after = self.vapi.det44_get_timeouts()
 265
 266         self.assertNotEqual(timeouts_before.udp, timeouts_after.udp)
 267         self.assertNotEqual(timeouts_before.icmp, timeouts_after.icmp)
 268         self.assertNotEqual(timeouts_before.tcp_established,
 269                             timeouts_after.tcp_established)
 270         self.assertNotEqual(timeouts_before.tcp_transitory,
 271                             timeouts_after.tcp_transitory)
 272
 273     def test_in(self):
 274         """ DET44 translation test (TCP, UDP, ICMP) """
 275
 276         nat_ip = "10.0.0.10"
 277
 278         self.vapi.det44_add_del_map(is_add=1, in_addr=self.pg0.remote_ip4,
 279                                     in_plen=32,
 280                                     out_addr=socket.inet_aton(nat_ip),
 281                                     out_plen=32)
 282
 283         self.vapi.det44_interface_add_del_feature(
 284             sw_if_index=self.pg0.sw_if_index,
 285             is_add=1, is_inside=1)
 286         self.vapi.det44_interface_add_del_feature(
 287             sw_if_index=self.pg1.sw_if_index,
 288             is_add=1, is_inside=0)
 289
 290         # in2out
 291         pkts = self.create_stream_in(self.pg0, self.pg1)
 292         self.pg0.add_stream(pkts)
 293         self.pg_enable_capture(self.pg_interfaces)
 294         self.pg_start()
 295         capture = self.pg1.get_capture(len(pkts))
 296         self.verify_capture_out(capture, nat_ip)
 297
 298         # out2in
 299         pkts = self.create_stream_out(self.pg1, nat_ip)
 300         self.pg1.add_stream(pkts)
 301         self.pg_enable_capture(self.pg_interfaces)
 302         self.pg_start()
 303         capture = self.pg0.get_capture(len(pkts))
 304         self.verify_capture_in(capture, self.pg0)
 305
 306         # session dump test
 307         sessions = self.vapi.det44_session_dump(self.pg0.remote_ip4)
 308         self.assertEqual(len(sessions), 3)
 309
 310         # TCP session
 311         s = sessions[0]
 312         self.assertEqual(str(s.ext_addr), self.pg1.remote_ip4)
 313         self.assertEqual(s.in_port, self.tcp_port_in)
 314         self.assertEqual(s.out_port, self.tcp_port_out)
 315         self.assertEqual(s.ext_port, self.tcp_external_port)
 316
 317         # UDP session
 318         s = sessions[1]
 319         self.assertEqual(str(s.ext_addr), self.pg1.remote_ip4)
 320         self.assertEqual(s.in_port, self.udp_port_in)
 321         self.assertEqual(s.out_port, self.udp_port_out)
 322         self.assertEqual(s.ext_port, self.udp_external_port)
 323
 324         # ICMP session
 325         s = sessions[2]
 326         self.assertEqual(str(s.ext_addr), self.pg1.remote_ip4)
 327         self.assertEqual(s.in_port, self.icmp_id_in)
 328         self.assertEqual(s.out_port, self.icmp_external_id)
 329
 330     def test_multiple_users(self):
 331         """ Deterministic NAT multiple users """
 332
 333         nat_ip = "10.0.0.10"
 334         port_in = 80
 335         external_port = 6303
 336
 337         host0 = self.pg0.remote_hosts[0]
 338         host1 = self.pg0.remote_hosts[1]
 339
 340         self.vapi.det44_add_del_map(is_add=1, in_addr=host0.ip4, in_plen=24,
 341                                     out_addr=socket.inet_aton(nat_ip),
 342                                     out_plen=32)
 343         self.vapi.det44_interface_add_del_feature(
 344             sw_if_index=self.pg0.sw_if_index,
 345             is_add=1, is_inside=1)
 346         self.vapi.det44_interface_add_del_feature(
 347             sw_if_index=self.pg1.sw_if_index,
 348             is_add=1, is_inside=0)
 349
 350         # host0 to out
 351         p = (Ether(src=host0.mac, dst=self.pg0.local_mac) /
 352              IP(src=host0.ip4, dst=self.pg1.remote_ip4) /
 353              TCP(sport=port_in, dport=external_port))
 354         self.pg0.add_stream(p)
 355         self.pg_enable_capture(self.pg_interfaces)
 356         self.pg_start()
 357         capture = self.pg1.get_capture(1)
 358         p = capture[0]
 359         try:
 360             ip = p[IP]
 361             tcp = p[TCP]
 362             self.assertEqual(ip.src, nat_ip)
 363             self.assertEqual(ip.dst, self.pg1.remote_ip4)
 364             self.assertEqual(tcp.dport, external_port)
 365             port_out0 = tcp.sport
 366         except:
 367             self.logger.error(ppp("Unexpected or invalid packet:", p))
 368             raise
 369
 370         # host1 to out
 371         p = (Ether(src=host1.mac, dst=self.pg0.local_mac) /
 372              IP(src=host1.ip4, dst=self.pg1.remote_ip4) /
 373              TCP(sport=port_in, dport=external_port))
 374         self.pg0.add_stream(p)
 375         self.pg_enable_capture(self.pg_interfaces)
 376         self.pg_start()
 377         capture = self.pg1.get_capture(1)
 378         p = capture[0]
 379         try:
 380             ip = p[IP]
 381             tcp = p[TCP]
 382             self.assertEqual(ip.src, nat_ip)
 383             self.assertEqual(ip.dst, self.pg1.remote_ip4)
 384             self.assertEqual(tcp.dport, external_port)
 385             port_out1 = tcp.sport
 386         except:
 387             self.logger.error(ppp("Unexpected or invalid packet:", p))
 388             raise
 389
 390         dms = self.vapi.det44_map_dump()
 391         self.assertEqual(1, len(dms))
 392         self.assertEqual(2, dms[0].ses_num)
 393
 394         # out to host0
 395         p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
 396              IP(src=self.pg1.remote_ip4, dst=nat_ip) /
 397              TCP(sport=external_port, dport=port_out0))
 398         self.pg1.add_stream(p)
 399         self.pg_enable_capture(self.pg_interfaces)
 400         self.pg_start()
 401         capture = self.pg0.get_capture(1)
 402         p = capture[0]
 403         try:
 404             ip = p[IP]
 405             tcp = p[TCP]
 406             self.assertEqual(ip.src, self.pg1.remote_ip4)
 407             self.assertEqual(ip.dst, host0.ip4)
 408             self.assertEqual(tcp.dport, port_in)
 409             self.assertEqual(tcp.sport, external_port)
 410         except:
 411             self.logger.error(ppp("Unexpected or invalid packet:", p))
 412             raise
 413
 414         # out to host1
 415         p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
 416              IP(src=self.pg1.remote_ip4, dst=nat_ip) /
 417              TCP(sport=external_port, dport=port_out1))
 418         self.pg1.add_stream(p)
 419         self.pg_enable_capture(self.pg_interfaces)
 420         self.pg_start()
 421         capture = self.pg0.get_capture(1)
 422         p = capture[0]
 423         try:
 424             ip = p[IP]
 425             tcp = p[TCP]
 426             self.assertEqual(ip.src, self.pg1.remote_ip4)
 427             self.assertEqual(ip.dst, host1.ip4)
 428             self.assertEqual(tcp.dport, port_in)
 429             self.assertEqual(tcp.sport, external_port)
 430         except:
 431             self.logger.error(ppp("Unexpected or invalid packet", p))
 432             raise
 433
 434         # session close api test
 435         self.vapi.det44_close_session_out(socket.inet_aton(nat_ip),
 436                                           port_out1,
 437                                           self.pg1.remote_ip4,
 438                                           external_port)
 439         dms = self.vapi.det44_map_dump()
 440         self.assertEqual(dms[0].ses_num, 1)
 441
 442         self.vapi.det44_close_session_in(host0.ip4,
 443                                          port_in,
 444                                          self.pg1.remote_ip4,
 445                                          external_port)
 446         dms = self.vapi.det44_map_dump()
 447         self.assertEqual(dms[0].ses_num, 0)
 448
 449     def test_tcp_session_close_detection_in(self):
 450         """ DET44 TCP session close from inside network """
 451         self.vapi.det44_add_del_map(is_add=1, in_addr=self.pg0.remote_ip4,
 452                                     in_plen=32,
 453                                     out_addr=socket.inet_aton(self.nat_addr),
 454                                     out_plen=32)
 455         self.vapi.det44_interface_add_del_feature(
 456             sw_if_index=self.pg0.sw_if_index,
 457             is_add=1, is_inside=1)
 458         self.vapi.det44_interface_add_del_feature(
 459             sw_if_index=self.pg1.sw_if_index,
 460             is_add=1, is_inside=0)
 461
 462         self.initiate_tcp_session(self.pg0, self.pg1)
 463
 464         # close the session from inside
 465         try:
 466             # FIN packet in -> out
 467             p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
 468                  IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
 469                  TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
 470                      flags="F"))
 471             self.pg0.add_stream(p)
 472             self.pg_enable_capture(self.pg_interfaces)
 473             self.pg_start()
 474             self.pg1.get_capture(1)
 475
 476             pkts = []
 477
 478             # ACK packet out -> in
 479             p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
 480                  IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
 481                  TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
 482                      flags="A"))
 483             pkts.append(p)
 484
 485             # FIN packet out -> in
 486             p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
 487                  IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
 488                  TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
 489                      flags="F"))
 490             pkts.append(p)
 491
 492             self.pg1.add_stream(pkts)
 493             self.pg_enable_capture(self.pg_interfaces)
 494             self.pg_start()
 495             self.pg0.get_capture(2)
 496
 497             # ACK packet in -> out
 498             p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
 499                  IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
 500                  TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
 501                      flags="A"))
 502             self.pg0.add_stream(p)
 503             self.pg_enable_capture(self.pg_interfaces)
 504             self.pg_start()
 505             self.pg1.get_capture(1)
 506
 507             # Check if deterministic NAT44 closed the session
 508             dms = self.vapi.det44_map_dump()
 509             self.assertEqual(0, dms[0].ses_num)
 510         except:
 511             self.logger.error("TCP session termination failed")
 512             raise
 513
 514     def test_tcp_session_close_detection_out(self):
 515         """ Deterministic NAT TCP session close from outside network """
 516         self.vapi.det44_add_del_map(is_add=1, in_addr=self.pg0.remote_ip4,
 517                                     in_plen=32,
 518                                     out_addr=socket.inet_aton(self.nat_addr),
 519                                     out_plen=32)
 520         self.vapi.det44_interface_add_del_feature(
 521             sw_if_index=self.pg0.sw_if_index,
 522             is_add=1, is_inside=1)
 523         self.vapi.det44_interface_add_del_feature(
 524             sw_if_index=self.pg1.sw_if_index,
 525             is_add=1, is_inside=0)
 526
 527         self.initiate_tcp_session(self.pg0, self.pg1)
 528
 529         # close the session from outside
 530         try:
 531             # FIN packet out -> in
 532             p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
 533                  IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
 534                  TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
 535                      flags="F"))
 536             self.pg1.add_stream(p)
 537             self.pg_enable_capture(self.pg_interfaces)
 538             self.pg_start()
 539             self.pg0.get_capture(1)
 540
 541             pkts = []
 542
 543             # ACK packet in -> out
 544             p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
 545                  IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
 546                  TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
 547                      flags="A"))
 548             pkts.append(p)
 549
 550             # ACK packet in -> out
 551             p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
 552                  IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
 553                  TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
 554                      flags="F"))
 555             pkts.append(p)
 556
 557             self.pg0.add_stream(pkts)
 558             self.pg_enable_capture(self.pg_interfaces)
 559             self.pg_start()
 560             self.pg1.get_capture(2)
 561
 562             # ACK packet out -> in
 563             p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
 564                  IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
 565                  TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
 566                      flags="A"))
 567             self.pg1.add_stream(p)
 568             self.pg_enable_capture(self.pg_interfaces)
 569             self.pg_start()
 570             self.pg0.get_capture(1)
 571
 572             # Check if deterministic NAT44 closed the session
 573             dms = self.vapi.det44_map_dump()
 574             self.assertEqual(0, dms[0].ses_num)
 575         except:
 576             self.logger.error("TCP session termination failed")
 577             raise
 578
 579     def test_session_timeout(self):
 580         """ Deterministic NAT session timeouts """
 581         self.vapi.det44_add_del_map(is_add=1, in_addr=self.pg0.remote_ip4,
 582                                     in_plen=32,
 583                                     out_addr=socket.inet_aton(self.nat_addr),
 584                                     out_plen=32)
 585         self.vapi.det44_interface_add_del_feature(
 586             sw_if_index=self.pg0.sw_if_index,
 587             is_add=1, is_inside=1)
 588         self.vapi.det44_interface_add_del_feature(
 589             sw_if_index=self.pg1.sw_if_index,
 590             is_add=1, is_inside=0)
 591
 592         self.initiate_tcp_session(self.pg0, self.pg1)
 593         self.vapi.det44_set_timeouts(udp=5, tcp_established=5,
 594                                      tcp_transitory=5, icmp=5)
 595         pkts = self.create_stream_in(self.pg0, self.pg1)
 596         self.pg0.add_stream(pkts)
 597         self.pg_enable_capture(self.pg_interfaces)
 598         self.pg_start()
 599         self.pg1.get_capture(len(pkts))
 600         self.virtual_sleep(15)
 601
 602         dms = self.vapi.det44_map_dump()
 603         self.assertEqual(0, dms[0].ses_num)
 604
 605     # TODO: ipfix needs to be separated from NAT base plugin
 606     @unittest.skipUnless(running_extended_tests, "part of extended tests")
 607     def test_session_limit_per_user(self):
 608         """ Deterministic NAT maximum sessions per user limit """
 609         self.vapi.det44_add_del_map(is_add=1, in_addr=self.pg0.remote_ip4,
 610                                     in_plen=32,
 611                                     out_addr=socket.inet_aton(self.nat_addr),
 612                                     out_plen=32)
 613         self.vapi.det44_interface_add_del_feature(
 614             sw_if_index=self.pg0.sw_if_index,
 615             is_add=1, is_inside=1)
 616         self.vapi.det44_interface_add_del_feature(
 617             sw_if_index=self.pg1.sw_if_index,
 618             is_add=1, is_inside=0)
 619         self.vapi.set_ipfix_exporter(collector_address=self.pg2.remote_ip4,
 620                                      src_address=self.pg2.local_ip4,
 621                                      path_mtu=512,
 622                                      template_interval=10)
 623         self.vapi.nat_ipfix_enable_disable(domain_id=1, src_port=4739,
 624                                            enable=1)
 625
 626         pkts = []
 627         for port in range(1025, 2025):
 628             p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
 629                  IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
 630                  UDP(sport=port, dport=port))
 631             pkts.append(p)
 632
 633         self.pg0.add_stream(pkts)
 634         self.pg_enable_capture(self.pg_interfaces)
 635         self.pg_start()
 636         self.pg1.get_capture(len(pkts))
 637
 638         p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
 639              IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
 640              UDP(sport=3001, dport=3002))
 641         self.pg0.add_stream(p)
 642         self.pg_enable_capture(self.pg_interfaces)
 643         self.pg_start()
 644         self.pg1.assert_nothing_captured()
 645
 646         # verify ICMP error packet
 647         capture = self.pg0.get_capture(1)
 648         p = capture[0]
 649         self.assertTrue(p.haslayer(ICMP))
 650         icmp = p[ICMP]
 651         self.assertEqual(icmp.type, 3)
 652         self.assertEqual(icmp.code, 1)
 653         self.assertTrue(icmp.haslayer(IPerror))
 654         inner_ip = icmp[IPerror]
 655         self.assertEqual(inner_ip[UDPerror].sport, 3001)
 656         self.assertEqual(inner_ip[UDPerror].dport, 3002)
 657
 658         dms = self.vapi.det44_map_dump()
 659
 660         self.assertEqual(1000, dms[0].ses_num)
 661
 662         # verify IPFIX logging
 663         self.vapi.ipfix_flush()
 664         capture = self.pg2.get_capture(2)
 665         ipfix = IPFIXDecoder()
 666         # first load template
 667         for p in capture:
 668             self.assertTrue(p.haslayer(IPFIX))
 669             if p.haslayer(Template):
 670                 ipfix.add_template(p.getlayer(Template))
 671         # verify events in data set
 672         for p in capture:
 673             if p.haslayer(Data):
 674                 data = ipfix.decode_data_set(p.getlayer(Set))
 675                 self.verify_ipfix_max_entries_per_user(data,
 676                                                        1000,
 677                                                        self.pg0.remote_ip4)
 678         self.vapi.nat_ipfix_enable_disable(domain_id=1, src_port=4739,
 679                                            enable=0)