Install SNAT rules for ipv4 only

Change-Id: I37bd770aa0d54a985ac2bec708c571785084e0ec
Closes-Bug: #1309195
(cherry picked from commit d23bc8fa6e)

neutron/agent/l3_agent.py

@@ -399,7 +399,9 @@ class L3NATAgent(firewall_l3_agent.FWaaSL3AgentRpcCallback, manager.Manager):
             ri.internal_ports.remove(p)
             self.internal_network_removed(ri, p['id'], p['ip_cidr'])
 
-        internal_cidrs = [p['ip_cidr'] for p in ri.internal_ports]
+        # Get IPv4 only internal CIDRs
+        internal_cidrs = [p['ip_cidr'] for p in ri.internal_ports
+                          if netaddr.IPNetwork(p['ip_cidr']).version == 4]
         # TODO(salv-orlando): RouterInfo would be a better place for
         # this logic too
         ex_gw_port_id = (ex_gw_port and ex_gw_port['id'] or
@@ -444,11 +446,16 @@ class L3NATAgent(firewall_l3_agent.FWaaSL3AgentRpcCallback, manager.Manager):
         # And add them back if the action if add_rules
         if action == 'add_rules' and ex_gw_port:
             # ex_gw_port should not be None in this case
-            ex_gw_ip = ex_gw_port['fixed_ips'][0]['ip_address']
+            # NAT rules are added only if ex_gw_port has an IPv4 address
-            for rule in self.external_gateway_nat_rules(ex_gw_ip,
+            for ip_addr in ex_gw_port['fixed_ips']:
-                                                        internal_cidrs,
+                ex_gw_ip = ip_addr['ip_address']
-                                                        interface_name):
+                if netaddr.IPAddress(ex_gw_ip).version == 4:
-                ri.iptables_manager.ipv4['nat'].add_rule(*rule)
+                    rules = self.external_gateway_nat_rules(ex_gw_ip,
+                                                            internal_cidrs,
+                                                            interface_name)
+                    for rule in rules:
+                        ri.iptables_manager.ipv4['nat'].add_rule(*rule)
+                    break
         ri.iptables_manager.apply()
 
     def process_router_floating_ips(self, ri, ex_gw_port):

neutron/tests/unit/test_l3_agent.py

@@ -18,6 +18,7 @@
 import copy
 
 import mock
+import netaddr
 from oslo.config import cfg
 
 from neutron.agent.common import config as agent_config
@@ -303,24 +304,37 @@ class TestBasicRouterOperations(base.BaseTestCase):
             '! -i %s ! -o %s -m conntrack ! --ctstate DNAT -j ACCEPT' %
             (interface_name, interface_name)]
         for source_cidr in source_cidrs:
-            value_dict = {'source_cidr': source_cidr,
+            # Create SNAT rules for IPv4 only
-                          'source_nat_ip': source_nat_ip}
+            if (netaddr.IPNetwork(source_cidr).version == 4 and
-            expected_rules.append('-s %(source_cidr)s -j SNAT --to-source '
+                netaddr.IPNetwork(source_nat_ip).version == 4):
-                                  '%(source_nat_ip)s' % value_dict)
+                value_dict = {'source_cidr': source_cidr,
+                              'source_nat_ip': source_nat_ip}
+                expected_rules.append('-s %(source_cidr)s -j SNAT --to-source '
+                                      '%(source_nat_ip)s' % value_dict)
         for r in rules:
             if negate:
                 self.assertNotIn(r.rule, expected_rules)
             else:
                 self.assertIn(r.rule, expected_rules)
 
