summaryrefslogtreecommitdiffstats
path: root/src
AgeCommit message (Collapse)AuthorFilesLines
2017-02-06SNAT: fix snat_add_static_mapping_command() uninitialized variableMatus Fabian2-5/+16
Change-Id: I7775dd3b90d5a3449650c3102e24bfedd770beb1 Signed-off-by: Matus Fabian <matfabia@cisco.com>
2017-02-05SNAT: initialize outside and inside FIB index in snat_configMatus Fabian2-36/+5
Change-Id: If26d758997d71792cedad1afae8d6a38cfd364ac Signed-off-by: Matus Fabian <matfabia@cisco.com>
2017-02-05fix some 'stored but never read' warnings raised by clangGabriel Ganne2-6/+7
found by `scan-build make plugins-release` Signed-off-by: Gabriel Ganne <gabriel.ganne@enea.com> Change-Id: I52048e3a8ae3fb85eb3d91f6a5e15216dd7b9baa
2017-02-05Wheel-timer infraDave Barach8-0/+1289
Change-Id: I5499dd6b768425a56936afae50bd578620c83d30 Signed-off-by: Dave Barach <dave@barachs.net>
2017-02-04dpdk: move to uio_pci_genericDamjan Marion3-22/+4
Change-Id: I3d8b7947ae6d721e9b514a59a7d2de49aed419b5 Signed-off-by: Damjan Marion <damarion@cisco.com>
2017-02-03pg: add trace information for pg tx nodeDamjan Marion3-9/+46
Change-Id: I31730d58c34331f25f5b02cd065be94251f1302c Signed-off-by: Damjan Marion <damarion@cisco.com>
2017-02-03vhost-user: fix missing speculative enqueue unwindSteven1-0/+4
Running trex in a VM with a bad config, trex sent a bogus pack from the VM to the Virtual interface. It caused a crash. Change-Id: I64d0197b444265553ab4c24f21e6a962e89cb587 Signed-off-by: Steven <sluong@cisco.com>
2017-02-03Fix CLI help message for set ip addressMatej Klotton1-1/+1
Change-Id: Ie39bb26a9aea88cf2768ec537adcdd8df1de3be0 Signed-off-by: Matej Klotton <mklotton@cisco.com>
2017-02-03Next node frame over-flow after replicationNeale Ranns1-2/+16
Change-Id: I25077dd0739787de4f7512e5a70a62e8c34c28e4 Signed-off-by: Neale Ranns <nranns@cisco.com>
2017-02-03Plugin infrastructure improvementsDamjan Marion16-210/+367
This patch replaces requirement for vlib_plugin_register function in the plugin so file and introduces new macro: VLIB_PLUGIN_REGISTER () = { .version = "version string", .version_required = "requred version", .default_disabled = 1, .early_init = "early_init_function_name", }; Plugin will nor be loaded if .default_disabled is set to 1 unless explicitely enabled in startup.conf. If .verstion_required is set, plugin will not be loaded if there is version mismatch between plugin and vpp. This can be bypassed by setting "skip-version-check" for specific plugin. If .early-init string is present, plugin loader will try to resolve this specific symbol in the plugin namespace and make a function call. Following startup.conf configuration is added: plugins { path /path/to/plugin/directory plugin ila_plugin.so { enable skip-version-check } plugin acl_plugin.so { disable } } Change-Id: I706c691dd34d94ffe9e02b59831af8859a95f061 Signed-off-by: Damjan Marion <damarion@cisco.com>
2017-02-03SNAT: fix invalid outside FIB indexMatus Fabian1-0/+14
Change-Id: Ia5d3d81cbc2ef85fabf9e19c89a52c589a921d14 Signed-off-by: Matus Fabian <matfabia@cisco.com>
2017-02-03SNAT: Port allocation per protocolMatus Fabian7-131/+219
Ports are allocated per protocol (UDP, TCP, ICMP) 1:1 NAT with port is configured for specific protocol Change-Id: I37ae5eed3715b223d0620d4fdaed7a482bb7a834 Signed-off-by: Matus Fabian <matfabia@cisco.com>
2017-02-02Remove unnecessary block structure around CONTROL_PING messages.Jon Loeliger2-204/+168
Now that the M() and S() macros accept a message parameter, there is no longer a need to introduce a new block structure around the CONTROL_PING messages just to have a new unbound "mp" variable. Instead, just use one named "mp_ping" directly. Change-Id: I6b283562bb6eec25806e3d35c35b977680ecd1dd Signed-off-by: Jon Loeliger <jdl@netgate.com>
2017-02-02Refactor fragile msg macro W and W2 to not burry return control flow.Jon Loeliger12-576/+696
Instead, have them accept and assign a return paramter leaving the return control flow up to the caller. Clean up otherwise misleading returns present even after "NOT REACHED" comments. Change-Id: I0861921f73ab65d55b95eabd27514f0129152723 Signed-off-by: Jon Loeliger <jdl@netgate.com>
2017-02-02Localize the timeout variable within the W message macro.Jon Loeliger11-229/+5
Rather than rely on an unbound variable, explicitly introduce the timeout variable within the 'do { ... } while (0)' construct as a block-local variable. Change-Id: I6e78635290f9b5ab3f56b7f116c5fa762c88c9e9 Signed-off-by: Jon Loeliger <jdl@netgate.com>
2017-02-02Convert message macro S to accept a message pointer parameter;Jon Loeliger12-282/+293
Rather than blindly assume an unbound, fixed message parameter explicilty pass it as a paramter to the S() macro. Change-Id: Ieea1f1815cadd2eec7d9240408d69acdc3caa49a Signed-off-by: Jon Loeliger <jdl@netgate.com>
2017-02-02Convert M() and M2() macros to honor their second, mp, parameter.Jon Loeliger1-2/+2
Now that all the M() and M2() uses properly supply a message pointer as second parameter, fix the macros to use it. Change-Id: I0b8f4848416c3fa2e06755ad6ea7171b7c546124 Signed-off-by: Jon Loeliger <jdl@netgate.com>
2017-02-02Ensure all M() and M2() second parameters are the message pointer.Jon Loeliger11-283/+284
Rather than maintain (?) an unused second parameter, t, and pull an unbound message pointer, mp, out of context, explicitly list the message pointer as the second parameter. Change-Id: I92143efda6211cdf6b935470f8c71579742a6b64 Signed-off-by: Jon Loeliger <jdl@netgate.com>
2017-02-02Fix SR multicast post mfib commitNeale Ranns13-650/+508
1 - use the SR policy to construct the replicate DPO. Each bucket therein is a SR tunnel. 2 - install a special mfib entry that links via this replicate 3 - forwarding is now mfib-lookup -> replicate -> sr_rewrite (per-tunnel) no need for a separate sr_replicate node. 4 - Stack the sr tunnel on the forwarding DPO of the first-hop FIB entry. no need for a second lookup post SR encap. 5 - fix some path-list lock leaks in the MFIB entry. Change-Id: I20de96ea4c4be4fae252625bde159d9c435c8315 Signed-off-by: Neale Ranns <nranns@cisco.com>
2017-02-02Update default Vagrant box to Ubuntu 16.04, VPP-616Dave Wallace1-8/+5
- Make puppetlabs/ubuntu-16.04-64-nocm the default box - Enable x11 forwarding - Install x11-utils required for emacs to work over X11 - Refactor run.sh - Add VPP_VAGRANT_POST_BUILD environment variable to allow selection of installing VPP or run "make test". - Fix dependencies in src/vppapigen.am Change-Id: I0ec054fdc83feb71ca8396df53ed02bf82ecd7e7 Signed-off-by: Dave Wallace <dwallacelf@gmail.com>
2017-02-02BFD: SHA1 authenticationKlement Sekera9-258/+1459
Add authentication support to BFD feature. Out of three existing authentication types, implement SHA1 (sole RFC requirement). Simple password is insecure and MD5 is discouraged by the RFC, so ignore those. Add/change APIs to allow configuring BFD authentication keys and their usage with BFD sessions. Change-Id: Ifb0fb5b19c2e72196d84c1cde919bd4c074ea415 Signed-off-by: Klement Sekera <ksekera@cisco.com>
2017-02-02BFD: improve finding of ipv4/ipv6 headersKlement Sekera1-48/+62
Avoid coverity warning and improve safety by declaring a helper structure and working with it when searching for ip headers. Make sure the content following IPv6 header is actually UDP before parsing it. Bail out if unexpcted IPv6 header found ... Change-Id: I1c6b9fd42d6fdae226f12c91c53c07a932b29522 Signed-off-by: Klement Sekera <ksekera@cisco.com>
2017-02-02dpdk: fix crypto coverity warningSergio Gonzalez Monroy1-1/+2
Change-Id: I165b64fdc12dd2936df1958348e93b709ce0e784 Signed-off-by: Sergio Gonzalez Monroy <sergio.gonzalez.monroy@intel.com>
2017-02-02SNAT: changed source for outbound address FIB entry (VPP-613)Matus Fabian2-26/+13
Use FIB_SOURCE_PLUGIN_HI and modify ARP input to use non-source variants for flags and resolving interface get. Change-Id: I3bab76f36e0b1ee86e430a416099f1654e02740a Signed-off-by: Matus Fabian <matfabia@cisco.com>
2017-02-02LISP: enhance binary part of some APIsFilip Tehlar5-117/+41
Remote mapping and locator set binary APIs uses zero length arrays defined as 'u8 array[0]' in .api file. This path will change such cases to form 'type_t array[count];' in order to enhance maintainability. Change-Id: I98d0252b441020609c550d48186ed0d8338a3f2d Signed-off-by: Filip Tehlar <ftehlar@cisco.com>
2017-02-02VXLAN: further unify ip4/ip6 ctl plane handlingEyal Bari2-83/+55
fix wrong udp error codes in decap ip6 handling Change-Id: Ibf791a995128d38b31725c1ee67ec9d1c9dffca2 Signed-off-by: Eyal Bari <ebari@cisco.com>
2017-01-31BFD: reformat code to match vpp code styleKlement Sekera1-158/+186
Change-Id: I40deb8b40f5d3a96d2c0dcb400f489cd05a64348 Signed-off-by: Klement Sekera <ksekera@cisco.com>
2017-01-31Prep work for Coverity upload processing via JenkinsDave Barach1-0/+10
Change-Id: I2575d780d19e12ddf8a77e5596e5d7cc3dbf4233 Signed-off-by: Dave Barach <dave@barachs.net>
2017-01-31SNAT: Add outbound addresses to FIB (VPP-613)Matus Fabian1-1/+110
Add the external NAT address to the FIB as receive entries. This ensures that VPP will reply to ARP for these addresses and we don't need to enable proxy ARP on the outside interface. Change-Id: I1db153373c43fec4808845449a17085509ca588c Signed-off-by: Matus Fabian <matfabia@cisco.com>
2017-01-31jvpp: fix coverity warn: reliance on default encodingMarek Gradzki5-18/+35
Change-Id: I8333b7d19ebdacac5445b4505750dd0a46764b36 Signed-off-by: Marek Gradzki <mgradzki@cisco.com>
2017-01-31VXLAN: small changes for brevityEyal Bari1-49/+16
Change-Id: I9cfff0196845c6f1161848aa49442aa4df81b20e Signed-off-by: Eyal Bari <ebari@cisco.com>
2017-01-31MFIB Coverity warnings. The lock macro is functionally equivalent but more ↵Neale Ranns3-29/+44
expressive (and might appease coverity) Change-Id: Ie3c9b2896a487a0302903bfbdd6348f6f091c67d Signed-off-by: Neale Ranns <nranns@cisco.com>
2017-01-31Add vxlan-bypass feature to IP6 forwarding pathJohn Lo7-184/+470
Add vxlan-bypass feature which can be enabled on the IP6 underlay interface which receive VXLAN packets to accelerate VXLAN decap processing. The CLI to enable/disable it is: set interface ip6 vxlan-bypass <interface> [del] The vxlan-bypass feature is already supported on the IP4 underlay interface. The CLI to enable/disable it is: set interface ip vxlan-bypass <interface> [del] Move vxlan-bypass API/CLI support code from decap.c to vxlan.c. Also fixed two issues in the VXLAN decap path in the vxlan-input node: 1. Add verification of VXLAN packet FIB index with the encap-vrf-id of the VXLAN tunnel. 2. Fix checking of VXLANoIPv6 packet mcast DIP against that of the IP6 mcast VXLAN tunnel. Change-Id: I2bad4074a468c48fbb8bb5ac64f6437190756ed2 Signed-off-by: John Lo <loj@cisco.com>
2017-01-30VPP-621: ping: ICMP echo data size must be bounded by VLIB_BUFFER_DATA_SIZE ↵Andrew Yourtchenko2-29/+46
minus headers. Before the commit 878c6098 the VLIB_BUFFER_DATA_SIZE was different depending on whether building "vpp" or "vpp_lite", resulting in an overrun in vpp_lite build. Avoid the hardcoded value and make the upper bound for ICMP echo data size dependent on the buffer size. Change-Id: Id6c4d7fc73766a95af2610eb237881b5fe9ce9aa Signed-off-by: Andrew Yourtchenko <ayourtch@gmail.com>
2017-01-30Fixed set int state up for ipsec interfacesRadu Nicolau1-2/+2
Change-Id: I2330cb7c2ba0f5eaeb4e7a4c3de4f22283d3923d Signed-off-by: Radu Nicolau <radu.nicolau@intel.com>
2017-01-30Fix LISP Coverity warningsFlorin Coras2-6/+16
Change-Id: Iaca2ff453872e638ee83b11fc16472e44deb9a7e Signed-off-by: Florin Coras <fcoras@cisco.com>
2017-01-30add vnet/util/radix.h to nobase_include_HEADERS listGabriel Ganne1-1/+2
header is included by ip6.h and should therefore be made public Change-Id: Ieb7d14bb082d821c2f77e9204ec09d884fb2cb09 Signed-off-by: Gabriel Ganne <gabriel.ganne@enea.com>
2017-01-30LISP: add dump calls for GPE entries APIFilip Tehlar6-2/+505
Change-Id: Ie7f51643fd3522a0fa8df8d0309305481c211f5f Signed-off-by: Filip Tehlar <ftehlar@cisco.com>
2017-01-28Add files to CLEANFILES for robust make clean.Burt Silverman6-1/+11
At the least, $(BUILT_SOURCES) should be added to CLEANFILES. Also beneficial is $(api_DATA), and in the case of Java, *.files and *.h. Also there is a vpp/app/version.h, and some grammar and lex files in vppapigen. Change-Id: Ic6d3f2d40ce65e1d9a8b88217fa1f36de393ebb4 Signed-off-by: Burt Silverman <burtms@gmail.com>
2017-01-28sh not show in the mfib flags commandsNeale Ranns1-2/+2
Change-Id: If10b878cb05d695eaed9416202fd19bed2e0b793 Signed-off-by: Neale Ranns <nranns@cisco.com>
2017-01-27API refactoring : policerPavel Kotucek6-267/+387
Change-Id: Ia7d8b557bcdf45eb8e33bb3d297bc6f7ad321c72 Signed-off-by: Pavel Kotucek <pkotucek@cisco.com>
2017-01-27dpdk: rework cryptodev ipsec build and setupSergio Gonzalez Monroy25-412/+441
Build Cryptodev IPsec support by default when DPDK is enabled but only build hardware Cryptodev PMDs. To enable Cryptodev support, a new startup.conf option for dpdk has been introduced 'enable-cryptodev'. During VPP init, if Cryptodev support is not enabled or not enough cryptodev resources are available then default to OpenSSL ipsec implementation. Change-Id: I5aa7e0d5c2676bdb41d775ef40364536a081956d Signed-off-by: Sergio Gonzalez Monroy <sergio.gonzalez.monroy@intel.com>
2017-01-27API refactoring : copPavel Kotucek6-114/+237
Change-Id: I1596d7a130fb6d3b170b8e01f2116b323cc10c0d Signed-off-by: Pavel Kotucek <pkotucek@cisco.com>
2017-01-27dpdk : incorrect rx filter being installedPavel Kotucek3-1/+20
When mac address is set prior bringing interface up incorrect rx filter being installed into the e1000 mac. Change-Id: If59a2bf16f732e45221b3787d271307d369e54d3 Signed-off-by: Pavel Kotucek <pkotucek@cisco.com>
2017-01-27jvpp: utilize per-message CRCs (VPP-544)Marek Gradzki11-15/+143
Since messages ids are no longer statically referenced, fixes also VPP-611. Change-Id: Ic8e6ee2b7f1142c185595347984d69350be25ac3 Signed-off-by: Marek Gradzki <mgradzki@cisco.com>
2017-01-27IP Multicast FIB (mfib)Neale Ranns83-1670/+10388
- IPv[46] mfib tables with support for (*,G/m), (*,G) and (S,G) exact and longest prefix match - Replication represented via a new replicate DPO. - RPF configuration and data-plane checking - data-plane signals sent to listening control planes. The functions of multicast forwarding entries differ from their unicast conterparts, so we introduce a new mfib_table_t and mfib_entry_t objects. However, we re-use the fib_path_list to resolve and build the entry's output list. the fib_path_list provides the service to construct a replicate DPO for multicast. 'make tests' is added to with two new suites; TEST=mfib, this is invocation of the CLI command 'test mfib' which deals with many path add/remove, flag set/unset scenarios, TEST=ip-mcast, data-plane forwarding tests. Updated applications to use the new MIFB functions; - IPv6 NS/RA. - DHCPv6 unit tests for these are undated accordingly. Change-Id: I49ec37b01f1b170335a5697541c8fd30e6d3a961 Signed-off-by: Neale Ranns <nranns@cisco.com>
2017-01-27Cleanup some obfuscated code in next node handling.Vengada3-8/+8
The values of next node can be simply assigned by dereferencing the pointer instead of obfuscating the dereference. Change-Id: I1f4a3d51b768960699010591410695473728d3a2 Signed-off-by: Vengada <venggovi@cisco.com>
2017-01-27drop-and-count snat out2in packets with no translationsDave Barach1-3/+41
Send dhcp server-to-client packets to the ip stack, so we can acquire snat outside interface addresses from a dhcp server Change-Id: I7751356fa23d9f26b503c9796bd85f96275fe978 Signed-off-by: Dave Barach <dave@barachs.net>
2017-01-27Python API: event_callback not initialised.Ole Troan1-0/+1
Even when not requesting notification the VPP API sends event notifications on certain calls. E.g. creating interfaces. Traceback (most recent call last): File "/vpp/ipfix/src/vpp-api/python/vpp_papi/vpp_papi.py", line 447, in msg_handler_sync if self.event_callback: AttributeError: VPP instance has no attribute 'event_callback' Change-Id: I4ca30c49df298655dc8948c5ebd68de0b3d6a592 Signed-off-by: Ole Troan <ot@cisco.com>
2017-01-27Add multi-vpp support back into pythonic vppctlEd Warnicke1-7/+20
Back when vppctl was a shell script, you could use -p or --prefix to set the chroot prefix to drive multiple vpp instances. This patch adds that capability back. Change-Id: Iaa70a20eff13e8d7e206fcceadb7e5d06afa3fc5 Signed-off-by: Ed Warnicke <hagbard@gmail.com>
1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519 1520 1521 1522 1523 1524 1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537 1538 1539 1540 1541 1542 1543 1544 1545 1546 1547 1548 1549 1550 1551 1552 1553 1554 1555 1556 1557 1558 1559 1560 1561 1562 1563 1564 1565 1566 1567 1568 1569 1570 1571 1572 1573 1574 1575 1576 1577 1578 1579 1580 1581 1582 1583 1584 1585 1586 1587 1588 1589 1590 1591 1592 1593 1594 1595 1596 1597 1598 1599 1600 1601 1602 1603 1604 1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636 1637 1638 1639 1640 1641 1642 1643 1644 1645 1646 1647 1648 1649 1650 1651 1652 1653 1654 1655 1656 1657 1658 1659 1660 1661 1662 1663 1664 1665 1666 1667 1668 1669 1670 1671 1672 1673 1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684 1685 1686 1687 1688 1689 1690 1691 1692 1693 1694 1695 1696 1697 1698 1699 1700 1701 1702 1703 1704 1705 1706 1707 1708 1709 1710 1711 1712 1713 1714 1715 1716 1717 1718 1719 1720 1721 1722 1723 1724 1725 1726 1727 1728 1729 1730 1731 1732 1733 1734 1735 1736 1737 1738 1739 1740 1741 1742 1743 1744 1745 1746 1747 1748 1749 1750 1751 1752 1753 1754 1755 1756 1757 1758 1759 1760 1761 1762 1763 1764 1765 1766 1767 1768 1769 1770 1771 1772 1773 1774 1775 1776 1777 1778 1779 1780 1781 1782 1783 1784 1785 1786 1787 1788 1789 1790 1791 1792 1793 1794 1795 1796 1797 1798 1799 1800 1801 1802 1803 1804 1805 1806 1807 1808 1809 1810 1811 1812 1813 1814 1815 1816 1817 1818 1819 1820 1821 1822 1823 1824 1825 1826 1827 1828 1829 1830 1831 1832 1833 1834 1835 1836 1837 1838 1839 1840 1841 1842 1843 1844 1845 1846 1847 1848 1849 1850 1851 1852 1853 1854 1855 1856 1857 1858 1859 1860 1861 1862 1863 1864 1865 1866 1867 1868 1869 1870 1871 1872 1873 1874 1875 1876 1877 1878 1879 1880 1881 1882 1883 1884 1885 1886 1887 1888 1889 1890 1891 1892 1893 1894 1895 1896 1897 1898 1899 1900 1901 1902 1903 1904 1905 1906 1907 1908 1909 1910 1911 1912 1913 1914 1915 1916 1917 1918 1919 1920 1921 1922 1923 1924 1925 1926 1927 1928 1929 1930 1931 1932 1933 1934 1935 1936 1937 1938 1939 1940 1941 1942 1943 1944 1945 1946 1947 1948 1949 1950 1951 1952 1953 1954 1955 1956 1957 1958 1959 1960 1961 1962 1963 1964 1965 1966 1967 1968 1969 1970 1971 1972 1973 1974 1975 1976 1977 1978 1979 1980 1981 1982 1983 1984 1985 1986 1987 1988 1989 1990 1991 1992 1993 1994 1995 1996 1997 1998 1999 2000 2001 2002 2003 2004 2005 2006 2007 2008 2009 2010 2011 2012 2013 2014 2015 2016 2017 2018 2019 2020 2021 2022 2023 2024 2025 2026 2027 2028 2029 2030 2031 2032 2033 2034 2035 2036 2037 2038 2039 2040 2041 2042 2043 2044 2045 2046 2047 2048 2049 2050 2051 2052 2053 2054 2055 2056 2057 2058 2059 2060 2061 2062 2063 2064 2065 2066 2067 2068 2069 2070 2071 2072 2073 2074 2075 2076 2077 2078 2079 2080 2081 2082 2083 2084 2085 2086 2087 2088 2089 2090 2091 2092 2093 2094 2095 2096 2097 2098 2099 2100 2101 2102 2103 2104 2105 2106 2107 2108 2109 2110 2111 2112 2113 2114 2115 2116 2117 2118 2119 2120 2121 2122 2123 2124 2125 2126 2127 2128 2129 2130 2131 2132 2133 2134 2135 2136 2137 2138 2139 2140 2141 2142 2143 2144 2145 2146 2147 2148 2149 2150 2151 2152 2153 2154 2155 2156 2157 2158 2159 2160 2161 2162 2163 2164 2165 2166 2167 2168 2169 2170 2171 2172 2173 2174 2175 2176 2177 2178 2179 2180 2181 2182 2183 2184 2185 2186 2187 2188 2189 2190 2191 2192 2193 2194 2195 2196 2197 2198 2199 2200 2201 2202 2203 2204 2205 2206 2207 2208 2209 2210 2211 2212 2213 2214 2215 2216 2217 2218 2219 2220 2221 2222 2223 2224 2225 2226 2227 2228 2229 2230 2231 2232 2233 2234 2235 2236 2237 2238 2239 2240 2241 2242 2243 2244 2245 2246 2247 2248 2249 2250 2251 2252 2253 2254 2255 2256 2257 2258 2259 2260 2261 2262 2263 2264 2265 2266 2267 2268 2269 2270 2271 2272 2273 2274 2275 2276 2277 2278 2279 2280 2281 2282 2283 2284 2285 2286 2287 2288 2289 2290 2291 2292 2293 2294 2295 2296 2297 2298 2299 2300 2301 2302 2303 2304 2305 2306 2307 2308 2309 2310 2311 2312 2313 2314 2315 2316 2317 2318 2319 2320 2321 2322 2323 2324 2325 2326 2327 2328 2329 2330 2331 2332 2333 2334 2335 2336 2337 2338 2339 2340 2341 2342 2343 2344 2345 2346 2347 2348 2349 2350 2351 2352 2353 2354 2355 2356 2357 2358 2359 2360 2361 2362 2363 2364 2365 2366 2367 2368 2369 2370 2371 2372 2373 2374 2375 2376 2377 2378 2379 2380 2381 2382 2383 2384 2385 2386 2387 2388 2389 2390 2391 2392 2393 2394 2395 2396 2397 2398 2399 2400 2401 2402 2403 2404 2405 2406 2407 2408 2409 2410 2411 2412 2413 2414 2415 2416 2417 2418 2419 2420 2421 2422 2423 2424 2425 2426 2427 2428 2429 2430 2431 2432 2433 2434 2435 2436 2437 2438 2439 2440 2441 2442 2443 2444 2445 2446 2447 2448 2449 2450 2451 2452 2453 2454 2455 2456 2457 2458 2459 2460 2461 2462 2463 2464 2465 2466 2467 2468 2469 2470 2471 2472 2473 2474 2475 2476 2477 2478 2479 2480 2481 2482 2483 2484 2485 2486 2487 2488 2489 2490 2491 2492 2493 2494 2495 2496 2497 2498 2499 2500 2501 2502 2503 2504 2505 2506 2507 2508 2509 2510 2511 2512 2513 2514 2515 2516 2517 2518 2519 2520 2521 2522 2523 2524 2525 2526 2527 2528 2529 2530 2531 2532 2533 2534 2535 2536 2537 2538 2539 2540 2541 2542 2543 2544 2545 2546 2547 2548 2549 2550 2551 2552 2553 2554 2555 2556 2557 2558 2559 2560 2561 2562 2563 2564 2565 2566 2567 2568 2569 2570 2571 2572 2573 2574 2575 2576 2577 2578 2579 2580 2581 2582 2583 2584 2585 2586 2587 2588 2589 2590 2591 2592 2593 2594 2595 2596 2597 2598 2599 2600 2601 2602 2603 2604 2605 2606 2607 2608 2609 2610 2611 2612 2613 2614 2615 2616 2617 2618 2619 2620 2621 2622 2623 2624 2625 2626 2627 2628 2629 2630 2631 2632 2633 2634 2635 2636 2637 2638 2639 2640 2641 2642 2643 2644 2645 2646 2647 2648 2649 2650 2651 2652 2653 2654 2655 2656 2657 2658 2659 2660 2661 2662 2663 2664 2665 2666 2667 2668 2669 2670 2671 2672 2673 2674 2675 2676 2677 2678 2679 2680 2681 2682 2683 2684 2685 2686 2687 2688 2689 2690 2691 2692 2693 2694 2695 2696 2697 2698 2699 2700 2701 2702 2703 2704 2705 2706 2707 2708 2709 2710 2711 2712 2713 2714 2715 2716
#!/usr/bin/env python

