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

introduces new decorator/conditional helper in order to skip some STOCK-related test-cases (if running outside of stock-config environment).

pull/2025/head
sebres 2018-01-22 14:40:04 +01:00
parent 9af9ec25f5
commit 7a757645bb
6 changed files with 1042 additions and 1013 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
fail2ban/tests/action_d/test_smtp.py
View File

@ -52,6 +52,7 @@ class SMTPActionTest(unittest.TestCase):
def setUp(self): def setUp(self):
"""Call before every test case.""" """Call before every test case."""
unittest.F2B.SkipIfCfgMissing(action='smtp.py')
super(SMTPActionTest, self).setUp() super(SMTPActionTest, self).setUp()
self.jail = DummyJail() self.jail = DummyJail()
pythonModule = os.path.join(CONFIG_DIR, "action.d", "smtp.py") pythonModule = os.path.join(CONFIG_DIR, "action.d", "smtp.py")

415
fail2ban/tests/clientreadertestcase.py
View File

@ -45,8 +45,6 @@ TEST_FILES_DIR_SHARE_CFG = {}
from .utils import CONFIG_DIR from .utils import CONFIG_DIR
CONFIG_DIR_SHARE_CFG = unittest.F2B.share_config CONFIG_DIR_SHARE_CFG = unittest.F2B.share_config
STOCK = os.path.exists(os.path.join('config', 'fail2ban.conf'))
IMPERFECT_CONFIG = os.path.join(os.path.dirname(__file__), 'config') IMPERFECT_CONFIG = os.path.join(os.path.dirname(__file__), 'config')
IMPERFECT_CONFIG_SHARE_CFG = {} IMPERFECT_CONFIG_SHARE_CFG = {}
@ -246,15 +244,15 @@ class JailReaderTest(LogCaptureTestCase):
self.assertTrue(jail.isEnabled()) self.assertTrue(jail.isEnabled())
self.assertLogged("Invalid filter definition 'flt[test'") self.assertLogged("Invalid filter definition 'flt[test'")
if STOCK: def testStockSSHJail(self):
def testStockSSHJail(self): unittest.F2B.SkipIfCfgMissing(stock=True)
jail = JailReader('sshd', basedir=CONFIG_DIR, share_config=CONFIG_DIR_SHARE_CFG) # we are running tests from root project dir atm jail = JailReader('sshd', basedir=CONFIG_DIR, share_config=CONFIG_DIR_SHARE_CFG) # we are running tests from root project dir atm
self.assertTrue(jail.read()) self.assertTrue(jail.read())
self.assertTrue(jail.getOptions()) self.assertTrue(jail.getOptions())
self.assertFalse(jail.isEnabled()) self.assertFalse(jail.isEnabled())
self.assertEqual(jail.getName(), 'sshd') self.assertEqual(jail.getName(), 'sshd')
jail.setName('ssh-funky-blocker') jail.setName('ssh-funky-blocker')
self.assertEqual(jail.getName(), 'ssh-funky-blocker') self.assertEqual(jail.getName(), 'ssh-funky-blocker')
def testSplitOption(self): def testSplitOption(self):
# Simple example # Simple example
@ -307,6 +305,7 @@ class JailReaderTest(LogCaptureTestCase):
self.assertEqual(expected2, result) self.assertEqual(expected2, result)
def testVersionAgent(self): def testVersionAgent(self):
unittest.F2B.SkipIfCfgMissing(stock=True)
jail = JailReader('blocklisttest', force_enable=True, basedir=CONFIG_DIR) jail = JailReader('blocklisttest', force_enable=True, basedir=CONFIG_DIR)
# emulate jail.read(), because such jail not exists: # emulate jail.read(), because such jail not exists:
ConfigReader.read(jail, "jail"); ConfigReader.read(jail, "jail");
@ -597,222 +596,226 @@ class JailsReaderTest(LogCaptureTestCase):
self.assertNotLogged("Skipping...") self.assertNotLogged("Skipping...")
self.assertLogged("No file(s) found for glob /weapons/of/mass/destruction") self.assertLogged("No file(s) found for glob /weapons/of/mass/destruction")
if STOCK: def testReadStockActionConf(self):
def testReadStockActionConf(self): unittest.F2B.SkipIfCfgMissing(stock=True)
for actionConfig in glob.glob(os.path.join(CONFIG_DIR, 'action.d', '*.conf')): for actionConfig in glob.glob(os.path.join(CONFIG_DIR, 'action.d', '*.conf')):
actionName = os.path.basename(actionConfig).replace('.conf', '') actionName = os.path.basename(actionConfig).replace('.conf', '')
actionReader = ActionReader(actionName, "TEST", {}, basedir=CONFIG_DIR) actionReader = ActionReader(actionName, "TEST", {}, basedir=CONFIG_DIR)
self.assertTrue(actionReader.read()) self.assertTrue(actionReader.read())
try: try:
actionReader.getOptions({}) # populate _opts actionReader.getOptions({}) # populate _opts
except Exception as e: # pragma: no cover except Exception as e: # pragma: no cover
self.fail("action %r\n%s: %s" % (actionName, type(e).__name__, e)) self.fail("action %r\n%s: %s" % (actionName, type(e).__name__, e))
if not actionName.endswith('-common'): if not actionName.endswith('-common'):
self.assertIn('Definition', actionReader.sections(), self.assertIn('Definition', actionReader.sections(),
msg="Action file %r is lacking [Definition] section" % actionConfig) msg="Action file %r is lacking [Definition] section" % actionConfig)
# all must have some actionban defined # all must have some actionban defined
self.assertTrue(actionReader._opts.get('actionban', '').strip(), self.assertTrue(actionReader._opts.get('actionban', '').strip(),
msg="Action file %r is lacking actionban" % actionConfig) msg="Action file %r is lacking actionban" % actionConfig)
# test name of jail is set in options (also if not supplied within parameters): # test name of jail is set in options (also if not supplied within parameters):
opts = actionReader.getCombined( opts = actionReader.getCombined(
ignore=CommandAction._escapedTags | set(('timeout', 'bantime'))) ignore=CommandAction._escapedTags | set(('timeout', 'bantime')))
self.assertEqual(opts.get('name'), 'TEST', self.assertEqual(opts.get('name'), 'TEST',
msg="Action file %r does not contains jail-name 'f2b-TEST'" % actionConfig) msg="Action file %r does not contains jail-name 'f2b-TEST'" % actionConfig)
# and the name is substituted (test several actions surely contains name-interpolation): # and the name is substituted (test several actions surely contains name-interpolation):
if actionName in ('pf', 'iptables-allports', 'iptables-multiport'): if actionName in ('pf', 'iptables-allports', 'iptables-multiport'):
#print('****', actionName, opts.get('actionstart', '')) #print('****', actionName, opts.get('actionstart', ''))
self.assertIn('f2b-TEST', opts.get('actionstart', ''), self.assertIn('f2b-TEST', opts.get('actionstart', ''),
msg="Action file %r: interpolation of actionstart does not contains jail-name 'f2b-TEST'" % actionConfig) msg="Action file %r: interpolation of actionstart does not contains jail-name 'f2b-TEST'" % actionConfig)
def testReadStockJailConf(self): def testReadStockJailConf(self):
jails = JailsReader(basedir=CONFIG_DIR, share_config=CONFIG_DIR_SHARE_CFG) # we are running tests from root project dir atm unittest.F2B.SkipIfCfgMissing(stock=True)
self.assertTrue(jails.read()) # opens fine jails = JailsReader(basedir=CONFIG_DIR, share_config=CONFIG_DIR_SHARE_CFG) # we are running tests from root project dir atm
self.assertTrue(jails.getOptions()) # reads fine self.assertTrue(jails.read()) # opens fine
comm_commands = jails.convert() self.assertTrue(jails.getOptions()) # reads fine
# by default None of the jails is enabled and we get no comm_commands = jails.convert()
# commands to communicate to the server # by default None of the jails is enabled and we get no
self.assertEqual(comm_commands, []) # commands to communicate to the server
self.assertEqual(comm_commands, [])
# TODO: make sure this is handled well # TODO: make sure this is handled well
## We should not "read" some bogus jail ## We should not "read" some bogus jail
#old_comm_commands = comm_commands[:] # make a copy #old_comm_commands = comm_commands[:] # make a copy
#self.assertRaises(ValueError, jails.getOptions, "BOGUS") #self.assertRaises(ValueError, jails.getOptions, "BOGUS")
#self.printLog() #self.printLog()
#self.assertLogged("No section: 'BOGUS'") #self.assertLogged("No section: 'BOGUS'")
## and there should be no side-effects ## and there should be no side-effects
#self.assertEqual(jails.convert(), old_comm_commands) #self.assertEqual(jails.convert(), old_comm_commands)
allFilters = set() allFilters = set()
# All jails must have filter and action set # All jails must have filter and action set
# TODO: evolve into a parametric test # TODO: evolve into a parametric test
for jail in jails.sections(): for jail in jails.sections():
if jail == 'INCLUDES': if jail == 'INCLUDES':
continue continue
filterName = jails.get(jail, 'filter') filterName = jails.get(jail, 'filter')
filterName, filterOpt = extractOptions(filterName) filterName, filterOpt = extractOptions(filterName)
allFilters.add(filterName) allFilters.add(filterName)
self.assertTrue(len(filterName)) self.assertTrue(len(filterName))
# moreover we must have a file for it # moreover we must have a file for it
# and it must be readable as a Filter # and it must be readable as a Filter
filterReader = FilterReader(filterName, jail, filterOpt, filterReader = FilterReader(filterName, jail, filterOpt,
share_config=CONFIG_DIR_SHARE_CFG, basedir=CONFIG_DIR)
self.assertTrue(filterReader.read(),"Failed to read filter:" + filterName) # opens fine
filterReader.getOptions({}) # reads fine
# test if filter has failregex set
self.assertTrue(filterReader._opts.get('failregex', '').strip())
actions = jails.get(jail, 'action')
self.assertTrue(len(actions.strip()))
# somewhat duplicating here what is done in JailsReader if
# the jail is enabled
for act in actions.split('\n'):
actName, actOpt = extractOptions(act)
self.assertTrue(len(actName))
self.assertTrue(isinstance(actOpt, dict))
if actName == 'iptables-multiport':
self.assertIn('port', actOpt)
actionReader = ActionReader(actName, jail, {},
share_config=CONFIG_DIR_SHARE_CFG, basedir=CONFIG_DIR) share_config=CONFIG_DIR_SHARE_CFG, basedir=CONFIG_DIR)
self.assertTrue(filterReader.read(),"Failed to read filter:" + filterName) # opens fine self.assertTrue(actionReader.read())
filterReader.getOptions({}) # reads fine actionReader.getOptions({}) # populate _opts
cmds = actionReader.convert()
self.assertTrue(len(cmds))
# test if filter has failregex set # all must have some actionban
self.assertTrue(filterReader._opts.get('failregex', '').strip()) self.assertTrue(actionReader._opts.get('actionban', '').strip())
actions = jails.get(jail, 'action') # Verify that all filters found under config/ have a jail
self.assertTrue(len(actions.strip())) def testReadStockJailFilterComplete(self):
unittest.F2B.SkipIfCfgMissing(stock=True)
jails = JailsReader(basedir=CONFIG_DIR, force_enable=True, share_config=CONFIG_DIR_SHARE_CFG)
self.assertTrue(jails.read()) # opens fine
self.assertTrue(jails.getOptions()) # reads fine
# grab all filter names
filters = set(os.path.splitext(os.path.split(a)[1])[0]
for a in glob.glob(os.path.join('config', 'filter.d', '*.conf'))
if not (a.endswith('common.conf') or a.endswith('-aggressive.conf')))
# get filters of all jails (filter names without options inside filter[...])
filters_jail = set(
extractOptions(jail.options['filter'])[0] for jail in jails.jails
)
self.maxDiff = None
self.assertTrue(filters.issubset(filters_jail),
"More filters exists than are referenced in stock jail.conf %r" % filters.difference(filters_jail))
self.assertTrue(filters_jail.issubset(filters),
"Stock jail.conf references non-existent filters %r" % filters_jail.difference(filters))
# somewhat duplicating here what is done in JailsReader if def testReadStockJailConfForceEnabled(self):
# the jail is enabled unittest.F2B.SkipIfCfgMissing(stock=True)
for act in actions.split('\n'): # more of a smoke test to make sure that no obvious surprises
actName, actOpt = extractOptions(act) # on users' systems when enabling shipped jails
self.assertTrue(len(actName)) jails = JailsReader(basedir=CONFIG_DIR, force_enable=True, share_config=CONFIG_DIR_SHARE_CFG) # we are running tests from root project dir atm
self.assertTrue(isinstance(actOpt, dict)) self.assertTrue(jails.read()) # opens fine
if actName == 'iptables-multiport': self.assertTrue(jails.getOptions()) # reads fine
self.assertIn('port', actOpt) comm_commands = jails.convert(allow_no_files=True)
actionReader = ActionReader(actName, jail, {}, # by default we have lots of jails ;)
share_config=CONFIG_DIR_SHARE_CFG, basedir=CONFIG_DIR) self.assertTrue(len(comm_commands))
self.assertTrue(actionReader.read())
actionReader.getOptions({}) # populate _opts
cmds = actionReader.convert()
self.assertTrue(len(cmds))
# all must have some actionban # some common sanity checks for commands
self.assertTrue(actionReader._opts.get('actionban', '').strip()) for command in comm_commands:
if len(command) >= 3 and [command[0], command[2]] == ['set', 'bantime']:
self.assertTrue(MyTime.str2seconds(command[3]) > 0)
# Verify that all filters found under config/ have a jail # and we know even some of them by heart
def testReadStockJailFilterComplete(self): for j in ['sshd', 'recidive']:
jails = JailsReader(basedir=CONFIG_DIR, force_enable=True, share_config=CONFIG_DIR_SHARE_CFG) # by default we have 'auto' backend ATM, but some distributions can overwrite it,
self.assertTrue(jails.read()) # opens fine # (e.g. fedora default is 'systemd') therefore let check it without backend...
self.assertTrue(jails.getOptions()) # reads fine self.assertIn(['add', j],
# grab all filter names (cmd[:2] for cmd in comm_commands if len(cmd) == 3 and cmd[0] == 'add'))
filters = set(os.path.splitext(os.path.split(a)[1])[0] # and warn on useDNS
for a in glob.glob(os.path.join('config', 'filter.d', '*.conf')) self.assertIn(['set', j, 'usedns', 'warn'], comm_commands)
if not (a.endswith('common.conf') or a.endswith('-aggressive.conf'))) self.assertIn(['start', j], comm_commands)
# get filters of all jails (filter names without options inside filter[...])
filters_jail = set(
extractOptions(jail.options['filter'])[0] for jail in jails.jails
)
self.maxDiff = None
self.assertTrue(filters.issubset(filters_jail),
"More filters exists than are referenced in stock jail.conf %r" % filters.difference(filters_jail))
self.assertTrue(filters_jail.issubset(filters),
"Stock jail.conf references non-existent filters %r" % filters_jail.difference(filters))
def testReadStockJailConfForceEnabled(self): # last commands should be the 'start' commands
# more of a smoke test to make sure that no obvious surprises self.assertEqual(comm_commands[-1][0], 'start')
# on users' systems when enabling shipped jails
jails = JailsReader(basedir=CONFIG_DIR, force_enable=True, share_config=CONFIG_DIR_SHARE_CFG) # we are running tests from root project dir atm
self.assertTrue(jails.read()) # opens fine
self.assertTrue(jails.getOptions()) # reads fine
comm_commands = jails.convert(allow_no_files=True)
# by default we have lots of jails ;) for j in jails._JailsReader__jails:
self.assertTrue(len(comm_commands)) actions = j._JailReader__actions
jail_name = j.getName()
# make sure that all of the jails have actions assigned,
# otherwise it makes little to no sense
self.assertTrue(len(actions),
msg="No actions found for jail %s" % jail_name)
# some common sanity checks for commands # Test for presence of blocktype (in relation to gh-232)
for command in comm_commands: for action in actions:
if len(command) >= 3 and [command[0], command[2]] == ['set', 'bantime']: commands = action.convert()
self.assertTrue(MyTime.str2seconds(command[3]) > 0) action_name = action.getName()
if '<blocktype>' in str(commands):
# Verify that it is among cInfo
self.assertIn('blocktype', action._initOpts)
# Verify that we have a call to set it up
blocktype_present = False
target_command = [jail_name, 'action', action_name]
for command in commands:
if (len(command) > 4 and command[0] == 'multi-set' and
command[1:4] == target_command):
blocktype_present = ('blocktype' in [cmd[0] for cmd in command[4]])
elif (len(command) > 5 and command[0] == 'set' and
command[1:4] == target_command and command[4] == 'blocktype'): # pragma: no cover - because of multi-set
blocktype_present = True
if blocktype_present:
break
self.assertTrue(
blocktype_present,
msg="Found no %s command among %s"
% (target_command, str(commands)) )
# and we know even some of them by heart def testStockConfigurator(self):
for j in ['sshd', 'recidive']: unittest.F2B.SkipIfCfgMissing(stock=True)
# by default we have 'auto' backend ATM, but some distributions can overwrite it, configurator = Configurator()
# (e.g. fedora default is 'systemd') therefore let check it without backend... configurator.setBaseDir(CONFIG_DIR)
self.assertIn(['add', j], self.assertEqual(configurator.getBaseDir(), CONFIG_DIR)
(cmd[:2] for cmd in comm_commands if len(cmd) == 3 and cmd[0] == 'add'))
# and warn on useDNS
self.assertIn(['set', j, 'usedns', 'warn'], comm_commands)
self.assertIn(['start', j], comm_commands)
# last commands should be the 'start' commands configurator.readEarly()
self.assertEqual(comm_commands[-1][0], 'start') opts = configurator.getEarlyOptions()
# our current default settings
self.assertEqual(opts['socket'], '/var/run/fail2ban/fail2ban.sock')
self.assertEqual(opts['pidfile'], '/var/run/fail2ban/fail2ban.pid')
for j in jails._JailsReader__jails: configurator.readAll()
actions = j._JailReader__actions configurator.getOptions()
jail_name = j.getName() configurator.convertToProtocol()
# make sure that all of the jails have actions assigned, commands = configurator.getConfigStream()
# otherwise it makes little to no sense
self.assertTrue(len(actions),
msg="No actions found for jail %s" % jail_name)
# Test for presence of blocktype (in relation to gh-232) # verify that dbfile comes before dbpurgeage
for action in actions: def find_set(option):
commands = action.convert() for i, e in enumerate(commands):
action_name = action.getName() if e[0] == 'set' and e[1] == option:
if '<blocktype>' in str(commands): return i
# Verify that it is among cInfo raise ValueError("Did not find command 'set %s' among commands %s"
self.assertIn('blocktype', action._initOpts) % (option, commands))
# Verify that we have a call to set it up
blocktype_present = False
target_command = [jail_name, 'action', action_name]
for command in commands:
if (len(command) > 4 and command[0] == 'multi-set' and
command[1:4] == target_command):
blocktype_present = ('blocktype' in [cmd[0] for cmd in command[4]])
elif (len(command) > 5 and command[0] == 'set' and
command[1:4] == target_command and command[4] == 'blocktype'): # pragma: no cover - because of multi-set
blocktype_present = True
if blocktype_present:
break
self.assertTrue(
blocktype_present,
msg="Found no %s command among %s"
% (target_command, str(commands)) )
def testStockConfigurator(self): # Set up of logging should come first
configurator = Configurator() self.assertEqual(find_set('syslogsocket'), 0)
configurator.setBaseDir(CONFIG_DIR) self.assertEqual(find_set('loglevel'), 1)
self.assertEqual(configurator.getBaseDir(), CONFIG_DIR) self.assertEqual(find_set('logtarget'), 2)
# then dbfile should be before dbpurgeage
self.assertTrue(find_set('dbpurgeage') > find_set('dbfile'))
configurator.readEarly() # and there is logging information left to be passed into the
opts = configurator.getEarlyOptions() # server
# our current default settings self.assertSortedEqual(commands,
self.assertEqual(opts['socket'], '/var/run/fail2ban/fail2ban.sock') [['set', 'dbfile',
self.assertEqual(opts['pidfile'], '/var/run/fail2ban/fail2ban.pid') '/var/lib/fail2ban/fail2ban.sqlite3'],
['set', 'dbpurgeage', '1d'],
['set', 'loglevel', "INFO"],
['set', 'logtarget', '/var/log/fail2ban.log'],
['set', 'syslogsocket', 'auto']])
configurator.readAll() # and if we force change configurator's fail2ban's baseDir
configurator.getOptions() # there should be an error message (test visually ;) --
configurator.convertToProtocol() # otherwise just a code smoke test)
commands = configurator.getConfigStream() configurator._Configurator__jails.setBaseDir('/tmp')
self.assertEqual(configurator._Configurator__jails.getBaseDir(), '/tmp')
# verify that dbfile comes before dbpurgeage self.assertEqual(configurator.getBaseDir(), CONFIG_DIR)
def find_set(option):
for i, e in enumerate(commands):
if e[0] == 'set' and e[1] == option:
return i
raise ValueError("Did not find command 'set %s' among commands %s"
% (option, commands))
# Set up of logging should come first
self.assertEqual(find_set('syslogsocket'), 0)
self.assertEqual(find_set('loglevel'), 1)
self.assertEqual(find_set('logtarget'), 2)
# then dbfile should be before dbpurgeage
self.assertTrue(find_set('dbpurgeage') > find_set('dbfile'))
# and there is logging information left to be passed into the
# server
self.assertSortedEqual(commands,
[['set', 'dbfile',
'/var/lib/fail2ban/fail2ban.sqlite3'],
['set', 'dbpurgeage', '1d'],
['set', 'loglevel', "INFO"],
['set', 'logtarget', '/var/log/fail2ban.log'],
['set', 'syslogsocket', 'auto']])
# and if we force change configurator's fail2ban's baseDir
# there should be an error message (test visually ;) --
# otherwise just a code smoke test)
configurator._Configurator__jails.setBaseDir('/tmp')
self.assertEqual(configurator._Configurator__jails.getBaseDir(), '/tmp')
self.assertEqual(configurator.getBaseDir(), CONFIG_DIR)
@with_tmpdir @with_tmpdir
def testMultipleSameAction(self, basedir): def testMultipleSameAction(self, basedir):

