Assert problems in Glance raised by Bandit

Fix all assert problems raised by Bandit. Asserts are potentially
problematic, since Python optimization sometimes removes them, so code
needs to remain safe and functional without the assert.

Two asserts are safe to skip, so they are deleted for improved error
messages. Three asserts are probably necessary, and are converted to
exceptions. Two asserts are probably necessary, and are instead made to
fail safely, and `# nosec` is added to the assert line.

This also enables the assert test in bandit's configuration.

Change-Id: Ic69a204ceb15cac234c6b6bca3d950256a98016d
Partial-bug: 1511862

bandit.yaml

@@ -33,7 +33,7 @@ profiles:
         include:
 
             - any_other_function_with_shell_equals_true
-            # - assert_used # TODO: enable this test
+            - assert_used
             - blacklist_calls
             - blacklist_import_func
 

glance/api/v2/images.py

@@ -144,7 +144,6 @@ class ImagesController(object):
 
             for change in changes:
                 change_method_name = '_do_%s' % change['op']
-                assert hasattr(self, change_method_name)
                 change_method = getattr(self, change_method_name)
                 change_method(req, image, change)
 

glance/common/rpc.py

@@ -114,7 +114,7 @@ class Controller(object):
         :params excluded: List of methods to exclude.
         :params refiner: Callable to use as filter for methods.
 
-        :raises: AssertionError: If refiner is not callable.
+        :raises TypeError: If refiner is not callable.
         """
 
         funcs = filter(lambda x: not x.startswith("_"), dir(resource))
@@ -126,7 +126,6 @@ class Controller(object):
             funcs = [f for f in funcs if f not in excluded]
 
         if refiner:
-            assert callable(refiner), "Refiner must be callable"
             funcs = filter(refiner, funcs)
 
         for name in funcs:

glance/db/__init__.py

@@ -163,8 +163,9 @@ class ImageRepo(object):
     def get(self, image_id):
         try:
             db_api_image = dict(self.db_api.image_get(self.context, image_id))
-            assert not db_api_image['deleted']
-        except (exception.ImageNotFound, exception.Forbidden, AssertionError):
+            if db_api_image['deleted']:
+                raise exception.ImageNotFound()
+        except (exception.ImageNotFound, exception.Forbidden):
             msg = _("No image found with ID %s") % image_id
             raise exception.ImageNotFound(msg)
         tags = self.db_api.image_tag_get_all(self.context, image_id)

glance/db/sqlalchemy/api.py

@@ -330,17 +330,21 @@ def _paginate_query(query, model, limit, sort_keys, marker=None,
         # the actual primary key, rather than assuming its id
         LOG.warn(_LW('Id not in sort_keys; is sort_keys unique?'))
 
-    assert(not (sort_dir and sort_dirs))
+    assert(not (sort_dir and sort_dirs))  # nosec
+    # nosec: This function runs safely if the assertion fails.
 
     # Default the sort direction to ascending
-    if sort_dirs is None and sort_dir is None:
+    if sort_dir is None:
         sort_dir = 'asc'
 
     # Ensure a per-column sort direction
     if sort_dirs is None:
-        sort_dirs = [sort_dir for _sort_key in sort_keys]
+        sort_dirs = [sort_dir] * len(sort_keys)
 
-    assert(len(sort_dirs) == len(sort_keys))
+    assert(len(sort_dirs) == len(sort_keys))  # nosec
+    # nosec: This function runs safely if the assertion fails.
+    if len(sort_dirs) < len(sort_keys):
+        sort_dirs += [sort_dir] * (len(sort_keys) - len(sort_dirs))
 
     # Add sorting
     for current_sort_key, current_sort_dir in zip(sort_keys, sort_dirs):

glance/db/sqlalchemy/artifacts.py

@@ -328,14 +328,16 @@ def _get_all(context, session, filters=None, marker=None,
 def _do_paginate_query(query, sort_keys=None, sort_dirs=None,
                        marker=None, limit=None):
     # Default the sort direction to ascending
-    if sort_dirs is None:
-        sort_dir = 'asc'
+    sort_dir = 'asc'
 
     # Ensure a per-column sort direction
     if sort_dirs is None:
-        sort_dirs = [sort_dir for _sort_key in sort_keys]
+        sort_dirs = [sort_dir] * len(sort_keys)
 
-    assert(len(sort_dirs) == len(sort_keys))
+    assert(len(sort_dirs) == len(sort_keys))  # nosec
+    # nosec: This function runs safely if the assertion fails.
+    if len(sort_dirs) < len(sort_keys):
+        sort_dirs += [sort_dir] * (len(sort_keys) - len(sort_dirs))
 
     # Add sorting
     for current_sort_key, current_sort_dir in zip(sort_keys, sort_dirs):

glance/db/sqlalchemy/migrate_repo/versions/015_quote_swift_credentials.py

@@ -98,7 +98,9 @@ def legacy_parse_uri(uri, to_quote):
         raise exception.BadStoreUri(message=reason)
 
     pieces = urlparse.urlparse(uri)
-    assert pieces.scheme in ('swift', 'swift+http', 'swift+https')
+    if pieces.scheme not in ('swift', 'swift+http', 'swift+https'):
+        raise exception.BadStoreUri(message="Unacceptable scheme: '%s'" %
+                                    pieces.scheme)
     scheme = pieces.scheme
     netloc = pieces.netloc
     path = pieces.path.lstrip('/')

glance/db/sqlalchemy/migrate_repo/versions/017_quote_encrypted_swift_credentials.py

@@ -162,7 +162,9 @@ def legacy_parse_uri(uri, to_quote):
         raise exception.BadStoreUri(message=reason)
 
     pieces = urlparse.urlparse(uri)
-    assert pieces.scheme in ('swift', 'swift+http', 'swift+https')
+    if pieces.scheme not in ('swift', 'swift+http', 'swift+https'):
+        raise exception.BadStoreUri(message="Unacceptable scheme: '%s'" %
+                                    pieces.scheme)
     scheme = pieces.scheme
     netloc = pieces.netloc
     path = pieces.path.lstrip('/')

glance/tests/unit/common/test_rpc.py

@@ -108,7 +108,7 @@ class TestRPCController(base.IsolatedUnitTest):
         controller = rpc.Controller()
 
         # Not callable
-        self.assertRaises(AssertionError,
+        self.assertRaises(TypeError,
                           controller.register,
                           res, refiner="get_all_images")
 

glance/tests/unit/test_migrations.py

@@ -420,7 +420,7 @@ class MigrationsMixin(test_migrations.WalkVersionsMixin):
 
         invalid_scheme_uri = ('http://acct:usr:pass@example.com'
                               '/container/obj-id')
-        self.assertRaises(AssertionError,
+        self.assertRaises(exception.BadStoreUri,
                           legacy_parse_uri_fn,
                           invalid_scheme_uri,
                           True)

glance/tests/unit/v2/test_images_resource.py

@@ -1899,8 +1899,8 @@ class TestImagesController(base.IsolatedUnitTest):
         change = {'op': 'test', 'path': 'options', 'value': 'puts'}
         try:
             self.controller.update(request, UUID1, [change])
-        except AssertionError:
-            pass  # AssertionError is the desired behavior
+        except AttributeError:
+            pass  # AttributeError is the desired behavior
         else:
             self.fail('Failed to raise AssertionError on %s' % change)