github
/
fail2ban
mirror of https://github.com/fail2ban/fail2ban
1
0
Fork 0
Code Issues Releases Wiki Activity

amend with better (common) handling, documentation and tests

pull/2631/head
sebres 2020-08-25 18:01:34 +02:00
parent 7b05c1ce7a
commit d9b8796792
2 changed files with 71 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

93
fail2ban/client/configreader.py
View File

@ -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):

30
fail2ban/tests/clientreadertestcase.py
View File

@ -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>', '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=<IP>']]
filterReader = FilterReader('substition', "jailname", {},