nat: in2out-output nodes work with acl reflect

Type: feature

The current feature ordering of NAT44 nodes with respect to the
ACL plugin's IPv4 input/output features is:

ip4-output: acl-plugin-out-ip4-fa runs before any NAT44 nodes
ip4-unicast: acl-plugin-in-ip4-fa runs before any NAT44 nodes

ACL rules with action permit+reflect can keep track of outbound
flows and allow the replies inbound without an explicit inbound rule.
If ACL permit+reflect rules are configured on an interface that also
has NAT44 configured with output-feature/postrouting translation of
outbound packets, the ACL rules cannot allow inbound packets. The
ACL state that was stored on the outbound flow contains the IP
addresses of the original packet, prior to translation. The inbound
packets are being evaluated by the ACL node using the translated
addresses.

The order of processing inbound needs to be the opposite of what it
was outbound for this to work. Change the NAT44 features on
ip4-output so that they run before outbound ACL nodes. This matches
the existing behavior of the NAT44 nodes which rewrite
source addresses as an input feature instead of an output feature.

This was only done for endpoint dependent mode because the regular
endpoint independent in2out-output node currently selects an
explicit next node rather than using the next node on the feature
arc.

Unit test added to configure both NAT and an ACL and ensure that
out2in packets matching an in2out flow are permitted by the ACL
and translated by NAT.

Change-Id: Ibd679c28b64c3fc3cc8c0606ea93123e384e839f
Signed-off-by: Matthew Smith <mgsmith@netgate.com>

1 files changed, 103 insertions, 0 deletions

diff --git a/src/plugins/nat/test/test_nat.py b/src/plugins/nat/test/test_nat.py
index 46b97c05dbe..f87065598f1 100644
--- a/src/plugins/nat/test/test_nat.py
+++ b/src/plugins/nat/test/test_nat.py
@@ -32,6 +32,7 @@ from scapy.all import bind_layers, Packet, ByteEnumField, ShortField, \
     PacketListField
 from ipaddress import IPv6Network
 from util import ppc, ppp
+from socket import inet_pton, AF_INET
 
 
 # NAT HA protocol event data
@@ -6386,6 +6387,108 @@ class TestNAT44EndpointDependent(MethodHolder):
         capture = self.pg0.get_capture(len(pkts))
         self.verify_capture_in(capture, self.pg0)
 
+    def test_output_feature_stateful_acl(self):
+        """ NAT44 endpoint-dependent output feature works with stateful ACL """
+        self.nat44_add_address(self.nat_addr)
+        self.vapi.nat44_interface_add_del_output_feature(
+            sw_if_index=self.pg0.sw_if_index,
+            flags=self.config_flags.NAT_IS_INSIDE,
+            is_add=1)
+        self.vapi.nat44_interface_add_del_output_feature(
+            sw_if_index=self.pg1.sw_if_index,
+            flags=self.config_flags.NAT_IS_OUTSIDE,
+            is_add=1)
+
+        # First ensure that the NAT is working sans ACL
+
+        # send packets out2in, no sessions yet so packets should drop
+        pkts_out2in = self.create_stream_out(self.pg1)
+        self.send_and_assert_no_replies(self.pg1, pkts_out2in)
+
+        # send packets into inside intf, ensure received via outside intf
+        pkts_in2out = self.create_stream_in(self.pg0, self.pg1)
+        capture = self.send_and_expect(self.pg0, pkts_in2out, self.pg1,
+                                       len(pkts_in2out))
+        self.verify_capture_out(capture)
+
+        # send out2in again, with sessions created it should work now
+        pkts_out2in = self.create_stream_out(self.pg1)
+        capture = self.send_and_expect(self.pg1, pkts_out2in, self.pg0,
+                                       len(pkts_out2in))
+        self.verify_capture_in(capture, self.pg0)
+
+        # Create an ACL blocking everything
+        out2in_deny_rule = {
+            'is_permit': 0,
+            'is_ipv6': 0,
+            'src_ip_addr': inet_pton(AF_INET, "0.0.0.0"),
+            'src_ip_prefix_len': 0,
+            'dst_ip_addr':  inet_pton(AF_INET, "0.0.0.0"),
+            'dst_ip_prefix_len': 0,
+            'srcport_or_icmptype_first': 0,
+            'srcport_or_icmptype_last': 65535,
+            'dstport_or_icmpcode_first': 0,
+            'dstport_or_icmpcode_last': 65535,
+            'proto': 0,
+        }
+        out2in_rules = [out2in_deny_rule]
+        res = self.vapi.acl_add_replace(0xffffffff, out2in_rules)
+        self.assertEqual(res.retval, 0, "error adding out2in ACL")
+        out2in_acl = res.acl_index
+
+        # apply as input acl on interface and confirm it blocks everything
+        self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg1.sw_if_index,
+                                             n_input=1,
+                                             acls=[out2in_acl])
+        self.send_and_assert_no_replies(self.pg1, pkts_out2in)
+
+        # create an ACL to permit/reflect everything
+        in2out_reflect_rule = {
+            'is_permit': 2,
+            'is_ipv6': 0,
+            'src_ip_addr': inet_pton(AF_INET, "0.0.0.0"),
+            'src_ip_prefix_len': 0,
+            'dst_ip_addr':  inet_pton(AF_INET, "0.0.0.0"),
+            'dst_ip_prefix_len': 0,
+            'srcport_or_icmptype_first': 0,
+            'srcport_or_icmptype_last': 65535,
+            'dstport_or_icmpcode_first': 0,
+            'dstport_or_icmpcode_last': 65535,
+            'proto': 0,
+        }
+        in2out_rules = [in2out_reflect_rule]
+        res = self.vapi.acl_add_replace(0xffffffff, in2out_rules)
+        self.assertEqual(res.retval, 0, "error adding in2out ACL")
+        in2out_acl = res.acl_index
+
+        # apply output acl
+        self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg1.sw_if_index,
+                                             n_input=1,
+                                             acls=[out2in_acl, in2out_acl])
+        # send in2out to generate ACL state (NAT state was created earlier)
+        capture = self.send_and_expect(self.pg0, pkts_in2out, self.pg1,
+                                       len(pkts_in2out))
+        self.verify_capture_out(capture)
+
+        # send out2in again. ACL state exists so it should work now.
+        # TCP packets with the syn flag set also need the ack flag
+        for p in pkts_out2in:
+            if p.haslayer(TCP) and p[TCP].flags & 0x02:
+                p[TCP].flags |= 0x10
+        capture = self.send_and_expect(self.pg1, pkts_out2in, self.pg0,
+                                       len(pkts_out2in))
+        self.verify_capture_in(capture, self.pg0)
+        self.logger.info(self.vapi.cli("show trace"))
+
+        # Clean up
+        # Remove ACLs from interface
+        self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg1.sw_if_index,
+                                             n_input=0,
+                                             acls=[])
+        # delete ACLs
+        self.vapi.acl_del(acl_index=out2in_acl, expected_retval=0)
+        self.vapi.acl_del(acl_index=in2out_acl, expected_retval=0)
+
     def test_multiple_vrf(self):
         """ Multiple VRF setup """
         external_addr = '1.2.3.4'