7
fail2ban/tests/fail2banclienttestcase.py
View File

@ -43,16 +43,14 @@ from .. import protocol
from ..server import server from ..server import server
from ..server.mytime import MyTime from ..server.mytime import MyTime
from ..server.utils import Utils from ..server.utils import Utils
from .utils import LogCaptureTestCase, logSys as DefLogSys, with_tmpdir, shutil, logging from .utils import LogCaptureTestCase, logSys as DefLogSys, with_tmpdir, shutil, logging, \
from .utils import CONFIG_DIR as STOCK_CONF_DIR STOCK, CONFIG_DIR as STOCK_CONF_DIR
from ..helpers import getLogger from ..helpers import getLogger
# Gets the instance of the logger. # Gets the instance of the logger.
logSys = getLogger(__name__) logSys = getLogger(__name__)
STOCK = exists(pjoin(STOCK_CONF_DIR, 'fail2ban.conf'))
CLIENT = "fail2ban-client" CLIENT = "fail2ban-client"
SERVER = "fail2ban-server" SERVER = "fail2ban-server"
BIN = dirname(Fail2banServer.getServerPath()) BIN = dirname(Fail2banServer.getServerPath())
@ -1171,6 +1169,7 @@ class Fail2banServerTest(Fail2banClientServerBase):
"Jail 'test-jail1' started", all=True) "Jail 'test-jail1' started", all=True)
# test action.d/nginx-block-map.conf -- # test action.d/nginx-block-map.conf --
@unittest.F2B.skip_if_cfg_missing(action="nginx-block-map")
@with_foreground_server_thread(startextra={ @with_foreground_server_thread(startextra={
# create log-file (avoid "not found" errors): # create log-file (avoid "not found" errors):
'create_before_start': ('%(tmp)s/blck-failures.log',), 'create_before_start': ('%(tmp)s/blck-failures.log',),

10
fail2ban/tests/filtertestcase.py
View File

@ -24,7 +24,6 @@ __license__ = "GPL"
from __builtin__ import open as fopen from __builtin__ import open as fopen
import unittest import unittest
import getpass
import os import os
import sys import sys
import time, datetime import time, datetime
@ -43,14 +42,12 @@ from ..server.failmanager import FailManagerEmpty
from ..server.ipdns import DNSUtils, IPAddr from ..server.ipdns import DNSUtils, IPAddr
from ..server.mytime import MyTime from ..server.mytime import MyTime
from ..server.utils import Utils, uni_decode from ..server.utils import Utils, uni_decode
from .utils import setUpMyTime, tearDownMyTime, mtimesleep, with_tmpdir, LogCaptureTestCase from .utils import setUpMyTime, tearDownMyTime, mtimesleep, with_tmpdir, LogCaptureTestCase, \
CONFIG_DIR as STOCK_CONF_DIR
from .dummyjail import DummyJail from .dummyjail import DummyJail
TEST_FILES_DIR = os.path.join(os.path.dirname(__file__), "files") TEST_FILES_DIR = os.path.join(os.path.dirname(__file__), "files")
STOCK_CONF_DIR = "config"
STOCK = os.path.exists(os.path.join(STOCK_CONF_DIR, 'fail2ban.conf'))
# yoh: per Steven Hiscocks's insight while troubleshooting # yoh: per Steven Hiscocks's insight while troubleshooting
# https://github.com/fail2ban/fail2ban/issues/103#issuecomment-15542836 # https://github.com/fail2ban/fail2ban/issues/103#issuecomment-15542836
@ -445,8 +442,7 @@ class IgnoreIPDNS(LogCaptureTestCase):
self.assertFalse(self.filter.inIgnoreIPList("128.178.222.70")) self.assertFalse(self.filter.inIgnoreIPList("128.178.222.70"))
def testIgnoreCmdApacheFakegooglebot(self): def testIgnoreCmdApacheFakegooglebot(self):
if not STOCK: # pragma: no cover unittest.F2B.SkipIfCfgMissing(stock=True)
raise unittest.SkipTest('Skip test because of no STOCK config')
cmd = os.path.join(STOCK_CONF_DIR, "filter.d/ignorecommands/apache-fakegooglebot") cmd = os.path.join(STOCK_CONF_DIR, "filter.d/ignorecommands/apache-fakegooglebot")
## below test direct as python module: ## below test direct as python module:
mod = Utils.load_python_module(cmd) mod = Utils.load_python_module(cmd)

1594
fail2ban/tests/servertestcase.py
View File

File diff suppressed because it is too large Load Diff

28
fail2ban/tests/utils.py
View File

@ -59,6 +59,9 @@ if not CONFIG_DIR:
else: else:
CONFIG_DIR = '/etc/fail2ban' CONFIG_DIR = '/etc/fail2ban'
# Indicates that we've stock config:
STOCK = os.path.exists(os.path.join(CONFIG_DIR, 'fail2ban.conf'))
# During the test cases (or setup) use fail2ban modules from main directory: # During the test cases (or setup) use fail2ban modules from main directory:
os.putenv('PYTHONPATH', os.path.dirname(os.path.dirname(os.path.dirname( os.putenv('PYTHONPATH', os.path.dirname(os.path.dirname(os.path.dirname(
os.path.abspath(__file__))))) os.path.abspath(__file__)))))
@ -187,6 +190,31 @@ class F2B(DefaultTestOptions):
pass pass
def SkipIfNoNetwork(self): def SkipIfNoNetwork(self):
pass pass
def SkipIfCfgMissing(self, **kwargs):
"""Helper to check action/filter config is available
"""
if not STOCK: # pragma: no cover
if kwargs.get('stock'):
raise unittest.SkipTest('Skip test because of missing stock-config files')
for t in ('action', 'filter'):
v = kwargs.get(t)
if v is None: continue
if os.path.splitext(v)[1] == '': v += '.conf'
if not os.path.exists(os.path.join(CONFIG_DIR, t+'.d', v)):
raise unittest.SkipTest('Skip test because of missing %s-config for %r' % (t, v))
def skip_if_cfg_missing(self, **decargs):
"""Helper decorator to check action/filter config is available
"""
def _deco_wrapper(f):
@wraps(f)
def wrapper(self, *args, **kwargs):
unittest.F2B.SkipIfCfgMissing(**decargs)
return f(self, *args, **kwargs)
return wrapper
return _deco_wrapper
def maxWaitTime(self,wtime): def maxWaitTime(self,wtime):
if self.fast: if self.fast:
wtime = float(wtime) / 10 wtime = float(wtime) / 10