-    def _prepare_router_data(self, enable_snat=None, num_internal_ports=1):
+    def _prepare_router_data(self, ip_version=4,
+                             enable_snat=None, num_internal_ports=1):
+        if ip_version == 4:
+            ip_addr = '19.4.4.4'
+            cidr = '19.4.4.0/24'
+            gateway_ip = '19.4.4.1'
+        elif ip_version == 6:
+            ip_addr = 'fd00::4'
+            cidr = 'fd00::/64'
+            gateway_ip = 'fd00::1'
+
         router_id = _uuid()
         ex_gw_port = {'id': _uuid(),
                       'network_id': _uuid(),
-                      'fixed_ips': [{'ip_address': '19.4.4.4',
+                      'fixed_ips': [{'ip_address': ip_addr,
                                      'subnet_id': _uuid()}],
-                      'subnet': {'cidr': '19.4.4.0/24',
+                      'subnet': {'cidr': cidr,
-                                 'gateway_ip': '19.4.4.1'}}
+                                 'gateway_ip': gateway_ip}}
         int_ports = []
         for i in range(0, num_internal_ports):
             int_ports.append({'id': _uuid(),
@@ -525,6 +539,99 @@ class TestBasicRouterOperations(base.BaseTestCase):
         self.assertEqual(len(nat_rules_delta), 1)
         self._verify_snat_rules(nat_rules_delta, router)
 
+    def test_process_ipv6_only_gw(self):
+        agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
+        router = self._prepare_router_data(ip_version=6)
+        # Get NAT rules without the gw_port
+        gw_port = router['gw_port']
+        router['gw_port'] = None
+        ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
+                                 self.conf.use_namespaces, router=router)
+        agent.external_gateway_added = mock.Mock()
+        agent.process_router(ri)
+        orig_nat_rules = ri.iptables_manager.ipv4['nat'].rules[:]
+
+        # Get NAT rules with the gw_port
+        router['gw_port'] = gw_port
+        ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
+                                 self.conf.use_namespaces, router=router)
+        with mock.patch.object(
+                agent,
+                'external_gateway_nat_rules') as external_gateway_nat_rules:
+            agent.process_router(ri)
+            new_nat_rules = ri.iptables_manager.ipv4['nat'].rules[:]
+
+            # There should be no change with the NAT rules
+            self.assertFalse(external_gateway_nat_rules.called)
+            self.assertEqual(orig_nat_rules, new_nat_rules)
+
+    def test_process_router_ipv6_interface_added(self):
+        agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
+        router = self._prepare_router_data()
+        ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
+                                 self.conf.use_namespaces, router=router)
+        agent.external_gateway_added = mock.Mock()
+        # Process with NAT
+        agent.process_router(ri)
+        orig_nat_rules = ri.iptables_manager.ipv4['nat'].rules[:]
+        # Add an IPv6 interface and reprocess
+        router[l3_constants.INTERFACE_KEY].append(
+            {'id': _uuid(),
+             'network_id': _uuid(),
+             'admin_state_up': True,
+             'fixed_ips': [{'ip_address': 'fd00::2',
+                            'subnet_id': _uuid()}],
+             'mac_address': 'ca:fe:de:ad:be:ef',
+             'subnet': {'cidr': 'fd00::/64',
+                        'gateway_ip': 'fd00::1'}})
+        # Reassign the router object to RouterInfo
+        ri.router = router
+        agent.process_router(ri)
+        # For some reason set logic does not work well with
+        # IpTablesRule instances
+        nat_rules_delta = [r for r in ri.iptables_manager.ipv4['nat'].rules
+                           if r not in orig_nat_rules]
+        self.assertFalse(nat_rules_delta)
+
+    def test_process_router_ipv6v4_interface_added(self):
+        agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
+        router = self._prepare_router_data()
+        ri = l3_agent.RouterInfo(router['id'], self.conf.root_helper,
+                                 self.conf.use_namespaces, router=router)
+        agent.external_gateway_added = mock.Mock()
+        # Process with NAT
+        agent.process_router(ri)
+        orig_nat_rules = ri.iptables_manager.ipv4['nat'].rules[:]
+        # Add an IPv4 and IPv6 interface and reprocess
+        router[l3_constants.INTERFACE_KEY].append(
+            {'id': _uuid(),
+             'network_id': _uuid(),
+             'admin_state_up': True,
+             'fixed_ips': [{'ip_address': '35.4.1.4',
+                            'subnet_id': _uuid()}],
+             'mac_address': 'ca:fe:de:ad:be:ef',
+             'subnet': {'cidr': '35.4.1.0/24',
+                        'gateway_ip': '35.4.1.1'}})
+
+        router[l3_constants.INTERFACE_KEY].append(
+            {'id': _uuid(),
+             'network_id': _uuid(),
+             'admin_state_up': True,
+             'fixed_ips': [{'ip_address': 'fd00::2',
+                            'subnet_id': _uuid()}],
+             'mac_address': 'ca:fe:de:ad:be:ef',
+             'subnet': {'cidr': 'fd00::/64',
+                        'gateway_ip': 'fd00::1'}})
+        # Reassign the router object to RouterInfo
+        ri.router = router
+        agent.process_router(ri)
+        # For some reason set logic does not work well with
+        # IpTablesRule instances
+        nat_rules_delta = [r for r in ri.iptables_manager.ipv4['nat'].rules
+                           if r not in orig_nat_rules]
+        self.assertEqual(1, len(nat_rules_delta))
+        self._verify_snat_rules(nat_rules_delta, router)
+
     def test_process_router_interface_removed(self):
         agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
         router = self._prepare_router_data(num_internal_ports=2)