Security: Add bandit job to tox

Change-Id: Idcf6efade852e7de0c636184c21f35a03fe0d980

promenade/builder.py

@@ -3,6 +3,7 @@ import io
 import itertools
 import os
 import requests
+import stat
 import tarfile
 
 __all__ = ['Builder']
@@ -119,5 +120,8 @@ def _join_name(node_name):
 def _write_script(output_dir, name, script):
     path = os.path.join(output_dir, name)
     with open(path, 'w') as f:
-        os.fchmod(f.fileno(), 0o555)
         f.write(script)
+
+    os.chmod(
+        path,
+        os.stat(path).st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)

promenade/config.py

@@ -59,7 +59,9 @@ class Configuration:
 
     def iterate(self, *, kind=None, schema=None, labels=None):
         if kind is not None:
-            assert schema is None
+            if schema is not None:
+                raise AssertionError(
+                    'Logic error: specified both kind and schema')
             schema = 'promenade/%s/v1' % kind
 
         for document in self.documents:
@@ -144,7 +146,8 @@ def _matches_filter(document, *, schema, labels):
 
 def _get(documents, kind=None, schema=None, name=None):
     if kind is not None:
-        assert schema is None
+        if schema is not None:
+            raise AssertionError('Logic error: specified both kind and schema')
         schema = 'promenade/%s/v1' % kind
 
     for document in documents:

promenade/pki.py

@@ -1,7 +1,9 @@
 from . import logging
 import json
 import os
-import subprocess
+# Ignore bandit false positive: B404:blacklist
+# The purpose of this module is to safely encapsulate calls via fork.
+import subprocess  # nosec
 import tempfile
 import yaml
 
@@ -96,7 +98,10 @@ class PKI:
                 with open(os.path.join(tmp, filename), 'w') as f:
                     f.write(data)
 
-            return json.loads(
+            # Ignore bandit false positive:
+            #   B603:subprocess_without_shell_equals_true
+            # This method wraps cfssl calls originating from this module.
+            return json.loads(  # nosec
                 subprocess.check_output(
                     ['cfssl'] + command, cwd=tmp, stderr=subprocess.PIPE))
 
@@ -109,8 +114,13 @@ class PKI:
                 with open(os.path.join(tmp, filename), 'w') as f:
                     f.write(data)
 
-            subprocess.check_call(
+            # Ignore bandit false positive:
-                ['openssl'] + command, cwd=tmp, stderr=subprocess.PIPE)
+            #   B603:subprocess_without_shell_equals_true
+            # This method wraps openssl calls originating from this module.
+            subprocess.check_call(  # nosec
+                ['openssl'] + command,
+                cwd=tmp,
+                stderr=subprocess.PIPE)
 
             result = {}
             for filename in os.listdir(tmp):

promenade/renderer.py

@@ -93,7 +93,9 @@ def render_template(config, *, template, context=None):
 
 
 def _build_env():
-    env = jinja2.Environment(
+    # Ignore bandit false positive: B701:jinja2_autoescape_false
+    # This env is not used to render content that is vulnerable to XSS.
+    env = jinja2.Environment(  # nosec
         loader=jinja2.PackageLoader('promenade', 'templates/include'),
         undefined=jinja2.StrictUndefined)
     env.filters['b64enc'] = _base64_encode

promenade/tar_bundler.py

@@ -26,8 +26,11 @@ class TarBundler:
         tar_info.mode = mode
 
         if tar_info.size > 0:
-            LOG.debug('Adding file path=%s size=%s md5=%s', path,
+            # Ignore bandit false positive: B303:blacklist
-                      tar_info.size, hashlib.md5(data_bytes).hexdigest())
+            # This is a basic checksum for debugging not a secure hash.
+            LOG.debug(  # nosec
+                'Adding file path=%s size=%s md5=%s', path, tar_info.size,
+                hashlib.md5(data_bytes).hexdigest())
         else:
             LOG.warning('Zero length file added to path=%s', path)
 

promenade/validation.py

@@ -46,7 +46,10 @@ def _load_schemas():
         with open(os.path.join(schema_dir, schema_file)) as f:
             for schema in yaml.safe_load_all(f):
                 name = schema['metadata']['name']
-                assert name not in SCHEMAS
+                if name in SCHEMAS:
+                    raise RuntimeError(
+                        'Duplicate schema specified for: %s' % name)
+
                 SCHEMAS[name] = schema['data']
 
 

tox.ini

@@ -1,5 +1,5 @@
 [tox]
-envlist = py35
+envlist = bandit,lint
 
 [testenv]
 deps = -r{toxinidir}/requirements.txt
@@ -10,6 +10,11 @@ commands=
   pytest \
     {posargs}
 
+[testenv:bandit]
+deps = bandit
+commands =
+    bandit -r promenade
+
 [testenv:docs]
 whitelist_externals=rm
 commands =