import socket
import unittest
import struct

from framework import VppTestCase, VppTestRunner, running_extended_tests
from scapy.layers.inet import IP, TCP, UDP, ICMP
from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6EchoReply
from scapy.layers.l2 import Ether, ARP
from scapy.data import IP_PROTOS
from scapy.packet import bind_layers
from util import ppp
from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
from time import sleep


class MethodHolder(VppTestCase):
    """ SNAT create capture and verify method holder """

    @classmethod
    def setUpClass(cls):
        super(MethodHolder, cls).setUpClass()

    def tearDown(self):
        super(MethodHolder, self).tearDown()

    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=20))
        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=20))
        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_in_ip6(self, in_if, out_if, hlim=64):
        """
        Create IPv6 packet stream for inside network

        :param in_if: Inside interface
        :param out_if: Outside interface
        :param ttl: Hop Limit of generated packets
        """
        pkts = []
        dst = ''.join(['64:ff9b::', out_if.remote_ip4])
        # TCP
        p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
             IPv6(src=in_if.remote_ip6, dst=dst, hlim=hlim) /
             TCP(sport=self.tcp_port_in, dport=20))
        pkts.append(p)

        # UDP
        p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
             IPv6(src=in_if.remote_ip6, dst=dst, hlim=hlim) /
             UDP(sport=self.udp_port_in, dport=20))
        pkts.append(p)

        # ICMP
        p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
             IPv6(src=in_if.remote_ip6, dst=dst, hlim=hlim) /
             ICMPv6EchoRequest(id=self.icmp_id_in))
        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 SNAT address)
        :param ttl: TTL of generated packets
        """
        if dst_ip is None:
            dst_ip = self.snat_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=20))
        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=20))
        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_id_out, type='echo-reply'))
        pkts.append(p)

        return pkts

    def verify_capture_out(self, capture, nat_ip=None, same_port=False,
                           packet_num=3, dst_ip=None):
        """
        Verify captured packets on outside network

        :param capture: Captured packets
        :param nat_ip: Translated IP address (Default use global SNAT address)
        :param same_port: Sorce port number is not translated (Default False)
        :param packet_num: Expected number of packets (Default 3)
        :param dst_ip: Destination IP address (Default do not verify)
        """
        if nat_ip is None:
            nat_ip = self.snat_addr
        self.assertEqual(packet_num, len(capture))
        for packet in capture:
            try:
                self.assertEqual(packet[IP].src, nat_ip)
                if dst_ip is not None:
                    self.assertEqual(packet[IP].dst, dst_ip)
                if packet.haslayer(TCP):
                    if same_port:
                        self.assertEqual(packet[TCP].sport, self.tcp_port_in)
                    else:
                        self.assertNotEqual(
                            packet[TCP].sport, self.tcp_port_in)
                    self.tcp_port_out = packet[TCP].sport
                elif packet.haslayer(UDP):
                    if same_port:
                        self.assertEqual(packet[UDP].sport, self.udp_port_in)
                    else:
                        self.assertNotEqual(
                            packet[UDP].sport, self.udp_port_in)
                    self.udp_port_out = packet[UDP].sport
                else:
                    if same_port:
                        self.assertEqual(packet[ICMP].id, self.icmp_id_in)
                    else:
                        self.assertNotEqual(packet[ICMP].id, self.icmp_id_in)
                    self.icmp_id_out = packet[ICMP].id
            except:
                self.logger.error(ppp("Unexpected or invalid packet "
                                      "(outside network):", packet))
                raise

    def verify_capture_in(self, capture, in_if, packet_num=3):
        """
        Verify captured packets on inside network

        :param capture: Captured packets
        :param in_if: Inside interface
        :param packet_num: Expected number of packets (Default 3)
        """
        self.assertEqual(packet_num, len(capture))
        for packet in capture:
            try:
                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:
                self.logger.error(ppp("Unexpected or invalid packet "
                                      "(inside network):", packet))
                raise

    def verify_capture_in_ip6(self, capture, src_ip, dst_ip, packet_num=3):
        """
        Verify captured IPv6 packets on inside network

        :param capture: Captured packets
        :param src_ip: Source IP
        :param dst_ip: Destination IP address
        :param packet_num: Expected number of packets (Default 3)
        """
        self.assertEqual(packet_num, len(capture))
        for packet in capture:
            try:
                self.assertEqual(packet[IPv6].src, src_ip)
                self.assertEqual(packet[IPv6].dst, dst_ip)
                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[ICMPv6EchoReply].id,
                                     self.icmp_id_in)
            except:
                self.logger.error(ppp("Unexpected or invalid packet "
                                      "(inside network):", packet))
                raise

    def verify_capture_no_translation(self, capture, ingress_if, egress_if):
        """
        Verify captured packet that don't have to be translated

        :param capture: Captured packets
        :param ingress_if: Ingress interface
        :param egress_if: Egress interface
        """
        for packet in capture:
            try:
                self.assertEqual(packet[IP].src, ingress_if.remote_ip4)
                self.assertEqual(packet[IP].dst, egress_if.remote_ip4)
                if packet.haslayer(TCP):
                    self.assertEqual(packet[TCP].sport, self.tcp_port_in)
                elif packet.haslayer(UDP):
                    self.assertEqual(packet[UDP].sport, self.udp_port_in)
                else:
                    self.assertEqual(packet[ICMP].id, self.icmp_id_in)
            except:
                self.logger.error(ppp("Unexpected or invalid packet "
                                      "(inside network):", packet))
                raise

    def verify_capture_out_with_icmp_errors(self, capture, src_ip=None,
                                            packet_num=3, icmp_type=11):
        """
        Verify captured packets with ICMP errors on outside network

        :param capture: Captured packets
        :param src_ip: Translated IP address or IP address of VPP
                       (Default use global SNAT address)
        :param packet_num: Expected number of packets (Default 3)
        :param icmp_type: Type of error ICMP packet
                          we are expecting (Default 11)
        """
        if src_ip is None:
            src_ip = self.snat_addr
        self.assertEqual(packet_num, len(capture))
        for packet in capture:
            try:
                self.assertEqual(packet[IP].src, src_ip)
                self.assertTrue(packet.haslayer(ICMP))
                icmp = packet[ICMP]
                self.assertEqual(icmp.type, icmp_type)
                self.assertTrue(icmp.haslayer(IPerror))
                inner_ip = icmp[IPerror]
                if inner_ip.haslayer(TCPerror):
                    self.assertEqual(inner_ip[TCPerror].dport,
                                     self.tcp_port_out)
                elif inner_ip.haslayer(UDPerror):
                    self.assertEqual(inner_ip[UDPerror].dport,
                                     self.udp_port_out)
                else:
                    self.assertEqual(inner_ip[ICMPerror].id, self.icmp_id_out)
            except:
                self.logger.error(ppp("Unexpected or invalid packet "
                                      "(outside network):", packet))
                raise

    def verify_capture_in_with_icmp_errors(self, capture, in_if, packet_num=3,
                                           icmp_type=11):
        """
        Verify captured packets with ICMP errors on inside network

        :param capture: Captured packets
        :param in_if: Inside interface
        :param packet_num: Expected number of packets (Default 3)
        :param icmp_type: Type of error ICMP packet
                          we are expecting (Default 11)
        """
        self.assertEqual(packet_num, len(capture))
        for packet in capture:
            try:
                self.assertEqual(packet[IP].dst, in_if.remote_ip4)
                self.assertTrue(packet.haslayer(ICMP))
                icmp = packet[ICMP]
                self.assertEqual(icmp.type, icmp_type)
                self.assertTrue(icmp.haslayer(IPerror))
                inner_ip = icmp[IPerror]
                if inner_ip.haslayer(TCPerror):
                    self.assertEqual(inner_ip[TCPerror].sport,
                                     self.tcp_port_in)
                elif inner_ip.haslayer(UDPerror):
                    self.assertEqual(inner_ip[UDPerror].sport,
                                     self.udp_port_in)
                else:
                    self.assertEqual(inner_ip[ICMPerror].id, self.icmp_id_in)
            except:
                self.logger.error(ppp("Unexpected or invalid packet "
                                      "(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])


class TestSNAT(MethodHolder):
    """ SNAT Test Cases """

    @classmethod
    def setUpClass(cls):
        super(TestSNAT, cls).setUpClass()

        try:
            cls.tcp_port_in = 6303
            cls.tcp_port_out = 6303
            cls.udp_port_in = 6304
            cls.udp_port_out = 6304
            cls.icmp_id_in = 6305
            cls.icmp_id_out = 6305
            cls.snat_addr = '10.0.0.3'
            cls.ipfix_src_port = 4739
            cls.ipfix_domain_id = 1

            cls.create_pg_interfaces(range(9))
            cls.interfaces = list(cls.pg_interfaces[0:4])

            for i in cls.interfaces:
                i.admin_up()
                i.config_ip4()
                i.resolve_arp()

            cls.pg0.generate_remote_hosts(3)
            cls.pg0.configure_ipv4_neighbors()

            cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7]))

            cls.pg4._local_ip4 = "172.16.255.1"
            cls.pg4._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
            cls.pg4._remote_hosts[0]._ip4 = "172.16.255.2"
            cls.pg4.set_table_ip4(10)
            cls.pg5._local_ip4 = "172.16.255.3"
            cls.pg5._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
            cls.pg5._remote_hosts[0]._ip4 = "172.16.255.4"
            cls.pg5.set_table_ip4(10)
            cls.pg6._local_ip4 = "172.16.255.1"
            cls.pg6._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
            cls.pg6._remote_hosts[0]._ip4 = "172.16.255.2"
            cls.pg6.set_table_ip4(20)
            for i in cls.overlapping_interfaces:
                i.config_ip4()
                i.admin_up()
                i.resolve_arp()

            cls.pg7.admin_up()
            cls.pg8.admin_up()

        except Exception:
            super(TestSNAT, cls).tearDownClass()
            raise

    def clear_snat(self):
        """
        Clear SNAT configuration.
        """
        # I found no elegant way to do this
        self.vapi.ip_add_del_route(dst_address=self.pg7.remote_ip4n,
                                   dst_address_length=32,
                                   next_hop_address=self.pg7.remote_ip4n,
                                   next_hop_sw_if_index=self.pg7.sw_if_index,
                                   is_add=0)
        self.vapi.ip_add_del_route(dst_address=self.pg8.remote_ip4n,
                                   dst_address_length=32,
                                   next_hop_address=self.pg8.remote_ip4n,
                                   next_hop_sw_if_index=self.pg8.sw_if_index,
                                   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)

        if self.pg7.has_ip4_config:
            self.pg7.unconfig_ip4()

        interfaces = self.vapi.snat_interface_addr_dump()
        for intf in interfaces:
            self.vapi.snat_add_interface_addr(intf.sw_if_index, is_add=0)

        self.vapi.snat_ipfix(enable=0, src_port=self.ipfix_src_port,
                             domain_id=self.ipfix_domain_id)
        self.ipfix_src_port = 4739
        self.ipfix_domain_id = 1

        interfaces = self.vapi.snat_interface_dump()
        for intf in interfaces:
            self.vapi.snat_interface_add_del_feature(intf.sw_if_index,
                                                     intf.is_inside,
                                                     is_add=0)

        static_mappings = self.vapi.snat_static_mapping_dump()
        for sm in static_mappings:
            self.vapi.snat_add_static_mapping(sm.local_ip_address,
                                              sm.external_ip_address,
                                              local_port=sm.local_port,
                                              external_port=sm.external_port,
                                              addr_only=sm.addr_only,
                                              vrf_id=sm.vrf_id,
                                              protocol=sm.protocol,
                                              is_add=0)

        adresses = self.vapi.snat_address_dump()
        for addr in adresses:
            self.vapi.snat_add_address_range(addr.ip_address,
                                             addr.ip_address,
                                             is_add=0)

    def snat_add_static_mapping(self, local_ip, external_ip='0.0.0.0',
                                local_port=0, external_port=0, vrf_id=0,
                                is_add=1, external_sw_if_index=0xFFFFFFFF,
                                proto=0):
        """
        Add/delete S-NAT static mapping

        :param local_ip: Local IP address
        :param external_ip: External IP address
        :param local_port: Local port number (Optional)
        :param external_port: External port number (Optional)
        :param vrf_id: VRF ID (Default 0)
        :param is_add: 1 if add, 0 if delete (Default add)
        :param external_sw_if_index: External interface instead of IP address
        :param proto: IP protocol (Mandatory if port specified)
        """
        addr_only = 1
        if local_port and external_port:
            addr_only = 0
        l_ip = socket.inet_pton(socket.AF_INET, local_ip)
        e_ip = socket.inet_pton(socket.AF_INET, external_ip)
        self.vapi.snat_add_static_mapping(
            l_ip,
            e_ip,
            external_sw_if_index,
            local_port,
            external_port,
            addr_only,
            vrf_id,
            proto,
            is_add)

    def snat_add_address(self, ip, is_add=1, vrf_id=0xFFFFFFFF):
        """
        Add/delete S-NAT address

        :param ip: IP address
        :param is_add: 1 if add, 0 if delete (Default add)
        """
        snat_addr = socket.inet_pton(socket.AF_INET, ip)
        self.vapi.snat_add_address_range(snat_addr, snat_addr, is_add,
                                         vrf_id=vrf_id)

    def test_dynamic(self):
        """ SNAT dynamic translation test """

        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)

        # 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)

        # out2in
        pkts = self.create_stream_out(self.pg1)
        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)

    def test_dynamic_icmp_errors_in2out_ttl_1(self):
        """ SNAT handling of client packets with TTL=1 """

        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)

        # Client side - generate traffic
        pkts = self.create_stream_in(self.pg0, self.pg1, ttl=1)
        self.pg0.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()

        # Client side - verify ICMP type 11 packets
        capture = self.pg0.get_capture(len(pkts))
        self.verify_capture_in_with_icmp_errors(capture, self.pg0)

    def test_dynamic_icmp_errors_out2in_ttl_1(self):
        """ SNAT handling of server packets with TTL=1 """

        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)

        # Client side - 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()

        # Server side - generate traffic
        capture = self.pg1.get_capture(len(pkts))
        self.verify_capture_out(capture)
        pkts = self.create_stream_out(self.pg1, ttl=1)
        self.pg1.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()

        # Server side - verify ICMP type 11 packets
        capture = self.pg1.get_capture(len(pkts))
        self.verify_capture_out_with_icmp_errors(capture,
                                                 src_ip=self.pg1.local_ip4)

    def test_dynamic_icmp_errors_in2out_ttl_2(self):
        """ SNAT handling of error responses to client packets with TTL=2 """

        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)

        # Client side - generate traffic
        pkts = self.create_stream_in(self.pg0, self.pg1, ttl=2)
        self.pg0.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()

        # Server side - simulate ICMP type 11 response
        capture = self.pg1.get_capture(len(pkts))
        pkts = [Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
                IP(src=self.pg1.remote_ip4, dst=self.snat_addr) /
                ICMP(type=11) / packet[IP] for packet in capture]
        self.pg1.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()

        # Client side - verify ICMP type 11 packets
        capture = self.pg0.get_capture(len(pkts))
        self.verify_capture_in_with_icmp_errors(capture, self.pg0)

    def test_dynamic_icmp_errors_out2in_ttl_2(self):
        """ SNAT handling of error responses to server packets with TTL=2 """

        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)

        # Client side - 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()

        # Server side - generate traffic
        capture = self.pg1.get_capture(len(pkts))
        self.verify_capture_out(capture)
        pkts = self.create_stream_out(self.pg1, ttl=2)
        self.pg1.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()

        # Client side - simulate ICMP type 11 response
        capture = self.pg0.get_capture(len(pkts))
        pkts = [Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
                IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
                ICMP(type=11) / packet[IP] for packet in capture]
        self.pg0.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()

        # Server side - verify ICMP type 11 packets
        capture = self.pg1.get_capture(len(pkts))
        self.verify_capture_out_with_icmp_errors(capture)

    def test_ping_out_interface_from_outside(self):
        """ Ping SNAT out interface from outside network """

        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)

        p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
             IP(src=self.pg1.remote_ip4, dst=self.pg1.local_ip4) /
             ICMP(id=self.icmp_id_out, type='echo-request'))
        pkts = [p]
        self.pg1.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg1.get_capture(len(pkts))
        self.assertEqual(1, len(capture))
        packet = capture[0]
        try:
            self.assertEqual(packet[IP].src, self.pg1.local_ip4)
            self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
            self.assertEqual(packet[ICMP].id, self.icmp_id_in)
            self.assertEqual(packet[ICMP].type, 0)  # echo reply
        except:
            self.logger.error(ppp("Unexpected or invalid packet "
                                  "(outside network):", packet))
            raise

    def test_ping_internal_host_from_outside(self):
        """ Ping internal host from outside network """

        self.snat_add_static_mapping(self.pg0.remote_ip4, 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)

        # out2in
        pkt = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
               IP(src=self.pg1.remote_ip4, dst=self.snat_addr, ttl=64) /
               ICMP(id=self.icmp_id_out, type='echo-request'))
        self.pg1.add_stream(pkt)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg0.get_capture(1)
        self.verify_capture_in(capture, self.pg0, packet_num=1)
        self.assert_equal(capture[0][IP].proto, IP_PROTOS.icmp)

        # in2out
        pkt = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
               IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, ttl=64) /
               ICMP(id=self.icmp_id_in, type='echo-reply'))
        self.pg0.add_stream(pkt)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg1.get_capture(1)
        self.verify_capture_out(capture, same_port=True, packet_num=1)
        self.assert_equal(capture[0][IP].proto, IP_PROTOS.icmp)

    def test_static_in(self):
        """ SNAT 1:1 NAT initialized from inside network """

        nat_ip = "10.0.0.10"
        self.tcp_port_out = 6303
        self.udp_port_out = 6304
        self.icmp_id_out = 6305

        self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
        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)

        # 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, True)

        # 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)

    def test_static_out(self):
        """ SNAT 1:1 NAT initialized from outside network """

        nat_ip = "10.0.0.20"
        self.tcp_port_out = 6303
        self.udp_port_out = 6304
        self.icmp_id_out = 6305

        self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
        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)

        # 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)

        # 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, True)

    def test_static_with_port_in(self):
        """ SNAT 1:1 NAT with port initialized from inside network """

        self.tcp_port_out = 3606
        self.udp_port_out = 3607
        self.icmp_id_out = 3608

        self.snat_add_address(self.snat_addr)
        self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
                                     self.tcp_port_in, self.tcp_port_out,
                                     proto=IP_PROTOS.tcp)
        self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
                                     self.udp_port_in, self.udp_port_out,
                                     proto=IP_PROTOS.udp)
        self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
                                     self.icmp_id_in, self.icmp_id_out,
                                     proto=IP_PROTOS.icmp)
        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)

        # 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)

        # out2in
        pkts = self.create_stream_out(self.pg1)
        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)

    def test_static_with_port_out(self):
        """ SNAT 1:1 NAT with port initialized from outside network """

        self.tcp_port_out = 30606
        self.udp_port_out = 30607
        self.icmp_id_out = 30608

        self.snat_add_address(self.snat_addr)
        self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
                                     self.tcp_port_in, self.tcp_port_out,
                                     proto=IP_PROTOS.tcp)
        self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
                                     self.udp_port_in, self.udp_port_out,
                                     proto=IP_PROTOS.udp)
        self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
                                     self.icmp_id_in, self.icmp_id_out,
                                     proto=IP_PROTOS.icmp)
        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)

        # out2in
        pkts = self.create_stream_out(self.pg1)
        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)

        # 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)

    def test_static_vrf_aware(self):
        """ SNAT 1:1 NAT VRF awareness """

        nat_ip1 = "10.0.0.30"
        nat_ip2 = "10.0.0.40"
        self.tcp_port_out = 6303
        self.udp_port_out = 6304
        self.icmp_id_out = 6305

        self.snat_add_static_mapping(self.pg4.remote_ip4, nat_ip1,
                                     vrf_id=10)
        self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip2,
                                     vrf_id=10)
        self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
                                                 is_inside=0)
        self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
        self.vapi.snat_interface_add_del_feature(self.pg4.sw_if_index)

        # inside interface VRF match SNAT static mapping VRF
        pkts = self.create_stream_in(self.pg4, self.pg3)
        self.pg4.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg3.get_capture(len(pkts))
        self.verify_capture_out(capture, nat_ip1, True)

        # inside interface VRF don't match SNAT static mapping VRF (packets
        # are dropped)
        pkts = self.create_stream_in(self.pg0, self.pg3)
        self.pg0.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        self.pg3.assert_nothing_captured()

    def test_multiple_inside_interfaces(self):
        """ SNAT multiple inside interfaces (non-overlapping address space) """

        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)
        self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
                                                 is_inside=0)

        # between two S-NAT inside interfaces (no translation)
        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_no_translation(capture, self.pg0, self.pg1)

        # from S-NAT inside to interface without S-NAT feature (no translation)
        pkts = self.create_stream_in(self.pg0, self.pg2)
        self.pg0.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg2.get_capture(len(pkts))
        self.verify_capture_no_translation(capture, self.pg0, self.pg2)

        # in2out 1st interface
        pkts = self.create_stream_in(self.pg0, self.pg3)
        self.pg0.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg3.get_capture(len(pkts))
        self.verify_capture_out(capture)

        # out2in 1st interface
        pkts = self.create_stream_out(self.pg3)
        self.pg3.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)

        # in2out 2nd interface
        pkts = self.create_stream_in(self.pg1, self.pg3)
        self.pg1.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg3.get_capture(len(pkts))
        self.verify_capture_out(capture)

        # out2in 2nd interface
        pkts = self.create_stream_out(self.pg3)
        self.pg3.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg1.get_capture(len(pkts))
        self.verify_capture_in(capture, self.pg1)

    def test_inside_overlapping_interfaces(self):
        """ SNAT multiple inside interfaces with overlapping address space """

        static_nat_ip = "10.0.0.10"
        self.snat_add_address(self.snat_addr)
        self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
                                                 is_inside=0)
        self.vapi.snat_interface_add_del_feature(self.pg4.sw_if_index)
        self.vapi.snat_interface_add_del_feature(self.pg5.sw_if_index)
        self.vapi.snat_interface_add_del_feature(self.pg6.sw_if_index)
        self.snat_add_static_mapping(self.pg6.remote_ip4, static_nat_ip,
                                     vrf_id=20)

        # between S-NAT inside interfaces with same VRF (no translation)
        pkts = self.create_stream_in(self.pg4, self.pg5)
        self.pg4.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg5.get_capture(len(pkts))
        self.verify_capture_no_translation(capture, self.pg4, self.pg5)

        # between S-NAT inside interfaces with different VRF (hairpinning)
        p = (Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac) /
             IP(src=self.pg4.remote_ip4, dst=static_nat_ip) /
             TCP(sport=1234, dport=5678))
        self.pg4.add_stream(p)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg6.get_capture(1)
        p = capture[0]
        try:
            ip = p[IP]
            tcp = p[TCP]
            self.assertEqual(ip.src, self.snat_addr)
            self.assertEqual(ip.dst, self.pg6.remote_ip4)
            self.assertNotEqual(tcp.sport, 1234)
            self.assertEqual(tcp.dport, 5678)
        except:
            self.logger.error(ppp("Unexpected or invalid packet:", p))
            raise

        # in2out 1st interface
        pkts = self.create_stream_in(self.pg4, self.pg3)
        self.pg4.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg3.get_capture(len(pkts))
        self.verify_capture_out(capture)

        # out2in 1st interface
        pkts = self.create_stream_out(self.pg3)
        self.pg3.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg4.get_capture(len(pkts))
        self.verify_capture_in(capture, self.pg4)

        # in2out 2nd interface
        pkts = self.create_stream_in(self.pg5, self.pg3)
        self.pg5.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg3.get_capture(len(pkts))
        self.verify_capture_out(capture)

        # out2in 2nd interface
        pkts = self.create_stream_out(self.pg3)
        self.pg3.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg5.get_capture(len(pkts))
        self.verify_capture_in(capture, self.pg5)

        # pg5 session dump
        addresses = self.vapi.snat_address_dump()
        self.assertEqual(len(addresses), 1)
        sessions = self.vapi.snat_user_session_dump(self.pg5.remote_ip4n, 10)
        self.assertEqual(len(sessions), 3)
        for session in sessions:
            self.assertFalse(session.is_static)
            self.assertEqual(session.inside_ip_address[0:4],
                             self.pg5.remote_ip4n)
            self.assertEqual(session.outside_ip_address,
                             addresses[0].ip_address)
        self.assertEqual(sessions[0].protocol, IP_PROTOS.tcp)
        self.assertEqual(sessions[1].protocol, IP_PROTOS.udp)
        self.assertEqual(sessions[2].protocol, IP_PROTOS.icmp)
        self.assertEqual(sessions[0].inside_port, self.tcp_port_in)
        self.assertEqual(sessions[1].inside_port, self.udp_port_in)
        self.assertEqual(sessions[2].inside_port, self.icmp_id_in)
        self.assertEqual(sessions[0].outside_port, self.tcp_port_out)
        self.assertEqual(sessions[1].outside_port, self.udp_port_out)
        self.assertEqual(sessions[2].outside_port, self.icmp_id_out)

        # in2out 3rd interface
        pkts = self.create_stream_in(self.pg6, self.pg3)
        self.pg6.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg3.get_capture(len(pkts))
        self.verify_capture_out(capture, static_nat_ip, True)

        # out2in 3rd interface
        pkts = self.create_stream_out(self.pg3, static_nat_ip)
        self.pg3.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg6.get_capture(len(pkts))
        self.verify_capture_in(capture, self.pg6)

        # general user and session dump verifications
        users = self.vapi.snat_user_dump()
        self.assertTrue(len(users) >= 3)
        addresses = self.vapi.snat_address_dump()
        self.assertEqual(len(addresses), 1)
        for user in users:
            sessions = self.vapi.snat_user_session_dump(user.ip_address,
                                                        user.vrf_id)
            for session in sessions:
                self.assertEqual(user.ip_address, session.inside_ip_address)
                self.assertTrue(session.total_bytes > session.total_pkts > 0)
                self.assertTrue(session.protocol in
                                [IP_PROTOS.tcp, IP_PROTOS.udp,
                                 IP_PROTOS.icmp])

        # pg4 session dump
        sessions = self.vapi.snat_user_session_dump(self.pg4.remote_ip4n, 10)
        self.assertTrue(len(sessions) >= 4)
        for session in sessions:
            self.assertFalse(session.is_static)
            self.assertEqual(session.inside_ip_address[0:4],
                             self.pg4.remote_ip4n)
            self.assertEqual(session.outside_ip_address,
                             addresses[0].ip_address)

        # pg6 session dump
        sessions = self.vapi.snat_user_session_dump(self.pg6.remote_ip4n, 20)
        self.assertTrue(len(sessions) >= 3)
        for session in sessions:
            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.assertTrue(session.inside_port in
                            [self.tcp_port_in, self.udp_port_in,
                             self.icmp_id_in])

    def test_hairpinning(self):
        """ SNAT hairpinning - 1:1 NAT with port"""

        host = self.pg0.remote_hosts[0]
        server = self.pg0.remote_hosts[1]
        host_in_port = 1234
        host_out_port = 0
        server_in_port = 5678
        server_out_port = 8765

        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)
        # add static mapping for server
        self.snat_add_static_mapping(server.ip4, self.snat_addr,
                                     server_in_port, server_out_port,
                                     proto=IP_PROTOS.tcp)

        # send packet from host to server
        p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
             IP(src=host.ip4, dst=self.snat_addr) /
             TCP(sport=host_in_port, dport=server_out_port))
        self.pg0.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.snat_addr)
            self.assertEqual(ip.dst, server.ip4)
            self.assertNotEqual(tcp.sport, host_in_port)
            self.assertEqual(tcp.dport, server_in_port)
            host_out_port = tcp.sport
        except:
            self.logger.error(ppp("Unexpected or invalid packet:", p))
            raise

        # send reply from server to host
        p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
             IP(src=server.ip4, dst=self.snat_addr) /
             TCP(sport=server_in_port, dport=host_out_port))
        self.pg0.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.snat_addr)
            self.assertEqual(ip.dst, host.ip4)
            self.assertEqual(tcp.sport, server_out_port)
            self.assertEqual(tcp.dport, host_in_port)
        except:
            self.logger.error(ppp("Unexpected or invalid packet:"), p)
            raise

    def test_hairpinning2(self):
        """ SNAT hairpinning - 1:1 NAT"""

        server1_nat_ip = "10.0.0.10"
        server2_nat_ip = "10.0.0.11"
        host = self.pg0.remote_hosts[0]
        server1 = self.pg0.remote_hosts[1]
        server2 = self.pg0.remote_hosts[2]
        server_tcp_port = 22
        server_udp_port = 20

        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)

        # add static mapping for servers
        self.snat_add_static_mapping(server1.ip4, server1_nat_ip)
        self.snat_add_static_mapping(server2.ip4, server2_nat_ip)

        # host to server1
        pkts = []
        p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
             IP(src=host.ip4, dst=server1_nat_ip) /
             TCP(sport=self.tcp_port_in, dport=server_tcp_port))
        pkts.append(p)
        p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
             IP(src=host.ip4, dst=server1_nat_ip) /
             UDP(sport=self.udp_port_in, dport=server_udp_port))
        pkts.append(p)
        p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
             IP(src=host.ip4, dst=server1_nat_ip) /
             ICMP(id=self.icmp_id_in, type='echo-request'))
        pkts.append(p)
        self.pg0.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg0.get_capture(len(pkts))
        for packet in capture:
            try:
                self.assertEqual(packet[IP].src, self.snat_addr)
                self.assertEqual(packet[IP].dst, server1.ip4)
                if packet.haslayer(TCP):
                    self.assertNotEqual(packet[TCP].sport, self.tcp_port_in)
                    self.assertEqual(packet[TCP].dport, server_tcp_port)
                    self.tcp_port_out = packet[TCP].sport
                elif packet.haslayer(UDP):
                    self.assertNotEqual(packet[UDP].sport, self.udp_port_in)
                    self.assertEqual(packet[UDP].dport, server_udp_port)
                    self.udp_port_out = packet[UDP].sport
                else:
                    self.assertNotEqual(packet[ICMP].id, self.icmp_id_in)
                    self.icmp_id_out = packet[ICMP].id
            except:
                self.logger.error(ppp("Unexpected or invalid packet:", packet))
                raise

        # server1 to host
        pkts = []
        p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
             IP(src=server1.ip4, dst=self.snat_addr) /
             TCP(sport=server_tcp_port, dport=self.tcp_port_out))
        pkts.append(p)
        p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
             IP(src=server1.ip4, dst=self.snat_addr) /
             UDP(sport=server_udp_port, dport=self.udp_port_out))
        pkts.append(p)
        p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
             IP(src=server1.ip4, dst=self.snat_addr) /
             ICMP(id=self.icmp_id_out, type='echo-reply'))
        pkts.append(p)
        self.pg0.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg0.get_capture(len(pkts))
        for packet in capture:
            try:
                self.assertEqual(packet[IP].src, server1_nat_ip)
                self.assertEqual(packet[IP].dst, host.ip4)
                if packet.haslayer(TCP):
                    self.assertEqual(packet[TCP].dport, self.tcp_port_in)
                    self.assertEqual(packet[TCP].sport, server_tcp_port)
                elif packet.haslayer(UDP):
                    self.assertEqual(packet[UDP].dport, self.udp_port_in)
                    self.assertEqual(packet[UDP].sport, server_udp_port)
                else:
                    self.assertEqual(packet[ICMP].id, self.icmp_id_in)
            except:
                self.logger.error(ppp("Unexpected or invalid packet:", packet))
                raise

        # server2 to server1
        pkts = []
        p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
             IP(src=server2.ip4, dst=server1_nat_ip) /
             TCP(sport=self.tcp_port_in, dport=server_tcp_port))
        pkts.append(p)
        p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
             IP(src=server2.ip4, dst=server1_nat_ip) /
             UDP(sport=self.udp_port_in, dport=server_udp_port))
        pkts.append(p)
        p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
             IP(src=server2.ip4, dst=server1_nat_ip) /
             ICMP(id=self.icmp_id_in, type='echo-request'))
        pkts.append(p)
        self.pg0.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg0.get_capture(len(pkts))
        for packet in capture:
            try:
                self.assertEqual(packet[IP].src, server2_nat_ip)
                self.assertEqual(packet[IP].dst, server1.ip4)
                if packet.haslayer(TCP):
                    self.assertEqual(packet[TCP].sport, self.tcp_port_in)
                    self.assertEqual(packet[TCP].dport, server_tcp_port)
                    self.tcp_port_out = packet[TCP].sport
                elif packet.haslayer(UDP):
                    self.assertEqual(packet[UDP].sport, self.udp_port_in)
                    self.assertEqual(packet[UDP].dport, server_udp_port)
                    self.udp_port_out = packet[UDP].sport
                else:
                    self.assertEqual(packet[ICMP].id, self.icmp_id_in)
                    self.icmp_id_out = packet[ICMP].id
            except:
                self.logger.error(ppp("Unexpected or invalid packet:", packet))
                raise

        # server1 to server2
        pkts = []
        p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
             IP(src=server1.ip4, dst=server2_nat_ip) /
             TCP(sport=server_tcp_port, dport=self.tcp_port_out))
        pkts.append(p)
        p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
             IP(src=server1.ip4, dst=server2_nat_ip) /
             UDP(sport=server_udp_port, dport=self.udp_port_out))
        pkts.append(p)
        p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
             IP(src=server1.ip4, dst=server2_nat_ip) /
             ICMP(id=self.icmp_id_out, type='echo-reply'))
        pkts.append(p)
        self.pg0.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg0.get_capture(len(pkts))
        for packet in capture:
            try:
                self.assertEqual(packet[IP].src, server1_nat_ip)
                self.assertEqual(packet[IP].dst, server2.ip4)
                if packet.haslayer(TCP):
                    self.assertEqual(packet[TCP].dport, self.tcp_port_in)
                    self.assertEqual(packet[TCP].sport, server_tcp_port)
                elif packet.haslayer(UDP):
                    self.assertEqual(packet[UDP].dport, self.udp_port_in)
                    self.assertEqual(packet[UDP].sport, server_udp_port)
                else:
                    self.assertEqual(packet[ICMP].id, self.icmp_id_in)
            except:
                self.logger.error(ppp("Unexpected or invalid packet:", packet))
                raise

    def test_max_translations_per_user(self):
        """ MAX translations per user - recycle the least recently used """

        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)

        # get maximum number of translations per user
        snat_config = self.vapi.snat_show_config()

        # send more than maximum number of translations per user packets
        pkts_num = snat_config.max_translations_per_user + 5
        pkts = []
        for port in range(0, pkts_num):
            p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
                 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
                 TCP(sport=1025 + port))
            pkts.append(p)
        self.pg0.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()

        # verify number of translated packet
        self.pg1.get_capture(pkts_num)

    def test_interface_addr(self):
        """ Acquire SNAT addresses from interface """
        self.vapi.snat_add_interface_addr(self.pg7.sw_if_index)

        # no address in NAT pool
        adresses = self.vapi.snat_address_dump()
        self.assertEqual(0, len(adresses))

        # configure interface address and check NAT address pool
        self.pg7.config_ip4()
        adresses = self.vapi.snat_address_dump()
        self.assertEqual(1, len(adresses))
        self.assertEqual(adresses[0].ip_address[0:4], self.pg7.local_ip4n)

        # remove interface address and check NAT address pool
        self.pg7.unconfig_ip4()
        adresses = self.vapi.snat_address_dump()
        self.assertEqual(0, len(adresses))

    def test_interface_addr_static_mapping(self):
        """ Static mapping with addresses from interface """
        self.vapi.snat_add_interface_addr(self.pg7.sw_if_index)
        self.snat_add_static_mapping('1.2.3.4',
                                     external_sw_if_index=self.pg7.sw_if_index)

        # static mappings with external interface
        static_mappings = self.vapi.snat_static_mapping_dump()
        self.assertEqual(1, len(static_mappings))
        self.assertEqual(self.pg7.sw_if_index,
                         static_mappings[0].external_sw_if_index)

        # configure interface address and check static mappings
        self.pg7.config_ip4()
        static_mappings = self.vapi.snat_static_mapping_dump()
        self.assertEqual(1, len(static_mappings))
        self.assertEqual(static_mappings[0].external_ip_address[0:4],
                         self.pg7.local_ip4n)
        self.assertEqual(0xFFFFFFFF, static_mappings[0].external_sw_if_index)

        # remove interface address and check static mappings
        self.pg7.unconfig_ip4()
        static_mappings = self.vapi.snat_static_mapping_dump()
        self.assertEqual(0, len(static_mappings))

    def test_ipfix_nat44_sess(self):
        """ S-NAT IPFIX logging NAT44 session created/delted """
        self.ipfix_domain_id = 10
        self.ipfix_src_port = 20202
        colector_port = 30303
        bind_layers(UDP, IPFIX, dport=30303)
        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,
                                     collector_port=colector_port)
        self.vapi.snat_ipfix(domain_id=self.ipfix_domain_id,
                             src_port=self.ipfix_src_port)

        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))
            self.assertEqual(p[IP].src, self.pg3.local_ip4)
            self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
            self.assertEqual(p[UDP].sport, self.ipfix_src_port)
            self.assertEqual(p[UDP].dport, colector_port)
            self.assertEqual(p[IPFIX].observationDomainID,
                             self.ipfix_domain_id)
            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(domain_id=self.ipfix_domain_id,
                             src_port=self.ipfix_src_port)

        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))
            self.assertEqual(p[IP].src, self.pg3.local_ip4)
            self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
            self.assertEqual(p[UDP].sport, self.ipfix_src_port)
            self.assertEqual(p[UDP].dport, 4739)
            self.assertEqual(p[IPFIX].observationDomainID,
                             self.ipfix_domain_id)
            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 test_pool_addr_fib(self):
        """ S-NAT add pool addresses to FIB """
        static_addr = '10.0.0.10'
        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.snat_add_static_mapping(self.pg0.remote_ip4, static_addr)

        # SNAT address
        p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
             ARP(op=ARP.who_has, pdst=self.snat_addr,
                 psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
        self.pg1.add_stream(p)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg1.get_capture(1)
        self.assertTrue(capture[0].haslayer(ARP))
        self.assertTrue(capture[0][ARP].op, ARP.is_at)

        # 1:1 NAT address
        p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
             ARP(op=ARP.who_has, pdst=static_addr,
                 psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
        self.pg1.add_stream(p)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg1.get_capture(1)
        self.assertTrue(capture[0].haslayer(ARP))
        self.assertTrue(capture[0][ARP].op, ARP.is_at)

        # send ARP to non-SNAT interface
        p = (Ether(src=self.pg2.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
             ARP(op=ARP.who_has, pdst=self.snat_addr,
                 psrc=self.pg2.remote_ip4, hwsrc=self.pg2.remote_mac))
        self.pg2.add_stream(p)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg1.get_capture(0)

        # remove addresses and verify
        self.snat_add_address(self.snat_addr, is_add=0)
        self.snat_add_static_mapping(self.pg0.remote_ip4, static_addr,
                                     is_add=0)

        p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
             ARP(op=ARP.who_has, pdst=self.snat_addr,
                 psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
        self.pg1.add_stream(p)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg1.get_capture(0)

        p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
             ARP(op=ARP.who_has, pdst=static_addr,
                 psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
        self.pg1.add_stream(p)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg1.get_capture(0)

    def test_vrf_mode(self):
        """ S-NAT tenant VRF aware address pool mode """

        vrf_id1 = 1
        vrf_id2 = 2
        nat_ip1 = "10.0.0.10"
        nat_ip2 = "10.0.0.11"

        self.pg0.unconfig_ip4()
        self.pg1.unconfig_ip4()
        self.pg0.set_table_ip4(vrf_id1)
        self.pg1.set_table_ip4(vrf_id2)
        self.pg0.config_ip4()
        self.pg1.config_ip4()

        self.snat_add_address(nat_ip1, vrf_id=vrf_id1)
        self.snat_add_address(nat_ip2, vrf_id=vrf_id2)
        self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
        self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index)
        self.vapi.snat_interface_add_del_feature(self.pg2.sw_if_index,
                                                 is_inside=0)

        # first VRF
        pkts = self.create_stream_in(self.pg0, self.pg2)
        self.pg0.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg2.get_capture(len(pkts))
        self.verify_capture_out(capture, nat_ip1)

        # second VRF
        pkts = self.create_stream_in(self.pg1, self.pg2)
        self.pg1.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg2.get_capture(len(pkts))
        self.verify_capture_out(capture, nat_ip2)

    def test_vrf_feature_independent(self):
        """ S-NAT tenant VRF independent address pool mode """

        nat_ip1 = "10.0.0.10"
        nat_ip2 = "10.0.0.11"

        self.snat_add_address(nat_ip1)
        self.snat_add_address(nat_ip2)
        self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
        self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index)
        self.vapi.snat_interface_add_del_feature(self.pg2.sw_if_index,
                                                 is_inside=0)

        # first VRF
        pkts = self.create_stream_in(self.pg0, self.pg2)
        self.pg0.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg2.get_capture(len(pkts))
        self.verify_capture_out(capture, nat_ip1)

        # second VRF
        pkts = self.create_stream_in(self.pg1, self.pg2)
        self.pg1.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg2.get_capture(len(pkts))
        self.verify_capture_out(capture, nat_ip1)

    def test_dynamic_ipless_interfaces(self):
        """ SNAT interfaces without configured ip dynamic map """

        self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
                                      self.pg7.remote_mac,
                                      self.pg7.remote_ip4n,
                                      is_static=1)
        self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index,
                                      self.pg8.remote_mac,
                                      self.pg8.remote_ip4n,
                                      is_static=1)

        self.vapi.ip_add_del_route(dst_address=self.pg7.remote_ip4n,
                                   dst_address_length=32,
                                   next_hop_address=self.pg7.remote_ip4n,
                                   next_hop_sw_if_index=self.pg7.sw_if_index)
        self.vapi.ip_add_del_route(dst_address=self.pg8.remote_ip4n,
                                   dst_address_length=32,
                                   next_hop_address=self.pg8.remote_ip4n,
                                   next_hop_sw_if_index=self.pg8.sw_if_index)

        self.snat_add_address(self.snat_addr)
        self.vapi.snat_interface_add_del_feature(self.pg7.sw_if_index)
        self.vapi.snat_interface_add_del_feature(self.pg8.sw_if_index,
                                                 is_inside=0)

        # in2out
        pkts = self.create_stream_in(self.pg7, self.pg8)
        self.pg7.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg8.get_capture(len(pkts))
        self.verify_capture_out(capture)

        # out2in
        pkts = self.create_stream_out(self.pg8, self.snat_addr)
        self.pg8.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg7.get_capture(len(pkts))
        self.verify_capture_in(capture, self.pg7)

    def test_static_ipless_interfaces(self):
        """ SNAT 1:1 NAT interfaces without configured ip """

        self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
                                      self.pg7.remote_mac,
                                      self.pg7.remote_ip4n,
                                      is_static=1)
        self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index,
                                      self.pg8.remote_mac,
                                      self.pg8.remote_ip4n,
                                      is_static=1)

        self.vapi.ip_add_del_route(dst_address=self.pg7.remote_ip4n,
                                   dst_address_length=32,
                                   next_hop_address=self.pg7.remote_ip4n,
                                   next_hop_sw_if_index=self.pg7.sw_if_index)
        self.vapi.ip_add_del_route(dst_address=self.pg8.remote_ip4n,
                                   dst_address_length=32,
                                   next_hop_address=self.pg8.remote_ip4n,
                                   next_hop_sw_if_index=self.pg8.sw_if_index)

        self.snat_add_static_mapping(self.pg7.remote_ip4, self.snat_addr)
        self.vapi.snat_interface_add_del_feature(self.pg7.sw_if_index)
        self.vapi.snat_interface_add_del_feature(self.pg8.sw_if_index,
                                                 is_inside=0)

        # out2in
        pkts = self.create_stream_out(self.pg8)
        self.pg8.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg7.get_capture(len(pkts))
        self.verify_capture_in(capture, self.pg7)

        # in2out
        pkts = self.create_stream_in(self.pg7, self.pg8)
        self.pg7.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg8.get_capture(len(pkts))
        self.verify_capture_out(capture, self.snat_addr, True)

    def test_static_with_port_ipless_interfaces(self):
        """ SNAT 1:1 NAT with port interfaces without configured ip """

        self.tcp_port_out = 30606
        self.udp_port_out = 30607
        self.icmp_id_out = 30608

        self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
                                      self.pg7.remote_mac,
                                      self.pg7.remote_ip4n,
                                      is_static=1)
        self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index,
                                      self.pg8.remote_mac,
                                      self.pg8.remote_ip4n,
                                      is_static=1)

        self.vapi.ip_add_del_route(dst_address=self.pg7.remote_ip4n,
                                   dst_address_length=32,
                                   next_hop_address=self.pg7.remote_ip4n,
                                   next_hop_sw_if_index=self.pg7.sw_if_index)
        self.vapi.ip_add_del_route(dst_address=self.pg8.remote_ip4n,
                                   dst_address_length=32,
                                   next_hop_address=self.pg8.remote_ip4n,
                                   next_hop_sw_if_index=self.pg8.sw_if_index)

        self.snat_add_address(self.snat_addr)
        self.snat_add_static_mapping(self.pg7.remote_ip4, self.snat_addr,
                                     self.tcp_port_in, self.tcp_port_out,
                                     proto=IP_PROTOS.tcp)
        self.snat_add_static_mapping(self.pg7.remote_ip4, self.snat_addr,
                                     self.udp_port_in, self.udp_port_out,
                                     proto=IP_PROTOS.udp)
        self.snat_add_static_mapping(self.pg7.remote_ip4, self.snat_addr,
                                     self.icmp_id_in, self.icmp_id_out,
                                     proto=IP_PROTOS.icmp)
        self.vapi.snat_interface_add_del_feature(self.pg7.sw_if_index)
        self.vapi.snat_interface_add_del_feature(self.pg8.sw_if_index,
                                                 is_inside=0)

        # out2in
        pkts = self.create_stream_out(self.pg8)
        self.pg8.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg7.get_capture(len(pkts))
        self.verify_capture_in(capture, self.pg7)

        # in2out
        pkts = self.create_stream_in(self.pg7, self.pg8)
        self.pg7.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg8.get_capture(len(pkts))
        self.verify_capture_out(capture)

    def tearDown(self):
        super(TestSNAT, self).tearDown()
        if not self.vpp_dead:
            self.logger.info(self.vapi.cli("show snat verbose"))
            self.clear_snat()


class TestDeterministicNAT(MethodHolder):
    """ Deterministic NAT Test Cases """

    @classmethod
    def setUpConstants(cls):
        super(TestDeterministicNAT, cls).setUpConstants()
        cls.vpp_cmdline.extend(["snat", "{", "deterministic", "}"])

    @classmethod
    def setUpClass(cls):
        super(TestDeterministicNAT, cls).setUpClass()

        try:
            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.snat_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()

        except Exception:
            super(TestDeterministicNAT, cls).tearDownClass()
            raise

    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 SNAT address)
        :param ttl: TTL of generated packets
        """
        if dst_ip is None:
            dst_ip = self.snat_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, packet_num=3):
        """
        Verify captured packets on outside network

        :param capture: Captured packets
        :param nat_ip: Translated IP address (Default use global SNAT address)
        :param same_port: Sorce port number is not translated (Default False)
        :param packet_num: Expected number of packets (Default 3)
        """
        if nat_ip is None:
            nat_ip = self.snat_addr
        self.assertEqual(packet_num, len(capture))
        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 initiate_tcp_session(self, in_if, out_if):
        """
        Initiates TCP session

        :param in_if: Inside interface
        :param out_if: Outside interface
        """
        try:
            # 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.snat_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)

        except:
            self.logger.error("TCP 3 way handshake failed")
            raise

    def verify_ipfix_max_entries_per_user(self, data):
        """
        Verify IPFIX maximum entries per user exceeded event

        :param data: Decoded IPFIX data records
        """
        self.assertEqual(1, len(data))
        record = data[0]
        # natEvent
        self.assertEqual(ord(record[230]), 13)
        # natQuotaExceededEvent
        self.assertEqual('\x03\x00\x00\x00', record[466])
        # sourceIPv4Address
        self.assertEqual(self.pg0.remote_ip4n, record[8])

    def test_deterministic_mode(self):
        """ S-NAT run deterministic mode """
        in_addr = '172.16.255.0'
        out_addr = '172.17.255.50'
        in_addr_t = '172.16.255.20'
        in_addr_n = socket.inet_aton(in_addr)
        out_addr_n = socket.inet_aton(out_addr)
        in_addr_t_n = socket.inet_aton(in_addr_t)
        in_plen = 24
        out_plen = 32

        snat_config = self.vapi.snat_show_config()
        self.assertEqual(1, snat_config.deterministic)

        self.vapi.snat_add_det_map(in_addr_n, in_plen, out_addr_n, out_plen)

        rep1 = self.vapi.snat_det_forward(in_addr_t_n)
        self.assertEqual(rep1.out_addr[:4], out_addr_n)
        rep2 = self.vapi.snat_det_reverse(out_addr_n, rep1.out_port_hi)
        self.assertEqual(rep2.in_addr[:4], in_addr_t_n)

        deterministic_mappings = self.vapi.snat_det_map_dump()
        self.assertEqual(len(deterministic_mappings), 1)
        dsm = deterministic_mappings[0]
        self.assertEqual(in_addr_n, dsm.in_addr[:4])
        self.assertEqual(in_plen, dsm.in_plen)
        self.assertEqual(out_addr_n, dsm.out_addr[:4])
        self.assertEqual(out_plen, dsm.out_plen)

        self.clear_snat()
        deterministic_mappings = self.vapi.snat_det_map_dump()
        self.assertEqual(len(deterministic_mappings), 0)

    def test_set_timeouts(self):
        """ Set deterministic NAT timeouts """
        timeouts_before = self.vapi.snat_det_get_timeouts()

        self.vapi.snat_det_set_timeouts(timeouts_before.udp + 10,
                                        timeouts_before.tcp_established + 10,
                                        timeouts_before.tcp_transitory + 10,
                                        timeouts_before.icmp + 10)

        timeouts_after = self.vapi.snat_det_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_det_in(self):
        """ CGNAT translation test (TCP, UDP, ICMP) """

        nat_ip = "10.0.0.10"

        self.vapi.snat_add_det_map(self.pg0.remote_ip4n,
                                   32,
                                   socket.inet_aton(nat_ip),
                                   32)
        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)

        # 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.snat_det_session_dump(self.pg0.remote_ip4n)
        self.assertEqual(len(sessions), 3)

        # TCP session
        s = sessions[0]
        self.assertEqual(s.ext_addr[:4], self.pg1.remote_ip4n)
        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(s.ext_addr[:4], self.pg1.remote_ip4n)
        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(s.ext_addr[:4], self.pg1.remote_ip4n)
        self.assertEqual(s.in_port, self.icmp_id_in)
        self.assertEqual(s.out_port, self.icmp_external_id)

    def test_multiple_users(self):
        """ CGNAT 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.snat_add_det_map(host0.ip4n,
                                   24,
                                   socket.inet_aton(nat_ip),
                                   32)
        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)

        # 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.snat_det_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.snat_det_close_session_out(socket.inet_aton(nat_ip),
                                             port_out1,
                                             self.pg1.remote_ip4n,
                                             external_port)
        dms = self.vapi.snat_det_map_dump()
        self.assertEqual(dms[0].ses_num, 1)

        self.vapi.snat_det_close_session_in(host0.ip4n,
                                            port_in,
                                            self.pg1.remote_ip4n,
                                            external_port)
        dms = self.vapi.snat_det_map_dump()
        self.assertEqual(dms[0].ses_num, 0)

    def test_tcp_session_close_detection_in(self):
        """ CGNAT TCP session close initiated from inside network """
        self.vapi.snat_add_det_map(self.pg0.remote_ip4n,
                                   32,
                                   socket.inet_aton(self.snat_addr),
                                   32)
        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.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.snat_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.snat_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 snat closed the session
            dms = self.vapi.snat_det_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):
        """ CGNAT TCP session close initiated from outside network """
        self.vapi.snat_add_det_map(self.pg0.remote_ip4n,
                                   32,
                                   socket.inet_aton(self.snat_addr),
                                   32)
        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.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.snat_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.snat_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 snat closed the session
            dms = self.vapi.snat_det_map_dump()
            self.assertEqual(0, dms[0].ses_num)
        except:
            self.logger.error("TCP session termination failed")
            raise

    @unittest.skipUnless(running_extended_tests(), "part of extended tests")
    def test_session_timeout(self):
        """ CGNAT session timeouts """
        self.vapi.snat_add_det_map(self.pg0.remote_ip4n,
                                   32,
                                   socket.inet_aton(self.snat_addr),
                                   32)
        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.initiate_tcp_session(self.pg0, self.pg1)
        self.vapi.snat_det_set_timeouts(5, 5, 5, 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()
        capture = self.pg1.get_capture(len(pkts))
        sleep(15)

        dms = self.vapi.snat_det_map_dump()
        self.assertEqual(0, dms[0].ses_num)

    def test_session_limit_per_user(self):
        """ CGNAT maximum 1000 sessions per user should be created """
        self.vapi.snat_add_det_map(self.pg0.remote_ip4n,
                                   32,
                                   socket.inet_aton(self.snat_addr),
                                   32)
        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.pg2.remote_ip4n,
                                     src_address=self.pg2.local_ip4n,
                                     path_mtu=512,
                                     template_interval=10)
        self.vapi.snat_ipfix()

        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()
        capture = 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()
        capture = 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.snat_det_map_dump()

        self.assertEqual(1000, dms[0].ses_num)

        # verify IPFIX logging
        self.vapi.cli("ipfix flush")  # FIXME this should be an API call
        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)

    def clear_snat(self):
        """
        Clear SNAT configuration.
        """
        self.vapi.snat_ipfix(enable=0)
        self.vapi.snat_det_set_timeouts()
        deterministic_mappings = self.vapi.snat_det_map_dump()
        for dsm in deterministic_mappings:
            self.vapi.snat_add_det_map(dsm.in_addr,
                                       dsm.in_plen,
                                       dsm.out_addr,
                                       dsm.out_plen,
                                       is_add=0)

        interfaces = self.vapi.snat_interface_dump()
        for intf in interfaces:
            self.vapi.snat_interface_add_del_feature(intf.sw_if_index,
                                                     intf.is_inside,
                                                     is_add=0)

    def tearDown(self):
        super(TestDeterministicNAT, self).tearDown()
        if not self.vpp_dead:
            self.logger.info(self.vapi.cli("show snat detail"))
            self.clear_snat()


class TestNAT64(MethodHolder):
    """ NAT64 Test Cases """

    @classmethod
    def setUpClass(cls):
        super(TestNAT64, cls).setUpClass()

        try:
            cls.tcp_port_in = 6303
            cls.tcp_port_out = 6303
            cls.udp_port_in = 6304
            cls.udp_port_out = 6304
            cls.icmp_id_in = 6305
            cls.icmp_id_out = 6305
            cls.nat_addr = '10.0.0.3'
            cls.nat_addr_n = socket.inet_pton(socket.AF_INET, cls.nat_addr)

            cls.create_pg_interfaces(range(2))
            cls.ip6_interfaces = list(cls.pg_interfaces[0:1])
            cls.ip4_interfaces = list(cls.pg_interfaces[1:2])

            for i in cls.ip6_interfaces:
                i.admin_up()
                i.config_ip6()
                i.resolve_ndp()

            for i in cls.ip4_interfaces:
                i.admin_up()
                i.config_ip4()
                i.resolve_arp()

        except Exception:
            super(TestNAT64, cls).tearDownClass()
            raise

    def test_pool(self):
        """ Add/delete address to NAT64 pool """
        nat_addr = socket.inet_pton(socket.AF_INET, '1.2.3.4')

        self.vapi.nat64_add_del_pool_addr_range(nat_addr, nat_addr)

        addresses = self.vapi.nat64_pool_addr_dump()
        self.assertEqual(len(addresses), 1)
        self.assertEqual(addresses[0].address, nat_addr)

        self.vapi.nat64_add_del_pool_addr_range(nat_addr, nat_addr, is_add=0)

        addresses = self.vapi.nat64_pool_addr_dump()
        self.assertEqual(len(addresses), 0)

    def test_interface(self):
        """ Enable/disable NAT64 feature on the interface """
        self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
        self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)

        interfaces = self.vapi.nat64_interface_dump()
        self.assertEqual(len(interfaces), 2)
        pg0_found = False
        pg1_found = False
        for intf in interfaces:
            if intf.sw_if_index == self.pg0.sw_if_index:
                self.assertEqual(intf.is_inside, 1)
                pg0_found = True
            elif intf.sw_if_index == self.pg1.sw_if_index:
                self.assertEqual(intf.is_inside, 0)
                pg1_found = True
        self.assertTrue(pg0_found)
        self.assertTrue(pg1_found)

        features = self.vapi.cli("show interface features pg0")
        self.assertNotEqual(features.find('nat64-in2out'), -1)
        features = self.vapi.cli("show interface features pg1")
        self.assertNotEqual(features.find('nat64-out2in'), -1)

        self.vapi.nat64_add_del_interface(self.pg0.sw_if_index, is_add=0)
        self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_add=0)

        interfaces = self.vapi.nat64_interface_dump()
        self.assertEqual(len(interfaces), 0)

    def test_static_bib(self):
        """ Add/delete static BIB entry """
        in_addr = socket.inet_pton(socket.AF_INET6,
                                   '2001:db8:85a3::8a2e:370:7334')
        out_addr = socket.inet_pton(socket.AF_INET, '10.1.1.3')
        in_port = 1234
        out_port = 5678
        proto = IP_PROTOS.tcp

        self.vapi.nat64_add_del_static_bib(in_addr,
                                           out_addr,
                                           in_port,
                                           out_port,
                                           proto)
        bib = self.vapi.nat64_bib_dump(IP_PROTOS.tcp)
        static_bib_num = 0
        for bibe in bib:
            if bibe.is_static:
                static_bib_num += 1
                self.assertEqual(bibe.i_addr, in_addr)
                self.assertEqual(bibe.o_addr, out_addr)
                self.assertEqual(bibe.i_port, in_port)
                self.assertEqual(bibe.o_port, out_port)
        self.assertEqual(static_bib_num, 1)

        self.vapi.nat64_add_del_static_bib(in_addr,
                                           out_addr,
                                           in_port,
                                           out_port,
                                           proto,
                                           is_add=0)
        bib = self.vapi.nat64_bib_dump(IP_PROTOS.tcp)
        static_bib_num = 0
        for bibe in bib:
            if bibe.is_static:
                static_bib_num += 1
        self.assertEqual(static_bib_num, 0)

    def test_set_timeouts(self):
        """ Set NAT64 timeouts """
        # verify default values
        timeouts = self.vapi.nat64_get_timeouts()
        self.assertEqual(timeouts.udp, 300)
        self.assertEqual(timeouts.icmp, 60)
        self.assertEqual(timeouts.tcp_trans, 240)
        self.assertEqual(timeouts.tcp_est, 7440)
        self.assertEqual(timeouts.tcp_incoming_syn, 6)

        # set and verify custom values
        self.vapi.nat64_set_timeouts(udp=200, icmp=30, tcp_trans=250,
                                     tcp_est=7450, tcp_incoming_syn=10)
        timeouts = self.vapi.nat64_get_timeouts()
        self.assertEqual(timeouts.udp, 200)
        self.assertEqual(timeouts.icmp, 30)
        self.assertEqual(timeouts.tcp_trans, 250)
        self.assertEqual(timeouts.tcp_est, 7450)
        self.assertEqual(timeouts.tcp_incoming_syn, 10)

    def test_dynamic(self):
        """ NAT64 dynamic translation test """
        self.tcp_port_in = 6303
        self.udp_port_in = 6304
        self.icmp_id_in = 6305

        ses_num_start = self.nat64_get_ses_num()

        self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
                                                self.nat_addr_n)
        self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
        self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)

        # in2out
        pkts = self.create_stream_in_ip6(self.pg0, self.pg1)
        self.pg0.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg1.get_capture(3)
        self.verify_capture_out(capture, packet_num=3, nat_ip=self.nat_addr,
                                dst_ip=self.pg1.remote_ip4)

        # out2in
        pkts = self.create_stream_out(self.pg1, dst_ip=self.nat_addr)
        self.pg1.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg0.get_capture(3)
        ip = IPv6(src=''.join(['64:ff9b::', self.pg1.remote_ip4]))
        self.verify_capture_in_ip6(capture, ip[IPv6].src, self.pg0.remote_ip6)

        # in2out
        pkts = self.create_stream_in_ip6(self.pg0, self.pg1)
        self.pg0.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg1.get_capture(3)
        self.verify_capture_out(capture, packet_num=3, nat_ip=self.nat_addr,
                                dst_ip=self.pg1.remote_ip4)

        # out2in
        pkts = self.create_stream_out(self.pg1, dst_ip=self.nat_addr)
        self.pg1.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg0.get_capture(3)
        ip = IPv6(src=''.join(['64:ff9b::', self.pg1.remote_ip4]))
        self.verify_capture_in_ip6(capture, ip[IPv6].src, self.pg0.remote_ip6)

        ses_num_end = self.nat64_get_ses_num()

        self.assertEqual(ses_num_end - ses_num_start, 3)

    def test_static(self):
        """ NAT64 static translation test """
        self.tcp_port_in = 60303
        self.udp_port_in = 60304
        self.icmp_id_in = 60305
        self.tcp_port_out = 60303
        self.udp_port_out = 60304
        self.icmp_id_out = 60305

        ses_num_start = self.nat64_get_ses_num()

        self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
                                                self.nat_addr_n)
        self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
        self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)

        self.vapi.nat64_add_del_static_bib(self.pg0.remote_ip6n,
                                           self.nat_addr_n,
                                           self.tcp_port_in,
                                           self.tcp_port_out,
                                           IP_PROTOS.tcp)
        self.vapi.nat64_add_del_static_bib(self.pg0.remote_ip6n,
                                           self.nat_addr_n,
                                           self.udp_port_in,
                                           self.udp_port_out,
                                           IP_PROTOS.udp)
        self.vapi.nat64_add_del_static_bib(self.pg0.remote_ip6n,
                                           self.nat_addr_n,
                                           self.icmp_id_in,
                                           self.icmp_id_out,
                                           IP_PROTOS.icmp)

        # in2out
        pkts = self.create_stream_in_ip6(self.pg0, self.pg1)
        self.pg0.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg1.get_capture(3)
        self.verify_capture_out(capture, packet_num=3, nat_ip=self.nat_addr,
                                dst_ip=self.pg1.remote_ip4, same_port=True)

        # out2in
        pkts = self.create_stream_out(self.pg1, dst_ip=self.nat_addr)
        self.pg1.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg0.get_capture(3)
        ip = IPv6(src=''.join(['64:ff9b::', self.pg1.remote_ip4]))
        self.verify_capture_in_ip6(capture, ip[IPv6].src, self.pg0.remote_ip6)

        ses_num_end = self.nat64_get_ses_num()

        self.assertEqual(ses_num_end - ses_num_start, 3)

    @unittest.skipUnless(running_extended_tests(), "part of extended tests")
    def test_session_timeout(self):
        """ NAT64 session timeout """
        self.icmp_id_in = 1234
        self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
                                                self.nat_addr_n)
        self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
        self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
        self.vapi.nat64_set_timeouts(icmp=5)

        pkts = self.create_stream_in_ip6(self.pg0, self.pg1)
        self.pg0.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg1.get_capture(3)

        ses_num_before_timeout = self.nat64_get_ses_num()

        sleep(15)

        # ICMP session after timeout
        ses_num_after_timeout = self.nat64_get_ses_num()
        self.assertNotEqual(ses_num_before_timeout, ses_num_after_timeout)

    def nat64_get_ses_num(self):
        """
        Return number of active NAT64 sessions.
        """
        ses_num = 0
        st = self.vapi.nat64_st_dump(IP_PROTOS.tcp)
        ses_num += len(st)
        st = self.vapi.nat64_st_dump(IP_PROTOS.udp)
        ses_num += len(st)
        st = self.vapi.nat64_st_dump(IP_PROTOS.icmp)
        ses_num += len(st)
        return ses_num

    def clear_nat64(self):
        """
        Clear NAT64 configuration.
        """
        self.vapi.nat64_set_timeouts()

        interfaces = self.vapi.nat64_interface_dump()
        for intf in interfaces:
            self.vapi.nat64_add_del_interface(intf.sw_if_index,
                                              intf.is_inside,
                                              is_add=0)

        bib = self.vapi.nat64_bib_dump(IP_PROTOS.tcp)
        for bibe in bib:
            if bibe.is_static:
                self.vapi.nat64_add_del_static_bib(bibe.i_addr,
                                                   bibe.o_addr,
                                                   bibe.i_port,
                                                   bibe.o_port,
                                                   bibe.proto,
                                                   bibe.vrf_id,
                                                   is_add=0)

        bib = self.vapi.nat64_bib_dump(IP_PROTOS.udp)
        for bibe in bib:
            if bibe.is_static:
                self.vapi.nat64_add_del_static_bib(bibe.i_addr,
                                                   bibe.o_addr,
                                                   bibe.i_port,
                                                   bibe.o_port,
                                                   bibe.proto,
                                                   bibe.vrf_id,
                                                   is_add=0)

        bib = self.vapi.nat64_bib_dump(IP_PROTOS.icmp)
        for bibe in bib:
            if bibe.is_static:
                self.vapi.nat64_add_del_static_bib(bibe.i_addr,
                                                   bibe.o_addr,
                                                   bibe.i_port,
                                                   bibe.o_port,
                                                   bibe.proto,
                                                   bibe.vrf_id,
                                                   is_add=0)

        adresses = self.vapi.nat64_pool_addr_dump()
        for addr in adresses:
            self.vapi.nat64_add_del_pool_addr_range(addr.address,
                                                    addr.address,
                                                    is_add=0)

    def tearDown(self):
        super(TestNAT64, self).tearDown()
        if not self.vpp_dead:
            self.logger.info(self.vapi.cli("show nat64 pool"))
            self.logger.info(self.vapi.cli("show nat64 interfaces"))
            self.logger.info(self.vapi.cli("show nat64 bib tcp"))
            self.logger.info(self.vapi.cli("show nat64 bib udp"))
            self.logger.info(self.vapi.cli("show nat64 bib icmp"))
            self.logger.info(self.vapi.cli("show nat64 session table tcp"))
            self.logger.info(self.vapi.cli("show nat64 session table udp"))
            self.logger.info(self.vapi.cli("show nat64 session table icmp"))
            self.clear_nat64()

if __name__ == '__main__':
    unittest.main(testRunner=VppTestRunner)