openstack/bandit
Code Issues Proposed changes

Merge "Fix manager having no attribute '_init_logger'"

Browse Source
This commit is contained in:
Jenkins 2015-09-04 22:33:52 +00:00 committed by Gerrit Code Review
commit db898e896d
13 changed files with 111 additions and 105 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
bandit/bandit.py
View File

@ -23,10 +23,12 @@ import sysconfig
import appdirs
from bandit.core import config as b_config
from bandit.core import manager as b_manager
from bandit.core import utils
BASE_CONFIG = 'bandit.yaml'
logger = logging.getLogger()
def _init_logger(debug=False, log_format=None):
@ -35,6 +37,7 @@ def _init_logger(debug=False, log_format=None):
:param debug: Whether to enable debug mode
:return: An instantiated logging instance
'''
logger.handlers = []
log_level = logging.INFO
if debug:
log_level = logging.DEBUG
@ -47,13 +50,11 @@ def _init_logger(debug=False, log_format=None):
logging.captureWarnings(True)
logger = logging.getLogger()
logger.setLevel(log_level)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(logging.Formatter(log_format_string))
logger.addHandler(handler)
logger.debug("logging initialized")
return logger
def _init_extensions():
@ -99,7 +100,7 @@ def _find_config():
def main():
# bring our logging stuff up as early as possible
debug = ('-d' in sys.argv or '--debug' in sys.argv)
logger = _init_logger(debug)
_init_logger(debug)
# By default path would be /etx/xdg/bandit, we want system paths
os.environ['XDG_CONFIG_DIRS'] = '/etc:/usr/local/etc'
extension_mgr = _init_extensions()
@ -185,13 +186,20 @@ def main():
sys.exit(2)
try:
b_mgr = b_manager.BanditManager(config_file, args.agg_type,
args.debug, profile_name=args.profile,
verbose=args.verbose)
b_conf = b_config.BanditConfig(config_file)
except (utils.ConfigFileUnopenable, utils.ConfigFileInvalidYaml) as e:
logger.error('%s', e)
sys.exit(2)
# if the log format string was set in the options, reinitialize
if b_conf.get_option('log_format'):
log_format = b_conf.get_option('log_format')
_init_logger(debug, log_format=log_format)
b_mgr = b_manager.BanditManager(b_conf, args.agg_type, args.debug,
profile_name=args.profile,
verbose=args.verbose)
if args.output_format != "json":
logger.info("using config: %s", config_file)
logger.info("running on Python %d.%d.%d", sys.version_info.major,

11
bandit/core/config.py
View File

@ -14,6 +14,7 @@
# License for the specific language governing permissions and limitations
# under the License.
import logging
import sys
import yaml
@ -22,17 +23,18 @@ from bandit.core import constants
from bandit.core import utils
logger = logging.getLogger(__name__)
class BanditConfig():
_config = dict()
_logger = None
_settings = dict()
def __init__(self, logger, config_file):
def __init__(self, config_file):
'''Attempt to initialize a config dictionary from a yaml file.
Error out if loading the yaml file fails for any reason.
:param logger: Logger to be used in the case of errors
:param config_file: The Bandit yaml config file
:raises bandit.utils.ConfigFileUnopenable: If the config file cannot be
@ -42,8 +44,7 @@ class BanditConfig():
'''
self._logger = logger
self.config_file = config_file
try:
f = open(config_file, 'r')
except IOError:

7
bandit/core/formatters.py
View File

@ -15,6 +15,7 @@ import collections
import csv
import datetime
import json
import logging
from operator import itemgetter
import six
@ -22,6 +23,9 @@ import six
from bandit.core import constants
logger = logging.getLogger(__name__)
def report_csv(result_store, file_list, scores, excluded_files):
'''Prints/returns warnings in JSON format
@ -215,8 +219,7 @@ def report_text(result_store, files_list, scores, excluded_files):
if result_store.out_file:
with open(result_store.out_file, 'w') as fout:
fout.write(result)
result_store.logger.info("Text output written to file: %s",
result_store.out_file)
logger.info("Text output written to file: %s", result_store.out_file)
else:
print(result)

