Merge "Require admin context for interfaces on ext network" into milestone-proposed

3 files changed, 41 insertions, 6 deletions

diff --git a/nova/exception.py b/nova/exception.py
index cce912549b..e8749272aa 100644
--- a/nova/exception.py
+++ b/nova/exception.py
@@ -625,6 +625,11 @@ class NetworkRequiresSubnet(Invalid):
                 " instances on.")
 
 
+class ExternalNetworkAttachForbidden(NotAuthorized):
+    msg_fmt = _("It is not allowed to create an interface on "
+                "external network %(network_uuid)s")
+
+
 class DatastoreNotFound(NotFound):
     msg_fmt = _("Could not find the datastore reference(s) which the VM uses.")
 
diff --git a/nova/network/neutronv2/api.py b/nova/network/neutronv2/api.py
index c9ab99ffff..752ba4bb98 100644
--- a/nova/network/neutronv2/api.py
+++ b/nova/network/neutronv2/api.py
@@ -136,6 +136,15 @@ class API(base.Base):
             nets,
             net_ids)
 
+        if not context.is_admin:
+            for net in nets:
+                # Perform this check here rather than in validate_networks to
+                # ensure the check is performed everytime allocate_for_instance
+                # is invoked
+                if net.get('router:external'):
+                    raise exception.ExternalNetworkAttachForbidden(
+                        network_uuid=net['id'])
+
         return nets
 
     def _create_port(self, port_client, instance, network_id, port_req_body,
diff --git a/nova/tests/network/test_neutronv2.py b/nova/tests/network/test_neutronv2.py
index 767d4fd868..9859aa330f 100644
--- a/nova/tests/network/test_neutronv2.py
+++ b/nova/tests/network/test_neutronv2.py
@@ -180,8 +180,13 @@ class TestNeutronv2Base(test.TestCase):
         self.nets4 = [{'id': 'his_netid4',
                       'name': 'his_netname4',
                       'tenant_id': 'his_tenantid'}]
-
-        self.nets = [self.nets1, self.nets2, self.nets3, self.nets4]
+        # A network request with external networks
+        self.nets5 = self.nets1 + [{'id': 'the-external-one',
+                                    'name': 'out-of-this-world',
+                                    'router:external': True,
+                                    'tenant_id': 'should-be-an-admin'}]
+        self.nets = [self.nets1, self.nets2, self.nets3,
+                     self.nets4, self.nets5]
 
         self.port_address = '10.0.1.2'
         self.port_data1 = [{'network_id': 'my_netid1',
@@ -1415,7 +1420,8 @@ class TestNeutronv2(TestNeutronv2Base):
                           api.get_fixed_ip_by_address,
                           self.context, address)
 
-    def _get_available_networks(self, prv_nets, pub_nets, req_ids=None):
+    def _get_available_networks(self, prv_nets, pub_nets,
+                                req_ids=None, context=None):
         api = neutronapi.API()
         nets = prv_nets + pub_nets
         if req_ids:
@@ -1432,9 +1438,10 @@ class TestNeutronv2(TestNeutronv2Base):
                 **mox_list_params).AndReturn({'networks': pub_nets})
 
         self.mox.ReplayAll()
-        rets = api._get_available_networks(self.context,
-                                           self.instance['project_id'],
-                                           req_ids)
+        rets = api._get_available_networks(
+            context if context else self.context,
+            self.instance['project_id'],
+            req_ids)
         self.assertEqual(rets, nets)
 
     def test_get_available_networks_all_private(self):
@@ -1453,6 +1460,20 @@ class TestNeutronv2(TestNeutronv2Base):
         req_ids = [net['id'] for net in (self.nets3[0], self.nets3[-1])]
         self._get_available_networks(prv_nets, pub_nets, req_ids)
 
+    def test_get_available_networks_with_externalnet_fails(self):
+        req_ids = [net['id'] for net in self.nets5]
+        self.assertRaises(
+            exception.ExternalNetworkAttachForbidden,
+            self._get_available_networks,
+            self.nets5, pub_nets=[], req_ids=req_ids)
+
+    def test_get_available_networks_with_externalnet_admin_ctx(self):
+        admin_ctx = context.RequestContext('userid', 'my_tenantid',
+                                           is_admin=True)
+        req_ids = [net['id'] for net in self.nets5]
+        self._get_available_networks(self.nets5, pub_nets=[],
+                                     req_ids=req_ids, context=admin_ctx)
+
     def test_get_floating_ip_pools(self):
         api = neutronapi.API()
         search_opts = {'router:external': True}