Validate scheme used in urlopen

An invalid url scheme could be used in urlopen in the python k8s
swagger client to violate security.  Validate that the URL is either
http or https and turn off noise from bandit.

Change-Id: I15fe8f9953e526beb25d84895ed3925a758ccda4
Implements: blueprint gate-bandit

magnum/common/exception.py

@@ -427,3 +427,7 @@ class BayTypeNotEnabled(MagnumException):
 
 class RequiredParameterNotProvided(MagnumException):
     message = _("Required parameter %(heat_param)s not provided.")
+
+
+class Urllib2InvalidScheme(MagnumException):
+    message = _("The urllib2 URL %(url) has an invalid scheme.")

magnum/common/pythonk8sclient/client/swagger.py

@@ -17,6 +17,7 @@ import mimetypes
 import random
 import string
 
+from magnum import utils
 from models import *
 
 
@@ -94,11 +95,13 @@ class ApiClient(object):
     else:
       raise Exception('Method ' + method + ' is not recognized.')
 
+    utils.raise_exception_invalid_scheme(url)
+
     request = MethodRequest(method=method, url=url, headers=headers,
                             data=data)
 
     # Make the request
-    response = urllib2.urlopen(request)
+    response = urllib2.urlopen(request) #nosec
     if 'Set-Cookie' in response.headers:
       self.cookie = response.headers['Set-Cookie']
     string = response.read()

magnum/common/utils.py

@@ -571,3 +571,14 @@ def allow_logical_names():
     except AttributeError:
         pass
     return True
+
+
+def raise_exception_invalid_scheme(url):
+    valid_schemes = ['http', 'https']
+
+    if not isinstance(url, basestring):
+        raise exception.Urllib2InvalidScheme(url=url)
+
+    scheme = url.split(':')[0]
+    if scheme not in valid_schemes:
+        raise exception.Urllib2InvalidScheme(url=url)

magnum/tests/unit/common/test_utils.py

@@ -499,3 +499,35 @@ class TempFilesTestCase(base.TestCase):
 
         rmtree_mock.assert_called_once_with(tempdir_created)
         self.assertTrue(log_mock.error.called)
+
+
+class Urllib2_invalid_scheme(base.TestCase):
+    def test_raise_exception_invalid_scheme_file(self):
+        self.assertRaises(
+            exception.Urllib2InvalidScheme,
+            utils.raise_exception_invalid_scheme,
+            'file:///etc/passwd')
+
+    def test_raise_exception_invalid_scheme_starting_colon(self):
+        self.assertRaises(
+            exception.Urllib2InvalidScheme,
+            utils.raise_exception_invalid_scheme,
+            ':///etc/passwd')
+
+    def test_raise_exception_invalid_scheme_None(self):
+        self.assertRaises(
+            exception.Urllib2InvalidScheme,
+            utils.raise_exception_invalid_scheme,
+            None)
+
+    def test_raise_exception_invalid_scheme_empty_string(self):
+        self.assertRaises(
+            exception.Urllib2InvalidScheme,
+            utils.raise_exception_invalid_scheme,
+            '')
+
+    def test_raise_exception_invalid_scheme_http(self):
+        utils.raise_exception_invalid_scheme(url='http://www.openstack.org')
+
+    def test_raise_exception_invalid_scheme_https(self):
+        utils.raise_exception_invalid_scheme(url='https://www.openstack.org')