SNAT: IPFIX logging (VPP-445)

Change-Id: I8450217dd43a1cd9f510e40dfb22274ffc33a4c6
Signed-off-by: Matus Fabian <matfabia@cisco.com>

1 files changed, 127 insertions, 1 deletions

diff --git a/test/test_snat.py b/test/test_snat.py
index 653496e2880..1abb3daaa40 100644
--- a/test/test_snat.py
+++ b/test/test_snat.py
@@ -2,12 +2,14 @@
 
 import socket
 import unittest
+import struct
 
 from framework import VppTestCase, VppTestRunner
-
 from scapy.layers.inet import IP, TCP, UDP, ICMP
 from scapy.layers.l2 import Ether
+from scapy.data import IP_PROTOS
 from util import ppp
+from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
 
 
 class TestSNAT(VppTestCase):
@@ -176,6 +178,62 @@ class TestSNAT(VppTestCase):
                                       "(inside network):", packet))
                 raise
 
+    def verify_ipfix_nat44_ses(self, data):
+        """
+        Verify IPFIX NAT44 session create/delete event
+
+        :param data: Decoded IPFIX data records
+        """
+        nat44_ses_create_num = 0
+        nat44_ses_delete_num = 0
+        self.assertEqual(6, len(data))
+        for record in data:
+            # natEvent
+            self.assertIn(ord(record[230]), [4, 5])
+            if ord(record[230]) == 4:
+                nat44_ses_create_num += 1
+            else:
+                nat44_ses_delete_num += 1
+            # sourceIPv4Address
+            self.assertEqual(self.pg0.remote_ip4n, record[8])
+            # postNATSourceIPv4Address
+            self.assertEqual(socket.inet_pton(socket.AF_INET, self.snat_addr),
+                             record[225])
+            # ingressVRFID
+            self.assertEqual(struct.pack("!I", 0), record[234])
+            # protocolIdentifier/sourceTransportPort/postNAPTSourceTransportPort
+            if IP_PROTOS.icmp == ord(record[4]):
+                self.assertEqual(struct.pack("!H", self.icmp_id_in), record[7])
+                self.assertEqual(struct.pack("!H", self.icmp_id_out),
+                                 record[227])
+            elif IP_PROTOS.tcp == ord(record[4]):
+                self.assertEqual(struct.pack("!H", self.tcp_port_in),
+                                 record[7])
+                self.assertEqual(struct.pack("!H", self.tcp_port_out),
+                                 record[227])
+            elif IP_PROTOS.udp == ord(record[4]):
+                self.assertEqual(struct.pack("!H", self.udp_port_in),
+                                 record[7])
+                self.assertEqual(struct.pack("!H", self.udp_port_out),
+                                 record[227])
+            else:
+                self.fail("Invalid protocol")
+        self.assertEqual(3, nat44_ses_create_num)
+        self.assertEqual(3, nat44_ses_delete_num)
+
+    def verify_ipfix_addr_exhausted(self, data):
+        """
+        Verify IPFIX NAT addresses event
+
+        :param data: Decoded IPFIX data records
+        """
+        self.assertEqual(1, len(data))
+        record = data[0]
+        # natEvent
+        self.assertEqual(ord(record[230]), 3)
+        # natPoolID
+        self.assertEqual(struct.pack("!I", 0), record[283])
+
     def clear_snat(self):
         """
         Clear SNAT configuration.
@@ -184,6 +242,8 @@ class TestSNAT(VppTestCase):
         for intf in interfaces:
             self.vapi.snat_add_interface_addr(intf.sw_if_index, is_add=0)
 
+        self.vapi.snat_ipfix(enable=0)
+
         interfaces = self.vapi.snat_interface_dump()
         for intf in interfaces:
             self.vapi.snat_interface_add_del_feature(intf.sw_if_index,
@@ -647,11 +707,77 @@ class TestSNAT(VppTestCase):
         adresses = self.vapi.snat_address_dump()
         self.assertEqual(0, len(adresses))
 
+    def test_ipfix_nat44_sess(self):
+        """ S-NAT IPFIX logging NAT44 session created/delted """
+        self.snat_add_address(self.snat_addr)
+        self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
+        self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
+                                                 is_inside=0)
+        self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
+                                     src_address=self.pg3.local_ip4n,
+                                     path_mtu=512,
+                                     template_interval=10)
+        self.vapi.snat_ipfix()
+
+        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)
+        self.snat_add_address(self.snat_addr, is_add=0)
+        self.vapi.cli("ipfix flush")  # FIXME this should be an API call
+        capture = self.pg3.get_capture(3)
+        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_nat44_ses(data)
+
+    def test_ipfix_addr_exhausted(self):
+        """ S-NAT IPFIX logging NAT addresses exhausted """
+        self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
+        self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
+                                                 is_inside=0)
+        self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
+                                     src_address=self.pg3.local_ip4n,
+                                     path_mtu=512,
+                                     template_interval=10)
+        self.vapi.snat_ipfix()
+
+        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=3025))
+        self.pg0.add_stream(p)
+        self.pg_enable_capture(self.pg_interfaces)
+        self.pg_start()
+        capture = self.pg1.get_capture(0)
+        self.vapi.cli("ipfix flush")  # FIXME this should be an API call
+        capture = self.pg3.get_capture(3)
+        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_addr_exhausted(data)
+
     def tearDown(self):
         super(TestSNAT, self).tearDown()
         if not self.vpp_dead:
             self.logger.info(self.vapi.cli("show snat verbose"))
             self.clear_snat()
 
+
 if __name__ == '__main__':
     unittest.main(testRunner=VppTestRunner)