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