From d9b8796792ff2d876fe02e5a469767d33469c276 Mon Sep 17 00:00:00 2001 From: sebres Date: Tue, 25 Aug 2020 18:01:34 +0200 Subject: [PATCH] amend with better (common) handling, documentation and tests --- fail2ban/client/configreader.py | 93 +++++++++++++------------- fail2ban/tests/clientreadertestcase.py | 30 +++++++-- 2 files changed, 71 insertions(+), 52 deletions(-) diff --git a/fail2ban/client/configreader.py b/fail2ban/client/configreader.py index f9a6d288..d8817bed 100644 --- a/fail2ban/client/configreader.py +++ b/fail2ban/client/configreader.py @@ -34,6 +34,30 @@ from ..helpers import getLogger, _as_bool, _merge_dicts, substituteRecursiveTags # Gets the instance of the logger. logSys = getLogger(__name__) +CONVERTER = { + "bool": _as_bool, + "int": int, +} +def _OptionsTemplateGen(options): + """Iterator over the options template with default options. + + Each options entry is composed of an array or tuple with: + [[type, name, ?default?], ...] + Or it is a dict: + {name: [type, default], ...} + """ + if isinstance(options, (list,tuple)): + for optname in options: + if len(optname) > 2: + opttype, optname, optvalue = optname + else: + (opttype, optname), optvalue = optname, None + yield opttype, optname, optvalue + else: + for optname in options: + opttype, optvalue = options[optname] + yield opttype, optname, optvalue + class ConfigReader(): """Generic config reader class. @@ -233,29 +257,17 @@ class ConfigReaderUnshared(SafeConfigParserWithIncludes): if pOptions is None: pOptions = {} # Get only specified options: - for optname in options: - if isinstance(options, (list,tuple)): - if len(optname) > 2: - opttype, optname, optvalue = optname - else: - (opttype, optname), optvalue = optname, None - else: - opttype, optvalue = options[optname] + for opttype, optname, optvalue in _OptionsTemplateGen(options): if optname in pOptions: continue try: - if convert: - if opttype == "bool": - v = self.getboolean(sec, optname) - if v is None: continue - elif opttype == "int": - v = self.getint(sec, optname) - if v is None: continue - else: - v = self.get(sec, optname, vars=pOptions) - else: - v = self.get(sec, optname, vars=pOptions) + v = self.get(sec, optname, vars=pOptions) values[optname] = v + if convert: + conv = CONVERTER.get(opttype) + if conv: + if v is None: continue + values[optname] = conv(v) except NoSectionError as e: if shouldExist: raise @@ -350,33 +362,20 @@ class DefinitionInitConfigReader(ConfigReader): if opt == '__name__' or opt in self._opts: continue self._opts[opt] = self.get("Definition", opt) - def convertOptions(self, opts, pOptions={}): - options = self._configOpts - for optname in options: - if isinstance(options, (list,tuple)): - if len(optname) > 2: - opttype, optname, optvalue = optname - else: - (opttype, optname), optvalue = optname, None - else: - opttype, optvalue = options[optname] - if optname in pOptions: - continue - try: - if opttype == "bool": - v = opts.get(optname) - if v is None or isinstance(v, bool): continue - v = _as_bool(v) - opts[optname] = v - elif opttype == "int": - v = opts.get(optname) - if v is None or isinstance(v, (int, long)): continue - v = int(v) - opts[optname] = v - except ValueError: - logSys.warning("Wrong %s value %r for %r. Using default one: %r", - opttype, v, optname, optvalue) - opts[optname] = optvalue + def convertOptions(self, opts, configOpts): + """Convert interpolated combined options to expected type. + """ + for opttype, optname, optvalue in _OptionsTemplateGen(configOpts): + conv = CONVERTER.get(opttype) + if conv: + v = opts.get(optname) + if v is None: continue + try: + opts[optname] = conv(v) + except ValueError: + logSys.warning("Wrong %s value %r for %r. Using default one: %r", + opttype, v, optname, optvalue) + opts[optname] = optvalue def getCombOption(self, optname): """Get combined definition option (as string) using pre-set and init @@ -413,7 +412,7 @@ class DefinitionInitConfigReader(ConfigReader): if not opts: raise ValueError('recursive tag definitions unable to be resolved') # convert options after all interpolations: - self.convertOptions(opts) + self.convertOptions(opts, self._configOpts) return opts def convert(self): diff --git a/fail2ban/tests/clientreadertestcase.py b/fail2ban/tests/clientreadertestcase.py index ae39c157..8abfd4a5 100644 --- a/fail2ban/tests/clientreadertestcase.py +++ b/fail2ban/tests/clientreadertestcase.py @@ -87,6 +87,21 @@ option = %s self.assertTrue(self.c.read(f)) # we got some now return self.c.getOptions('section', [("int", 'option')])['option'] + def testConvert(self): + self.c.add_section("Definition") + self.c.set("Definition", "a", "1") + self.c.set("Definition", "b", "1") + self.c.set("Definition", "c", "test") + opts = self.c.getOptions("Definition", + (('int', 'a', 0), ('bool', 'b', 0), ('int', 'c', 0))) + self.assertSortedEqual(opts, {'a': 1, 'b': True, 'c': 0}) + opts = self.c.getOptions("Definition", + (('int', 'a'), ('bool', 'b'), ('int', 'c'))) + self.assertSortedEqual(opts, {'a': 1, 'b': True, 'c': None}) + opts = self.c.getOptions("Definition", + {'a': ('int', 0), 'b': ('bool', 0), 'c': ('int', 0)}) + self.assertSortedEqual(opts, {'a': 1, 'b': True, 'c': 0}) + def testInaccessibleFile(self): f = os.path.join(self.d, "d.conf") # inaccessible file self._write('d.conf', 0) @@ -483,11 +498,7 @@ class JailReaderTest(LogCaptureTestCase): self.assertRaises(NoSectionError, c.getOptions, 'test', {}) -class FilterReaderTest(unittest.TestCase): - - def __init__(self, *args, **kwargs): - super(FilterReaderTest, self).__init__(*args, **kwargs) - self.__share_cfg = {} +class FilterReaderTest(LogCaptureTestCase): def testConvert(self): output = [ @@ -533,6 +544,15 @@ class FilterReaderTest(unittest.TestCase): output[0][-1] = 5; # maxlines = 5 self.assertSortedEqual(filterReader.convert(), output) + def testConvertOptions(self): + filterReader = FilterReader("testcase01", "testcase01", {'maxlines': '', 'test': 'X'}, + share_config=TEST_FILES_DIR_SHARE_CFG, basedir=TEST_FILES_DIR) + filterReader.read() + filterReader.getOptions(None) + opts = filterReader.getCombined(); + self.assertNotEqual(opts['maxlines'], 'X'); # wrong int value 'X' for 'maxlines' + self.assertLogged("Wrong int value 'X' for 'maxlines'. Using default one:") + def testFilterReaderSubstitionDefault(self): output = [['set', 'jailname', 'addfailregex', 'to=sweet@example.com fromip=']] filterReader = FilterReader('substition', "jailname", {},