mirror of https://github.com/fail2ban/fail2ban
ENH: first working unittest for checking polling and inotify backends
Yaroslav Halchenko
parent
baa09098f0
commit
60260bce3d
|
|
@ -64,7 +64,11 @@ def _assert_correct_last_attempt(utest, filter_, output):
|
|||
|
||||
Test filter to contain target ticket
|
||||
"""
|
||||
ticket = filter_.failManager.toBan()
|
||||
if isinstance(filter_, DummyJail):
|
||||
ticket = filter_.getFailTicket()
|
||||
else:
|
||||
# when we are testing without jails
|
||||
ticket = filter_.failManager.toBan()
|
||||
|
||||
attempts = ticket.getAttempt()
|
||||
date = ticket.getTime()
|
||||
|
|
@ -184,11 +188,7 @@ class LogFileMonitor(unittest.TestCase):
|
|||
# shorter wait time for not modified status
|
||||
return not self.isModified(0.4)
|
||||
|
||||
def _testNewChangeViaIsModified(self):
|
||||
if not hasattr(self.filter, 'isModified'):
|
||||
raise unittest.SkipTest(
|
||||
"%s does not have isModified (present only in poll atm"
|
||||
% (self.filter,))
|
||||
def testNewChangeViaIsModified(self):
|
||||
# it is a brand new one -- so first we think it is modified
|
||||
self.assertTrue(self.isModified())
|
||||
# but not any longer
|
||||
|
|
@ -265,6 +265,38 @@ class LogFileMonitor(unittest.TestCase):
|
|||
self.assertEqual(self.filter.failManager.getFailTotal(), 3)
|
||||
|
||||
|
||||
from threading import Lock
|
||||
class DummyJail(object):
|
||||
"""A simple 'jail' to suck in all the tickets generated by Filter's
|
||||
"""
|
||||
def __init__(self):
|
||||
self.lock = Lock()
|
||||
self.queue = []
|
||||
|
||||
def __len__(self):
|
||||
try:
|
||||
self.lock.acquire()
|
||||
return len(self.queue)
|
||||
finally:
|
||||
self.lock.release()
|
||||
|
||||
def putFailTicket(self, ticket):
|
||||
try:
|
||||
self.lock.acquire()
|
||||
self.queue.append(ticket)
|
||||
finally:
|
||||
self.lock.release()
|
||||
|
||||
def getFailTicket(self):
|
||||
try:
|
||||
self.lock.acquire()
|
||||
return self.queue.pop()
|
||||
finally:
|
||||
self.lock.release()
|
||||
|
||||
def getName(self):
|
||||
return "DummyJail #%s with %d tickets" % (id(self), len(self))
|
||||
|
||||
def get_monitor_failures_testcase(Filter_):
|
||||
"""Generator of TestCase's for different filters/backends
|
||||
"""
|
||||
|
|
@ -275,81 +307,77 @@ def get_monitor_failures_testcase(Filter_):
|
|||
self.filter = self.name = 'NA'
|
||||
_, self.name = tempfile.mkstemp('fail2ban', 'monitorfailures')
|
||||
self.file = open(self.name, 'a')
|
||||
self.filter = Filter_(None)
|
||||
self.jail = DummyJail()
|
||||
self.filter = Filter_(self.jail)
|
||||
self.filter.addLogPath(self.name)
|
||||
self.filter.setActive(True)
|
||||
self.filter.addFailRegex("(?:(?:Authentication failure|Failed [-/\w+]+) for(?: [iI](?:llegal|nvalid) user)?|[Ii](?:llegal|nvalid) user|ROOT LOGIN REFUSED) .*(?: from|FROM) <HOST>")
|
||||
self.filter.start()
|
||||
#print "D: started filter %s" % self.filter
|
||||
|
||||
|
||||
def tearDown(self):
|
||||
_killfile(self.file, self.name)
|
||||
#print "D: SLEEPING A BIT"
|
||||
#import time; time.sleep(5)
|
||||
#print "D: TEARING DOWN"
|
||||
self.filter.stop()
|
||||
#print "D: WAITING FOR FILTER TO STOP"
|
||||
self.filter.join() # wait for the thread to terminate
|
||||
#print "D: KILLING THE FILE"
|
||||
#_killfile(self.file, self.name)
|
||||
pass
|
||||
|
||||
def __str__(self):
|
||||
return "MonitorFailures%s(%s)" \
|
||||
% (Filter_, hasattr(self, 'name') and self.name or 'tempfile')
|
||||
|
||||
def isModified(self, delay=2.):
|
||||
def isFilled(self, delay=2.):
|
||||
"""Wait up to `delay` sec to assure that it was modified or not
|
||||
"""
|
||||
time0 = time.time()
|
||||
while time.time() < time0 + delay:
|
||||
if self.filter.isModified(self.name):
|
||||
if len(self.jail):
|
||||
return True
|
||||
time.sleep(0.1)
|
||||
return False
|
||||
|
||||
def notModified(self):
|
||||
def isEmpty(self):
|
||||
# shorter wait time for not modified status
|
||||
return not self.isModified(0.4)
|
||||
|
||||
def _testNewChangeViaIsModified(self):
|
||||
if not hasattr(self.filter, 'isModified'):
|
||||
raise unittest.SkipTest(
|
||||
"%s does not have isModified (present only in poll atm"
|
||||
% (self.filter,))
|
||||
# it is a brand new one -- so first we think it is modified
|
||||
self.assertTrue(self.isModified())
|
||||
# but not any longer
|
||||
self.assertTrue(self.notModified())
|
||||
self.assertTrue(self.notModified())
|
||||
for i in range(4): # few changes
|
||||
# unless we write into it
|
||||
self.file.write("line%d\n" % i)
|
||||
self.file.flush()
|
||||
self.assertTrue(self.isModified())
|
||||
self.assertTrue(self.notModified())
|
||||
os.rename(self.name, self.name + '.old')
|
||||
# we are not signaling as modified whenever
|
||||
# it gets away
|
||||
self.assertTrue(self.notModified())
|
||||
f = open(self.name, 'a')
|
||||
self.assertTrue(self.isModified())
|
||||
self.assertTrue(self.notModified())
|
||||
f.write("line%d\n" % i)
|
||||
f.flush()
|
||||
self.assertTrue(self.isModified())
|
||||
self.assertTrue(self.notModified())
|
||||
_killfile(f, self.name)
|
||||
_killfile(self.name, self.name + '.old')
|
||||
pass
|
||||
return not self.isFilled(0.4)
|
||||
|
||||
def testNewChangeViaGetFailures_simple(self):
|
||||
# suck in lines from this sample log file
|
||||
self.filter.getFailures(self.name)
|
||||
#self.filter.getFailures(self.name)
|
||||
self.assertRaises(FailManagerEmpty, self.filter.failManager.toBan)
|
||||
|
||||
# Now let's feed it with entries from the file
|
||||
_copy_lines_between_files(GetFailures.FILENAME_01, self.file, n=5)
|
||||
self.filter.getFailures(self.name)
|
||||
self.assertRaises(FailManagerEmpty, self.filter.failManager.toBan)
|
||||
# and it should have not been enough
|
||||
# and our dummy jail is empty as well
|
||||
self.assertFalse(len(self.jail))
|
||||
# since it should have not been enough
|
||||
|
||||
_copy_lines_between_files(GetFailures.FILENAME_01, self.file, skip=5)
|
||||
self.filter.getFailures(self.name)
|
||||
_assert_correct_last_attempt(self, self.filter, GetFailures.FAILURES_01)
|
||||
self.isFilled(4)
|
||||
# so we sleep for up to 2 sec for it not to become empty,
|
||||
# and meanwhile pass to other thread(s) and filter should
|
||||
# have gathered new failures and passed them into the
|
||||
# DummyJail
|
||||
self.assertEqual(len(self.jail), 1)
|
||||
# and there should be no "stuck" ticket in failManager
|
||||
self.assertRaises(FailManagerEmpty, self.filter.failManager.toBan)
|
||||
_assert_correct_last_attempt(self, self.jail, GetFailures.FAILURES_01)
|
||||
self.assertEqual(len(self.jail), 0)
|
||||
|
||||
def testNewChangeViaGetFailures_rewrite(self):
|
||||
#return
|
||||
# just for fun let's copy all of them again and see if that results
|
||||
# in a new ban
|
||||
_copy_lines_between_files(GetFailures.FILENAME_01, self.file, n=100)
|
||||
self.isFilled(4)
|
||||
_assert_correct_last_attempt(self, self.jail, GetFailures.FAILURES_01)
|
||||
|
||||
|
||||
def _testNewChangeViaGetFailures_rewrite(self):
|
||||
#
|
||||
# if we rewrite the file at once
|
||||
self.file.close()
|
||||
|
|
@ -367,7 +395,7 @@ def get_monitor_failures_testcase(Filter_):
|
|||
#self.assertRaises(FailManagerEmpty, self.filter.failManager.toBan)
|
||||
_assert_correct_last_attempt(self, self.filter, GetFailures.FAILURES_01)
|
||||
|
||||
def testNewChangeViaGetFailures_move(self):
|
||||
def _testNewChangeViaGetFailures_move(self):
|
||||
#
|
||||
# if we move file into a new location while it has been open already
|
||||
self.file = _copy_lines_between_files(GetFailures.FILENAME_01, self.name,
|
||||
|
|
|
|||
Loading…
Reference in New Issue