50
bandit/core/manager.py
View File

@ -19,7 +19,6 @@ import logging
import os
import sys
from bandit.core import config as b_config
from bandit.core import constants as constants
from bandit.core import meta_ast as b_meta_ast
from bandit.core import node_visitor as b_node_visitor
@ -27,55 +26,50 @@ from bandit.core import result_store as b_result_store
from bandit.core import test_set as b_test_set
logger = logging.getLogger(__name__)
class BanditManager():
scope = []
def __init__(self, config_file, agg_type, debug=False, verbose=False,
def __init__(self, config, agg_type, debug=False, verbose=False,
profile_name=None):
'''Get logger, config, AST handler, and result store ready
:param config_file: A file to read config from
:param config: config options object
:type config: bandit.core.BanditConfig
:param agg_type: aggregation type
:param debug: Whether to show debug messsages or not
:param verbose: Whether to show verbose output
:param profile_name: Optional name of profile to use (from cmd line)
:return:
'''
self.debug = debug
self.verbose = verbose
self.logger = logging.getLogger()
self.b_conf = b_config.BanditConfig(self.logger, config_file)
self.b_conf = config
self.files_list = []
self.excluded_files = []
# if the log format string was set in the options, reinitialize
if self.b_conf.get_option('log_format'):
# have to clear old handler
self.logger.handlers = []
log_format = self.b_conf.get_option('log_format')
self.logger = self._init_logger(debug, log_format=log_format)
self.b_ma = b_meta_ast.BanditMetaAst(self.logger)
self.b_rs = b_result_store.BanditResultStore(self.logger, self.b_conf,
agg_type, verbose)
self.b_ma = b_meta_ast.BanditMetaAst()
self.b_rs = b_result_store.BanditResultStore(self.b_conf, agg_type,
verbose)
# if the profile name was specified, try to find it in the config
if profile_name:
if profile_name in self.b_conf.config['profiles']:
profile = self.b_conf.config['profiles'][profile_name]
self.logger.debug(
logger.debug(
"read in profile '%s': %s",
profile_name, profile
)
else:
self.logger.error(
'unable to find profile (%s) in config file: '
'%s', profile_name, config_file
)
logger.error('unable to find profile (%s) in config file: %s',
profile_name, self.b_conf.config_file)
sys.exit(2)
else:
profile = None
self.b_ts = b_test_set.BanditTestSet(self.logger, config=self.b_conf,
self.b_ts = b_test_set.BanditTestSet(config=self.b_conf,
profile=profile)
# set the increment of after how many files to show progress
@ -86,10 +80,6 @@ class BanditManager():
def has_tests(self):
return self.b_ts.has_tests
@property
def get_logger(self):
return self.logger
@property
def get_resultstore(self):
return self.b_rs
@ -172,8 +162,8 @@ class BanditManager():
files_list.update(new_files)
excluded_files.update(newly_excluded)
else:
self.logger.warn("Skipping directory (%s), use -r flag to "
"scan contents", fname)
logger.warn("Skipping directory (%s), use -r flag to "
"scan contents", fname)
else:
# if the user explicitly mentions a file on command line,
@ -218,7 +208,7 @@ class BanditManager():
new_files_list = list(self.files_list)
for count, fname in enumerate(self.files_list):
self.logger.debug("working on file : %s", fname)
logger.debug("working on file : %s", fname)
if len(self.files_list) > self.progress:
# is it time to update the progress indicator?
@ -264,7 +254,7 @@ class BanditManager():
score = []
if fdata is not None:
res = b_node_visitor.BanditNodeVisitor(
fname, self.logger, self.b_conf, b_ma, b_rs, b_ts, self.debug
fname, self.b_conf, b_ma, b_rs, b_ts, self.debug
)
score = res.process(fdata)
return score

9
bandit/core/meta_ast.py
View File

@ -16,15 +16,16 @@
from collections import OrderedDict
import logging
logger = logging.getLogger(__name__)
class BanditMetaAst():
nodes = OrderedDict()
def __init__(self, logger):
self.logger = logger
def add_node(self, node, parent_id, depth):
'''Add a node to the AST node collection
@ -34,7 +35,7 @@ class BanditMetaAst():
:return: -
'''
node_id = hex(id(node))
self.logger.debug('adding node : %s [%s]', node_id, depth)
logger.debug('adding node : %s [%s]', node_id, depth)
self.nodes[node_id] = {
'raw': node, 'parent_id': parent_id, 'depth': depth
}

47
bandit/core/node_visitor.py
View File

@ -16,6 +16,7 @@
import ast
import copy
import logging
from bandit.core import constants
from bandit.core import tester as b_tester
@ -23,13 +24,16 @@ from bandit.core import utils as b_utils
from bandit.core.utils import InvalidModulePath
logger = logging.getLogger(__name__)
class BanditNodeVisitor(object):
context_template = {'node': None, 'filename': None,
'name': None, 'qualname': None, 'module': None,
'imports': None, 'import_aliases': None, 'call': None,
'function': None, 'lineno': None, 'skip_lines': None}
def __init__(self, fname, logger, config, metaast, results, testset,
def __init__(self, fname, config, metaast, results, testset,
debug):
self.debug = debug
self.seen = 0
@ -39,7 +43,6 @@ class BanditNodeVisitor(object):
}
self.depth = 0
self.fname = fname
self.logger = logger
self.config = config
self.metaast = metaast
self.results = results
@ -49,17 +52,17 @@ class BanditNodeVisitor(object):
self.import_aliases = {}
self.context_template['import_aliases'] = self.import_aliases
self.tester = b_tester.BanditTester(
self.logger, self.config, self.results, self.testset, self.debug
self.config, self.results, self.testset, self.debug
)
# in some cases we can't determine a qualified name
try:
self.namespace = b_utils.get_module_qualname_from_path(fname)
except InvalidModulePath:
self.logger.info('Unable to find qualified name for module: %s',
self.fname)
logger.info('Unable to find qualified name for module: %s',
self.fname)
self.namespace = ""
self.logger.debug('Module qualified name: %s', self.namespace)
logger.debug('Module qualified name: %s', self.namespace)
self.lines = []
def visit_ClassDef(self, node):
@ -71,7 +74,7 @@ class BanditNodeVisitor(object):
'''
if self.debug:
self.logger.debug("visit_ClassDef called (%s)", ast.dump(node))
logger.debug("visit_ClassDef called (%s)", ast.dump(node))
# For all child nodes, add this class name to current namespace
self.namespace = b_utils.namespace_path_join(self.namespace, node.name)
@ -91,7 +94,7 @@ class BanditNodeVisitor(object):
self.context['function'] = node
if self.debug:
self.logger.debug("visit_FunctionDef called (%s)", ast.dump(node))
logger.debug("visit_FunctionDef called (%s)", ast.dump(node))
qualname = self.namespace + '.' + b_utils.get_func_name(node)
name = qualname.split('.')[-1]
@ -118,7 +121,7 @@ class BanditNodeVisitor(object):
self.context['call'] = node
if self.debug:
self.logger.debug("visit_Call called (%s)", ast.dump(node))
logger.debug("visit_Call called (%s)", ast.dump(node))
qualname = b_utils.get_call_name(node, self.import_aliases)
name = qualname.split('.')[-1]
@ -138,7 +141,7 @@ class BanditNodeVisitor(object):
:return: -
'''
if self.debug:
self.logger.debug("visit_Import called (%s)", ast.dump(node))
logger.debug("visit_Import called (%s)", ast.dump(node))
for nodename in node.names:
if nodename.asname:
@ -157,7 +160,7 @@ class BanditNodeVisitor(object):
:return: -
'''
if self.debug:
self.logger.debug("visit_ImportFrom called (%s)", ast.dump(node))
logger.debug("visit_ImportFrom called (%s)", ast.dump(node))
module = node.module
if module is None:
@ -195,7 +198,7 @@ class BanditNodeVisitor(object):
self.context['str'] = node.s
if self.debug:
self.logger.debug("visit_Str called (%s)", ast.dump(node))
logger.debug("visit_Str called (%s)", ast.dump(node))
if not isinstance(node.parent, ast.Expr): # docstring
self.context['linerange'] = b_utils.linerange_fix(node.parent)
@ -213,7 +216,7 @@ class BanditNodeVisitor(object):
self.context['bytes'] = node.s
if self.debug:
self.logger.debug("visit_Bytes called (%s)", ast.dump(node))
logger.debug("visit_Bytes called (%s)", ast.dump(node))
if not isinstance(node.parent, ast.Expr): # docstring
self.context['linerange'] = b_utils.linerange_fix(node.parent)
@ -224,7 +227,7 @@ class BanditNodeVisitor(object):
self.context['str'] = 'exec'
if self.debug:
self.logger.debug("visit_Exec called (%s)", ast.dump(node))
logger.debug("visit_Exec called (%s)", ast.dump(node))
self.update_scores(self.tester.run_tests(self.context, 'Exec'))
self.generic_visit(node)
@ -233,15 +236,15 @@ class BanditNodeVisitor(object):
self.context['str'] = 'assert'
if self.debug:
self.logger.debug("visit_Assert called (%s)", ast.dump(node))
logger.debug("visit_Assert called (%s)", ast.dump(node))
self.update_scores(self.tester.run_tests(self.context, 'Assert'))
self.generic_visit(node)
def visit_ExceptHandler(self, node):
if self.debug:
self.logger.debug("visit_ExceptHandler called (%s)",
ast.dump(node))
logger.debug("visit_ExceptHandler called (%s)",
ast.dump(node))
self.update_scores(self.tester.run_tests(self.context,
'ExceptHandler'))
@ -257,7 +260,7 @@ class BanditNodeVisitor(object):
self.context = copy.copy(self.context_template)
if self.debug:
self.logger.debug(ast.dump(node))
logger.debug(ast.dump(node))
if self.debug:
self.metaast.add_node(node, '', self.depth)
@ -266,7 +269,7 @@ class BanditNodeVisitor(object):
self.context['lineno'] = node.lineno
if ("# nosec" in self.lines[node.lineno - 1] or
"#nosec" in self.lines[node.lineno - 1]):
self.logger.debug("skipped, nosec")
logger.debug("skipped, nosec")
return
self.context['node'] = node
@ -274,8 +277,8 @@ class BanditNodeVisitor(object):
self.context['filename'] = self.fname
self.seen += 1
self.logger.debug("entering: %s %s [%s]", hex(id(node)), type(node),
self.depth)
logger.debug("entering: %s %s [%s]", hex(id(node)), type(node),
self.depth)
self.depth += 1
method = 'visit_' + node.__class__.__name__
@ -283,7 +286,7 @@ class BanditNodeVisitor(object):
visitor(node)
self.depth -= 1
self.logger.debug("%s\texiting : %s", self.depth, hex(id(node)))
logger.debug("%s\texiting : %s", self.depth, hex(id(node)))
def generic_visit(self, node):
"""Drive the visitor."""

3
bandit/core/result_store.py
View File

@ -30,11 +30,10 @@ class BanditResultStore():
count = 0
skipped = None
def __init__(self, logger, config, agg_type, verbose):
def __init__(self, config, agg_type, verbose):
self.resstore = OrderedDict()
self.count = 0
self.skipped = []
self.logger = logger
self.config = config
self.agg_type = agg_type
self.sev_level = 0

3
bandit/core/test_properties.py
View File

@ -18,7 +18,8 @@ import logging
from bandit.core import constants
logger = logging.getLogger()
logger = logging.getLogger(__name__)
def severity(sev):

22
bandit/core/test_set.py
View File

@ -17,17 +17,20 @@
from collections import OrderedDict
import copy
import logging
import sys
from bandit.core import utils
logger = logging.getLogger(__name__)
class BanditTestSet():
tests = OrderedDict()
def __init__(self, logger, config, profile=None):
self.logger = logger
def __init__(self, config, profile=None):
self.config = config
filter_list = self._filter_list_from_config(profile=profile)
self.load_tests(filter=filter_list)
@ -62,7 +65,7 @@ class BanditTestSet():
for exc in profile['exclude']:
exclude_list.append(exc)
self.logger.debug(
logger.debug(
"_filter_list_from_config completed - include: %s, exclude %s",
include_list, exclude_list
)
@ -100,9 +103,9 @@ class BanditTestSet():
# copy tests back over from temp copy
self.tests = copy.deepcopy(temp_dict)
self.logger.debug('obtained filtered set of tests:')
logger.debug('obtained filtered set of tests:')
for k in self.tests:
self.logger.debug('\t%s : %s', k, self.tests[k])
logger.debug('\t%s : %s', k, self.tests[k])
def _get_extension_manager(self):
from bandit.core import extension_loader
@ -129,14 +132,14 @@ class BanditTestSet():
'(unknown)')
path2 = utils.get_path_for_function(
self.tests[check][fn_name]) or '(unknown)'
self.logger.error(
logger.error(
"Duplicate function definition "
"%s in %s and %s", fn_name, path1, path2
)
sys.exit(2)
else:
self.tests[check][fn_name] = function
self.logger.debug(
logger.debug(
'added function %s targetting %s',
fn_name, check
)
@ -149,11 +152,10 @@ class BanditTestSet():
:return: A dictionary of tests which are of the specified type
'''
scoped_tests = {}
self.logger.debug('get_tests called with check type: %s', checktype)
logger.debug('get_tests called with check type: %s', checktype)
if checktype in self.tests:
scoped_tests = self.tests[checktype]
self.logger.debug('get_tests returning scoped_tests : %s',
scoped_tests)
logger.debug('get_tests returning scoped_tests : %s', scoped_tests)
return scoped_tests
@property

11
bandit/core/tester.py
View File

@ -15,6 +15,7 @@
# under the License.
import copy
import logging
import warnings
import six
@ -24,14 +25,14 @@ from bandit.core import context as b_context
from bandit.core import utils
warnings.formatwarning = utils.warnings_formatter
logger = logging.getLogger(__name__)
class BanditTester():
results = None
def __init__(self, logger, config, results, testset, debug):
self.logger = logger
def __init__(self, config, results, testset, debug):
self.config = config
self.results = results
self.testset = testset
@ -89,7 +90,7 @@ class BanditTester():
# if we have a result, record it and update scores
if result is not None:
self.results.add(temp_context, name, result)
self.logger.debug(
logger.debug(
"Issue identified by %s: %s", name, result
)
sev = constants.RANKING.index(result[0])
@ -103,7 +104,7 @@ class BanditTester():
self.report_error(name, context, e)
if self.debug:
raise
self.logger.debug("Returning scores: %s", scores)
logger.debug("Returning scores: %s", scores)
return scores
def report_error(self, test, context, error):
@ -116,4 +117,4 @@ class BanditTester():
what += str(error)
import traceback
what += traceback.format_exc()
self.logger.error(what)
logger.error(what)

8
tests/functional/test_functional.py
View File

@ -21,6 +21,7 @@ import inspect
import six
import testtools
from bandit.core import config as b_config
from bandit.core import constants as C
from bandit.core import manager as b_manager
from bandit.core import test_set as b_test_set
@ -43,11 +44,10 @@ class FunctionalTests(testtools.TestCase):
# them up here for the testing environment.
#
path = os.path.join(os.getcwd(), 'bandit', 'plugins')
self.b_mgr = b_manager.BanditManager(cfg_file, 'file')
b_conf = b_config.BanditConfig(cfg_file)
self.b_mgr = b_manager.BanditManager(b_conf, 'file')
self.b_mgr.b_conf._settings['plugins_dir'] = path
self.b_mgr.b_ts = b_test_set.BanditTestSet(self.b_mgr.logger,
config=self.b_mgr.b_conf,
profile=None)
self.b_mgr.b_ts = b_test_set.BanditTestSet(config=b_conf)
def run_example(self, example_script):
'''A helper method to run the specified test

20
tests/unit/core/test_config.py
View File

@ -26,9 +26,6 @@ from bandit.core import config
from bandit.core import utils
LOG = logging.getLogger('bandit.test')
class TempFile(fixtures.Fixture):
def __init__(self, contents=None):
super(TempFile, self).__init__()
@ -56,7 +53,7 @@ class TestInit(testtools.TestCase):
# Can initialize a BanditConfig.
f = self.useFixture(TempFile())
b_config = config.BanditConfig(LOG, f.name)
b_config = config.BanditConfig(f.name)
# After initialization, can get settings.
self.assertEqual(50, b_config.get_setting('progress'))
@ -75,7 +72,7 @@ class TestInit(testtools.TestCase):
cfg_file = os.path.join(os.getcwd(), 'notafile')
self.assertRaisesRegex(utils.ConfigFileUnopenable, cfg_file,
config.BanditConfig, LOG, cfg_file)
config.BanditConfig, cfg_file)
def test_yaml_invalid(self):
# When the config yaml file isn't valid, sys.exit(2) is called.
@ -85,8 +82,7 @@ class TestInit(testtools.TestCase):
invalid_yaml = '- [ something'
f = self.useFixture(TempFile(invalid_yaml))
self.assertRaisesRegex(
utils.ConfigFileInvalidYaml, f.name, config.BanditConfig,
LOG, f.name)
utils.ConfigFileInvalidYaml, f.name, config.BanditConfig, f.name)
def test_progress_conf_setting(self):
# The progress setting can be set in bandit.yaml via
@ -96,7 +92,7 @@ class TestInit(testtools.TestCase):
sample_yaml = 'show_progress_every: %s' % example_value
f = self.useFixture(TempFile(sample_yaml))
b_config = config.BanditConfig(LOG, f.name)
b_config = config.BanditConfig(f.name)
self.assertEqual(example_value, b_config.get_setting('progress'))
def test_colors_isatty_defaults(self):
@ -107,7 +103,7 @@ class TestInit(testtools.TestCase):
self.useFixture(
fixtures.MockPatch('sys.stdout.isatty', return_value=True))
b_config = config.BanditConfig(LOG, f.name)
b_config = config.BanditConfig(f.name)
self.assertEqual('\x1b[95m', b_config.get_setting('color_HEADER'))
self.assertEqual('\x1b[0m', b_config.get_setting('color_DEFAULT'))
@ -127,7 +123,7 @@ output_colors:
"""
f = self.useFixture(TempFile(sample_yaml))
b_config = config.BanditConfig(LOG, f.name)
b_config = config.BanditConfig(f.name)
self.assertEqual('\x1b[23m', b_config.get_setting('color_HEADER'))
@ -145,7 +141,7 @@ class TestGetOption(testtools.TestCase):
""" % (self.example_key, self.example_subkey, self.example_subvalue)
f = self.useFixture(TempFile(sample_yaml))
self.b_config = config.BanditConfig(LOG, f.name)
self.b_config = config.BanditConfig(f.name)
def test_levels(self):
# get_option with .-separated string.
@ -165,7 +161,7 @@ class TestGetSetting(testtools.TestCase):
def setUp(self):
super(TestGetSetting, self).setUp()
f = self.useFixture(TempFile())
self.b_config = config.BanditConfig(LOG, f.name)
self.b_config = config.BanditConfig(f.name)
def test_not_exist(self):
# get_setting() when the name doesn't exist returns None

5
tests/unit/core/test_formatters.py
View File

@ -24,6 +24,7 @@ import testtools
import bandit
from bandit.core import constants
from bandit.core import config
from bandit.core import manager
from bandit.core import formatters
@ -33,8 +34,8 @@ class FormattersTests(testtools.TestCase):
def setUp(self):
super(FormattersTests, self).setUp()
cfg_file = os.path.join(os.getcwd(), 'bandit/config/bandit.yaml')
path = os.path.join(os.getcwd(), 'bandit', 'plugins')
self.manager = manager.BanditManager(cfg_file, 'file')
conf = config.BanditConfig(cfg_file)
self.manager = manager.BanditManager(conf, 'file')
(tmp_fd, self.tmp_fname) = tempfile.mkstemp()
self.context = {'filename': self.tmp_fname,
'lineno': 4,