diff --git a/oletools/common/log_helper/_json_formatter.py b/oletools/common/log_helper/_json_formatter.py index 8e1c6609b..b1b9beb48 100644 --- a/oletools/common/log_helper/_json_formatter.py +++ b/oletools/common/log_helper/_json_formatter.py @@ -5,12 +5,17 @@ class JsonFormatter(logging.Formatter): """ Format every message to be logged as a JSON object + + Uses the standard :py:class:`logging.Formatter` with standard arguments + to do the actual formatting, could save and use a user-supplied formatter + instead. """ _is_first_line = True def __init__(self, other_logger_has_first_line=False): if other_logger_has_first_line: self._is_first_line = False + self.msg_formatter = logging.Formatter() # could adjust this def format(self, record): """ @@ -18,12 +23,34 @@ def format(self, record): the output JSON-compatible. The only exception is when printing the first line, so we need to keep track of it. - We assume that all input comes from the OletoolsLoggerAdapter which - ensures that there is a `type` field in the record. Otherwise will have - to add a try-except around the access to `record.type`. + The actual conversion from :py:class:`logging.LogRecord` to a text message + (i.e. %-formatting, adding exception information, etc.) is delegated to the + standard :py:class:`logging.Formatter. + + The dumped json structure contains fields `msg` with the formatted message, + `level` with the log-level of the message and `type`, which is created by + :py:class:`oletools.common.log_helper.OletoolsLoggerAdapter` or added here + (for input from e.g. captured warnings, third-party libraries) """ - json_dict = dict(msg=record.msg.replace('\n', ' '), level=record.levelname) - json_dict['type'] = record.type + json_dict = dict(msg='', level='', type='') + try: + msg = self.msg_formatter.format(record) + json_dict['msg'] = msg.replace('\n', ' ') + json_dict['level'] = record.levelname + json_dict['type'] = record.type + except AttributeError: # most probably: record has no "type" field + if record.name == 'py.warnings': # this is from python's warning-capture logger + json_dict['type'] = 'warning' + else: + json_dict['type'] = 'msg' # message of unknown origin + except Exception as exc: + try: + json_dict['msg'] = "Ignore {0} when formatting '{1}': {2}".format(type(exc), record.msg, exc) + except Exception as exc2: + json_dict['msg'] = 'Caught {0} in logging'.format(str(exc2)) + json_dict['type'] = 'log-warning' + json_dict['level'] = 'warning' + formatted_message = ' ' + json.dumps(json_dict) if self._is_first_line: diff --git a/oletools/common/log_helper/_logger_adapter.py b/oletools/common/log_helper/_logger_adapter.py index dfc748f89..483144589 100644 --- a/oletools/common/log_helper/_logger_adapter.py +++ b/oletools/common/log_helper/_logger_adapter.py @@ -7,6 +7,7 @@ class OletoolsLoggerAdapter(logging.LoggerAdapter): Adapter class for all loggers returned by the logging module. """ _json_enabled = None + _is_warn_logger = False # this is always False def print_str(self, message, **kwargs): """ @@ -44,7 +45,10 @@ def process(self, msg, kwargs): kwargs['extra']['type'] = kwargs['type'] del kwargs['type'] # downstream loggers cannot deal with this if 'type' not in kwargs['extra']: - kwargs['extra']['type'] = 'msg' # type will be added to LogRecord + if self._is_warn_logger: + kwargs['extra']['type'] = 'warning' # this will add field + else: + kwargs['extra']['type'] = 'msg' # 'type' to LogRecord return msg, kwargs def set_json_enabled_function(self, json_enabled): @@ -53,6 +57,12 @@ def set_json_enabled_function(self, json_enabled): """ self._json_enabled = json_enabled + def set_warnings_logger(self): + """Make this the logger for warnings""" + # create a object attribute that shadows the class attribute which is + # always False + self._is_warn_logger = True + def level(self): """Return current level of logger.""" return self.logger.level diff --git a/oletools/common/log_helper/log_helper.py b/oletools/common/log_helper/log_helper.py index 9ec9843ae..448f8939d 100644 --- a/oletools/common/log_helper/log_helper.py +++ b/oletools/common/log_helper/log_helper.py @@ -152,6 +152,11 @@ def enable_logging(self, use_json=False, level='warning', log_format=DEFAULT_MES self._use_json = use_json sys.excepthook = self._get_except_hook(sys.excepthook) + # make sure warnings do not mess up our output + logging.captureWarnings(True) + warn_logger = self.get_or_create_silent_logger('py.warnings') + warn_logger.set_warnings_logger() + # since there could be loggers already created we go through all of them # and set their levels to 0 so they will use the root logger's level for name in self._all_names: @@ -174,6 +179,7 @@ def end_logging(self): # end logging self._all_names = set() + logging.captureWarnings(False) logging.shutdown() # end json list diff --git a/tests/common/log_helper/log_helper_test_imported.py b/tests/common/log_helper/log_helper_test_imported.py index 1be8181a5..45b41d7c1 100644 --- a/tests/common/log_helper/log_helper_test_imported.py +++ b/tests/common/log_helper/log_helper_test_imported.py @@ -4,6 +4,7 @@ """ from oletools.common.log_helper import log_helper +import warnings DEBUG_MESSAGE = 'imported: debug log' INFO_MESSAGE = 'imported: info log' @@ -11,7 +12,10 @@ ERROR_MESSAGE = 'imported: error log' CRITICAL_MESSAGE = 'imported: critical log' RESULT_MESSAGE = 'imported: result log' + RESULT_TYPE = 'imported: result' +ACTUAL_WARNING = 'Feature XYZ provided by this module might be deprecated at '\ + 'some point in the future ... or not' logger = log_helper.get_or_create_silent_logger('test_imported') @@ -27,3 +31,7 @@ def log(): logger.error(ERROR_MESSAGE) logger.critical(CRITICAL_MESSAGE) logger.info(RESULT_MESSAGE, type=RESULT_TYPE) + + +def warn(): + warnings.warn(ACTUAL_WARNING) diff --git a/tests/common/log_helper/log_helper_test_main.py b/tests/common/log_helper/log_helper_test_main.py index c82f9bcf8..aeba55562 100644 --- a/tests/common/log_helper/log_helper_test_main.py +++ b/tests/common/log_helper/log_helper_test_main.py @@ -2,6 +2,7 @@ import sys import logging +import warnings from tests.common.log_helper import log_helper_test_imported from oletools.common.log_helper import log_helper @@ -11,7 +12,9 @@ ERROR_MESSAGE = 'main: error log' CRITICAL_MESSAGE = 'main: critical log' RESULT_MESSAGE = 'main: result log' + RESULT_TYPE = 'main: result' +ACTUAL_WARNING = 'Warnings can pop up anywhere, have to be prepared!' logger = log_helper.get_or_create_silent_logger('test_main') @@ -24,7 +27,8 @@ def enable_logging(): def main(args): """ - Try to cover possible logging scenarios. For each scenario covered, here's the expected args and outcome: + Try to cover possible logging scenarios. For each scenario covered, here's + the expected args and outcome: - Log without enabling: [''] * logging when being imported - should never print - Log as JSON without enabling: ['as-json', ''] @@ -35,6 +39,8 @@ def main(args): * logging as JSON when being run as script - should log messages as JSON - Enable, log as JSON and throw: ['enable', 'as-json', 'throw', ''] * should produce JSON-compatible output, even after an unhandled exception + - Enable, log as JSON and warn: ['enable', 'as-json', 'warn', ''] + * should produce JSON-compatible output, even after a warning """ # the level should always be the last argument passed @@ -42,6 +48,9 @@ def main(args): use_json = 'as-json' in args throw = 'throw' in args percent_autoformat = '%-autoformat' in args + warn = 'warn' in args + exc_info = 'exc-info' in args + wrong_log_args = 'wrong-log-args' in args log_helper_test_imported.logger.setLevel(logging.ERROR) @@ -53,6 +62,22 @@ def main(args): if throw: raise Exception('An exception occurred before ending the logging') + if warn: + warnings.warn(ACTUAL_WARNING) + log_helper_test_imported.warn() + + if exc_info: + try: + raise Exception('This is an exception') + except Exception: + logger.exception('Caught exception') # has exc_info=True + + if wrong_log_args: + logger.info('Opening file /dangerous/file/with-%s-in-name') + logger.info('The result is %f') + logger.info('No result', 1.23) + logger.info('The result is %f', 'bla') + log_helper.end_logging() diff --git a/tests/common/log_helper/test_log_helper.py b/tests/common/log_helper/test_log_helper.py index f9b20a08b..449da0faf 100644 --- a/tests/common/log_helper/test_log_helper.py +++ b/tests/common/log_helper/test_log_helper.py @@ -16,8 +16,7 @@ from tests.test_utils import PROJECT_ROOT # test file we use as "main" module -TEST_FILE = relpath(join(dirname(abspath(__file__)), 'log_helper_test_main.py'), - PROJECT_ROOT) +TEST_FILE = join(dirname(abspath(__file__)), 'log_helper_test_main.py') # test file simulating a third party main module that only imports oletools TEST_FILE_3RD_PARTY = relpath(join(dirname(abspath(__file__)), @@ -26,6 +25,8 @@ PYTHON_EXECUTABLE = sys.executable +PERCENT_FORMAT_OUTPUT = 'The answer is 47.' + class TestLogHelper(unittest.TestCase): def test_it_doesnt_log_when_not_enabled(self): @@ -113,7 +114,7 @@ def test_logs_type_in_json(self): def test_percent_autoformat(self): """Test that auto-formatting of log strings with `%` works.""" output = self._run_test(['enable', '%-autoformat', 'info']) - self.assertIn('The answer is 47.', output) + self.assertIn(PERCENT_FORMAT_OUTPUT, output) def test_json_correct_on_exceptions(self): """ @@ -141,6 +142,93 @@ def test_import_by_third_party_enabled(self): self.assertIn('INFO:test_main:main: info log', output) self.assertIn('INFO:test_imported:imported: info log', output) + def test_json_correct_on_warnings(self): + """ + Test that even on warnings our JSON is always correct + """ + output = self._run_test(['enable', 'as-json', 'warn', 'warning']) + expected_messages = [ + log_helper_test_main.WARNING_MESSAGE, + log_helper_test_main.ERROR_MESSAGE, + log_helper_test_main.CRITICAL_MESSAGE, + log_helper_test_imported.WARNING_MESSAGE, + log_helper_test_imported.ERROR_MESSAGE, + log_helper_test_imported.CRITICAL_MESSAGE, + ] + + for msg in expected_messages: + self.assertIn(msg, output) + + # last two entries of output should be warnings + jout = json.loads(output) + self.assertEqual(jout[-2]['level'], 'WARNING') + self.assertEqual(jout[-1]['level'], 'WARNING') + self.assertEqual(jout[-2]['type'], 'warning') + self.assertEqual(jout[-1]['type'], 'warning') + self.assertIn(log_helper_test_main.ACTUAL_WARNING, jout[-2]['msg']) + self.assertIn(log_helper_test_imported.ACTUAL_WARNING, jout[-1]['msg']) + + def test_warnings(self): + """Check that warnings are captured and printed correctly""" + output = self._run_test(['enable', 'warn', 'warning']) + + # find out which line contains the call to warnings.warn: + warnings_line = None + with open(TEST_FILE, 'rt') as reader: + for line_idx, line in enumerate(reader): + if 'warnings.warn' in line: + warnings_line = line_idx + 1 + break + self.assertNotEqual(warnings_line, None) + + imported_file = join(dirname(abspath(__file__)), + 'log_helper_test_imported.py') + imported_line = None + with open(imported_file, 'rt') as reader: + for line_idx, line in enumerate(reader): + if 'warnings.warn' in line: + imported_line = line_idx + 1 + break + self.assertNotEqual(imported_line, None) + + expect = '\n'.join([ + 'WARNING ' + log_helper_test_main.WARNING_MESSAGE, + 'ERROR ' + log_helper_test_main.ERROR_MESSAGE, + 'CRITICAL ' + log_helper_test_main.CRITICAL_MESSAGE, + 'WARNING ' + log_helper_test_imported.WARNING_MESSAGE, + 'ERROR ' + log_helper_test_imported.ERROR_MESSAGE, + 'CRITICAL ' + log_helper_test_imported.CRITICAL_MESSAGE, + 'WARNING {0}:{1}: UserWarning: {2}' + .format(TEST_FILE, warnings_line, log_helper_test_main.ACTUAL_WARNING), + ' warnings.warn(ACTUAL_WARNING)', # warnings include source line + '', + 'WARNING {0}:{1}: UserWarning: {2}' + .format(imported_file, imported_line, log_helper_test_imported.ACTUAL_WARNING), + ' warnings.warn(ACTUAL_WARNING)', # warnings include source line + ]) + self.assertEqual(output.strip(), expect) + + def test_json_percent_formatting(self): + """Test that json-output has formatting args included in output.""" + output = self._run_test(['enable', 'as-json', '%-autoformat', 'info']) + json.loads(output) # check that this does not raise, so json is valid + self.assertIn(PERCENT_FORMAT_OUTPUT, output) + + def test_json_exception_formatting(self): + """Test that json-output has formatted exception info in output""" + output = self._run_test(['enable', 'as-json', 'exc-info', 'info']) + json.loads(output) # check that this does not raise, so json is valid + self.assertIn('Caught exception', output) # actual log message + self.assertIn('This is an exception', output) # message of caught exception + self.assertIn('Traceback (most recent call last)', output) # start of trace + self.assertIn(TEST_FILE.replace('\\', '\\\\'), output) # part of trace + + def test_json_wrong_args(self): + """Test that too many or missing args do not raise exceptions inside logger""" + output = self._run_test(['enable', 'as-json', 'wrong-log-args', 'info']) + json.loads(output) # check that this does not raise, so json is valid + # do not care about actual contents of output + def _assert_json_messages(self, output, messages): try: json_data = json.loads(output) @@ -160,8 +248,10 @@ def _run_test(self, args, should_succeed=True, run_third_party=False): we might get errors or false positives between sequential tests runs) When arg `run_third_party` is `True`, we do not run the `TEST_FILE` as - main moduel but the `TEST_FILE_3RD_PARTY` and return contents of + main module but the `TEST_FILE_3RD_PARTY` and return contents of `stderr` instead of `stdout`. + + TODO: use tests.utils.call_and_capture """ all_args = [PYTHON_EXECUTABLE, ] if run_third_party: