diff options
Diffstat (limited to 'fail2ban/tests/actiontestcase.py')
-rw-r--r-- | fail2ban/tests/actiontestcase.py | 262 |
1 files changed, 170 insertions, 92 deletions
diff --git a/fail2ban/tests/actiontestcase.py b/fail2ban/tests/actiontestcase.py index 5f4aa460..ce5de483 100644 --- a/fail2ban/tests/actiontestcase.py +++ b/fail2ban/tests/actiontestcase.py @@ -34,8 +34,8 @@ from ..server.actions import OrderedDict, Actions from ..server.utils import Utils from .dummyjail import DummyJail -from .utils import LogCaptureTestCase -from .utils import pid_exists +from .utils import pid_exists, with_tmpdir, LogCaptureTestCase + class CommandActionTest(LogCaptureTestCase): @@ -75,61 +75,59 @@ class CommandActionTest(LogCaptureTestCase): lambda: substituteRecursiveTags({'A': 'to=<B> fromip=<IP>', 'C': '<B>', 'B': '<C>', 'D': ''})) self.assertRaises(ValueError, lambda: substituteRecursiveTags({'failregex': 'to=<honeypot> fromip=<IP>', 'sweet': '<honeypot>', 'honeypot': '<sweet>', 'ignoreregex': ''})) - # We need here an ordered, because the sequence of iteration is very important for this test - if OrderedDict: - # No cyclic recursion, just multiple replacement of tag <T>, should be successful: - self.assertEqual(substituteRecursiveTags( OrderedDict( - (('X', 'x=x<T>'), ('T', '1'), ('Z', '<X> <T> <Y>'), ('Y', 'y=y<T>'))) - ), {'X': 'x=x1', 'T': '1', 'Y': 'y=y1', 'Z': 'x=x1 1 y=y1'} - ) - # No cyclic recursion, just multiple replacement of tag <T> in composite tags, should be successful: - self.assertEqual(substituteRecursiveTags( OrderedDict( - (('X', 'x=x<T> <Z> <<R1>> <<R2>>'), ('R1', 'Z'), ('R2', 'Y'), ('T', '1'), ('Z', '<T> <Y>'), ('Y', 'y=y<T>'))) - ), {'X': 'x=x1 1 y=y1 1 y=y1 y=y1', 'R1': 'Z', 'R2': 'Y', 'T': '1', 'Z': '1 y=y1', 'Y': 'y=y1'} - ) - # No cyclic recursion, just multiple replacement of same tags, should be successful: - self.assertEqual(substituteRecursiveTags( OrderedDict(( - ('actionstart', 'ipset create <ipmset> hash:ip timeout <bantime> family <ipsetfamily>\n<iptables> -I <chain> <actiontype>'), - ('ipmset', 'f2b-<name>'), - ('name', 'any'), - ('bantime', '600'), - ('ipsetfamily', 'inet'), - ('iptables', 'iptables <lockingopt>'), - ('lockingopt', '-w'), - ('chain', 'INPUT'), - ('actiontype', '<multiport>'), - ('multiport', '-p <protocol> -m multiport --dports <port> -m set --match-set <ipmset> src -j <blocktype>'), - ('protocol', 'tcp'), - ('port', 'ssh'), - ('blocktype', 'REJECT',), - )) - ), OrderedDict(( - ('actionstart', 'ipset create f2b-any hash:ip timeout 600 family inet\niptables -w -I INPUT -p tcp -m multiport --dports ssh -m set --match-set f2b-any src -j REJECT'), - ('ipmset', 'f2b-any'), - ('name', 'any'), - ('bantime', '600'), - ('ipsetfamily', 'inet'), - ('iptables', 'iptables -w'), - ('lockingopt', '-w'), - ('chain', 'INPUT'), - ('actiontype', '-p tcp -m multiport --dports ssh -m set --match-set f2b-any src -j REJECT'), - ('multiport', '-p tcp -m multiport --dports ssh -m set --match-set f2b-any src -j REJECT'), - ('protocol', 'tcp'), - ('port', 'ssh'), - ('blocktype', 'REJECT') - )) - ) - # Cyclic recursion by composite tag creation, tags "create" another tag, that closes cycle: - self.assertRaises(ValueError, lambda: substituteRecursiveTags( OrderedDict(( - ('A', '<<B><C>>'), - ('B', 'D'), ('C', 'E'), - ('DE', 'cycle <A>'), - )) )) - self.assertRaises(ValueError, lambda: substituteRecursiveTags( OrderedDict(( - ('DE', 'cycle <A>'), - ('A', '<<B><C>>'), - ('B', 'D'), ('C', 'E'), - )) )) + # No cyclic recursion, just multiple replacement of tag <T>, should be successful: + self.assertEqual(substituteRecursiveTags( OrderedDict( + (('X', 'x=x<T>'), ('T', '1'), ('Z', '<X> <T> <Y>'), ('Y', 'y=y<T>'))) + ), {'X': 'x=x1', 'T': '1', 'Y': 'y=y1', 'Z': 'x=x1 1 y=y1'} + ) + # No cyclic recursion, just multiple replacement of tag <T> in composite tags, should be successful: + self.assertEqual(substituteRecursiveTags( OrderedDict( + (('X', 'x=x<T> <Z> <<R1>> <<R2>>'), ('R1', 'Z'), ('R2', 'Y'), ('T', '1'), ('Z', '<T> <Y>'), ('Y', 'y=y<T>'))) + ), {'X': 'x=x1 1 y=y1 1 y=y1 y=y1', 'R1': 'Z', 'R2': 'Y', 'T': '1', 'Z': '1 y=y1', 'Y': 'y=y1'} + ) + # No cyclic recursion, just multiple replacement of same tags, should be successful: + self.assertEqual(substituteRecursiveTags( OrderedDict(( + ('actionstart', 'ipset create <ipmset> hash:ip timeout <bantime> family <ipsetfamily>\n<iptables> -I <chain> <actiontype>'), + ('ipmset', 'f2b-<name>'), + ('name', 'any'), + ('bantime', '600'), + ('ipsetfamily', 'inet'), + ('iptables', 'iptables <lockingopt>'), + ('lockingopt', '-w'), + ('chain', 'INPUT'), + ('actiontype', '<multiport>'), + ('multiport', '-p <protocol> -m multiport --dports <port> -m set --match-set <ipmset> src -j <blocktype>'), + ('protocol', 'tcp'), + ('port', 'ssh'), + ('blocktype', 'REJECT',), + )) + ), OrderedDict(( + ('actionstart', 'ipset create f2b-any hash:ip timeout 600 family inet\niptables -w -I INPUT -p tcp -m multiport --dports ssh -m set --match-set f2b-any src -j REJECT'), + ('ipmset', 'f2b-any'), + ('name', 'any'), + ('bantime', '600'), + ('ipsetfamily', 'inet'), + ('iptables', 'iptables -w'), + ('lockingopt', '-w'), + ('chain', 'INPUT'), + ('actiontype', '-p tcp -m multiport --dports ssh -m set --match-set f2b-any src -j REJECT'), + ('multiport', '-p tcp -m multiport --dports ssh -m set --match-set f2b-any src -j REJECT'), + ('protocol', 'tcp'), + ('port', 'ssh'), + ('blocktype', 'REJECT') + )) + ) + # Cyclic recursion by composite tag creation, tags "create" another tag, that closes cycle: + self.assertRaises(ValueError, lambda: substituteRecursiveTags( OrderedDict(( + ('A', '<<B><C>>'), + ('B', 'D'), ('C', 'E'), + ('DE', 'cycle <A>'), + )) )) + self.assertRaises(ValueError, lambda: substituteRecursiveTags( OrderedDict(( + ('DE', 'cycle <A>'), + ('A', '<<B><C>>'), + ('B', 'D'), ('C', 'E'), + )) )) # missing tags are ok self.assertEqual(substituteRecursiveTags({'A': '<C>'}), {'A': '<C>'}) @@ -206,15 +204,15 @@ class CommandActionTest(LogCaptureTestCase): "Text 890 text 123 ABC") self.assertEqual( self.__action.replaceTag("<matches>", - {'matches': "some >char< should \< be[ escap}ed&\n"}), + {'matches': "some >char< should \\< be[ escap}ed&\n"}), "some \\>char\\< should \\\\\\< be\\[ escap\\}ed\\&\\n") self.assertEqual( self.__action.replaceTag("<ipmatches>", - {'ipmatches': "some >char< should \< be[ escap}ed&\n"}), + {'ipmatches': "some >char< should \\< be[ escap}ed&\n"}), "some \\>char\\< should \\\\\\< be\\[ escap\\}ed\\&\\n") self.assertEqual( self.__action.replaceTag("<ipjailmatches>", - {'ipjailmatches': "some >char< should \< be[ escap}ed&\r\n"}), + {'ipjailmatches': "some >char< should \\< be[ escap}ed&\r\n"}), "some \\>char\\< should \\\\\\< be\\[ escap\\}ed\\&\\r\\n") # Recursive @@ -252,7 +250,7 @@ class CommandActionTest(LogCaptureTestCase): delattr(self.__action, 'ac') # produce self-referencing query except: self.assertRaisesRegexp(ValueError, r"possible self referencing definitions in query", - lambda: self.__action.replaceTag("<x<x<x<x<x<x<x<x<x<x<x<x<x<x<x<x<x<x<x<x<x>>>>>>>>>>>>>>>>>>>>>", + lambda: self.__action.replaceTag("<x"*30+">"*30, self.__action._properties, conditional="family=inet6") ) @@ -297,61 +295,128 @@ class CommandActionTest(LogCaptureTestCase): "Text 000-567 text 567 '567'") self.assertTrue(len(cache) >= 3) - - def testExecuteActionBan(self): - self.__action.actionstart = "touch /tmp/fail2ban.test" - self.assertEqual(self.__action.actionstart, "touch /tmp/fail2ban.test") - self.__action.actionstop = "rm -f /tmp/fail2ban.test" - self.assertEqual(self.__action.actionstop, 'rm -f /tmp/fail2ban.test') - self.__action.actionban = "echo -n" - self.assertEqual(self.__action.actionban, 'echo -n') - self.__action.actioncheck = "[ -e /tmp/fail2ban.test ]" - self.assertEqual(self.__action.actioncheck, '[ -e /tmp/fail2ban.test ]') + @with_tmpdir + def testExecuteActionBan(self, tmp): + tmp += "/fail2ban.test" + self.__action.actionstart = "touch '%s'" % tmp + self.__action.actionrepair = self.__action.actionstart + self.assertEqual(self.__action.actionstart, "touch '%s'" % tmp) + self.__action.actionstop = "rm -f '%s'" % tmp + self.assertEqual(self.__action.actionstop, "rm -f '%s'" % tmp) + self.__action.actionban = "<actioncheck> && echo -n" + self.assertEqual(self.__action.actionban, "<actioncheck> && echo -n") + self.__action.actioncheck = "[ -e '%s' ]" % tmp + self.assertEqual(self.__action.actioncheck, "[ -e '%s' ]" % tmp) self.__action.actionunban = "true" self.assertEqual(self.__action.actionunban, 'true') + self.pruneLog() self.assertNotLogged('returned') # no action was actually executed yet + # start on demand is false, so it should cause failure on first attempt of ban: self.__action.ban({'ip': None}) self.assertLogged('Invariant check failed') self.assertLogged('returned successfully') + self.__action.stop() + self.assertLogged(self.__action.actionstop) def testExecuteActionEmptyUnban(self): + # unban will be executed for actions with banned items only: + self.__action.actionban = "" self.__action.actionunban = "" + self.__action.actionflush = "echo -n 'flush'" + self.__action.actionstop = "echo -n 'stop'" + self.__action.start(); + self.__action.ban({}); + self.pruneLog() self.__action.unban({}) - self.assertLogged('Nothing to do') + self.assertLogged('Nothing to do', wait=True) + # same as above but with interim flush, so no unban anymore: + self.__action.ban({}); + self.pruneLog('[phase 2]') + self.__action.flush() + self.__action.unban({}) + self.__action.stop() + self.assertLogged('stop', wait=True) + self.assertNotLogged('Nothing to do') - def testExecuteActionStartCtags(self): + @with_tmpdir + def testExecuteActionStartCtags(self, tmp): + tmp += '/fail2ban.test' self.__action.HOST = "192.0.2.0" - self.__action.actionstart = "touch /tmp/fail2ban.test.<HOST>" - self.__action.actionstop = "rm -f /tmp/fail2ban.test.<HOST>" - self.__action.actioncheck = "[ -e /tmp/fail2ban.test.192.0.2.0 ]" + self.__action.actionstart = "touch '%s.<HOST>'" % tmp + self.__action.actionstop = "rm -f '%s.<HOST>'" % tmp + self.__action.actioncheck = "[ -e '%s.192.0.2.0' ]" % tmp self.__action.start() + self.__action.consistencyCheck() - def testExecuteActionCheckRestoreEnvironment(self): + @with_tmpdir + def testExecuteActionCheckRestoreEnvironment(self, tmp): + tmp += '/fail2ban.test' self.__action.actionstart = "" - self.__action.actionstop = "rm -f /tmp/fail2ban.test" - self.__action.actionban = "rm /tmp/fail2ban.test" - self.__action.actioncheck = "[ -e /tmp/fail2ban.test ]" + self.__action.actionstop = "rm -f '%s'" % tmp + self.__action.actionban = "rm '%s'" % tmp + self.__action.actioncheck = "[ -e '%s' ]" % tmp self.assertRaises(RuntimeError, self.__action.ban, {'ip': None}) self.assertLogged('Invariant check failed', 'Unable to restore environment', all=True) # 2nd time, try to restore with producing error in stop, but succeeded start hereafter: self.pruneLog('[phase 2]') - self.__action.actionstart = "touch /tmp/fail2ban.test" - self.__action.actionstop = "rm /tmp/fail2ban.test" - self.__action.actionban = 'printf "%%b\n" <ip> >> /tmp/fail2ban.test' - self.__action.actioncheck = "[ -e /tmp/fail2ban.test ]" + self.__action.actionstart = "touch '%s'" % tmp + self.__action.actionstop = "rm '%s'" % tmp + self.__action.actionban = """<actioncheck> && printf "%%%%b\n" <ip> >> '%s'""" % tmp + self.__action.actioncheck = "[ -e '%s' ]" % tmp self.__action.ban({'ip': None}) self.assertLogged('Invariant check failed') self.assertNotLogged('Unable to restore environment') - def testExecuteActionCheckRepairEnvironment(self): + @with_tmpdir + def testExecuteActionCheckOnBanFailure(self, tmp): + tmp += '/fail2ban.test' + self.__action.actionstart = "touch '%s'; echo 'started ...'" % tmp + self.__action.actionstop = "rm -f '%s'" % tmp + self.__action.actionban = "[ -e '%s' ] && echo 'banned '<ip>" % tmp + self.__action.actioncheck = "[ -e '%s' ] && echo 'check ok' || { echo 'check failed'; exit 1; }" % tmp + self.__action.actionrepair = "echo 'repair ...'; touch '%s'" % tmp + self.__action.actionstart_on_demand = False + self.__action.start() + # phase 1: with repair; + # phase 2: without repair (start/stop), not on demand; + # phase 3: without repair (start/stop), start on demand. + for i in (1, 2, 3): + self.pruneLog('[phase %s]' % i) + # 1st time with success ban: + self.__action.ban({'ip': '192.0.2.1'}) + self.assertLogged( + "stdout: %r" % 'banned 192.0.2.1', all=True) + self.assertNotLogged("Invariant check failed. Trying", + "stdout: %r" % 'check failed', + "stdout: %r" % ('repair ...' if self.__action.actionrepair else 'started ...'), + "stdout: %r" % 'check ok', all=True) + # force error in ban: + os.remove(tmp) + self.pruneLog() + # 2nd time with fail recognition, success repair, check and ban: + self.__action.ban({'ip': '192.0.2.2'}) + self.assertLogged("Invariant check failed. Trying", + "stdout: %r" % 'check failed', + "stdout: %r" % ('repair ...' if self.__action.actionrepair else 'started ...'), + "stdout: %r" % 'check ok', + "stdout: %r" % 'banned 192.0.2.2', all=True) + # repeat without repair (stop/start), herafter enable on demand: + if self.__action.actionrepair: + self.__action.actionrepair = "" + elif not self.__action.actionstart_on_demand: + self.__action.actionstart_on_demand = True + + @with_tmpdir + def testExecuteActionCheckRepairEnvironment(self, tmp): + tmp += '/fail2ban.test' self.__action.actionstart = "" self.__action.actionstop = "" - self.__action.actionban = "rm /tmp/fail2ban.test" - self.__action.actioncheck = "[ -e /tmp/fail2ban.test ]" - self.__action.actionrepair = "echo 'repair ...'; touch /tmp/fail2ban.test" + self.__action.actionban = "rm '%s'" % tmp + self.__action.actioncheck = "[ -e '%s' ]" % tmp + self.__action.actionrepair = "echo 'repair ...'; touch '%s'" % tmp # 1st time with success repair: self.__action.ban({'ip': None}) self.assertLogged("Invariant check failed. Trying", "echo 'repair ...'", all=True) @@ -379,13 +444,13 @@ class CommandActionTest(LogCaptureTestCase): 'user': "tester" } }) - self.__action.actionban = "touch /tmp/fail2ban.test.123; echo 'failure <F-ID> of <F-USER> -<F-TEST>- from <ip>:<F-PORT>'" - self.__action.actionunban = "rm /tmp/fail2ban.test.<ABC>; echo 'user <F-USER> unbanned'" + self.__action.actionban = "echo '<ABC>, failure <F-ID> of <F-USER> -<F-TEST>- from <ip>:<F-PORT>'" + self.__action.actionunban = "echo '<ABC>, user <F-USER> unbanned'" self.__action.ban(aInfo) self.__action.unban(aInfo) self.assertLogged( - " -- stdout: 'failure 111 of tester -- from 192.0.2.1:222'", - " -- stdout: 'user tester unbanned'", + " -- stdout: '123, failure 111 of tester -- from 192.0.2.1:222'", + " -- stdout: '123, user tester unbanned'", all=True ) @@ -560,6 +625,19 @@ class CommandActionTest(LogCaptureTestCase): self.assertEqual(len(m), 3) self.assertIn('c', m) self.assertEqual((m['a'], m['b'], m['c']), (5, 11, 'test')) + # immutability of copy: + m['d'] = 'dddd' + m2 = m.copy() + m2['c'] = lambda self: self['a'] + 7 + m2['a'] = 1 + del m2['b'] + del m2['d'] + self.assertTrue('b' in m) + self.assertTrue('d' in m) + self.assertFalse('b' in m2) + self.assertFalse('d' in m2) + self.assertEqual((m['a'], m['b'], m['c'], m['d']), (5, 11, 'test', 'dddd')) + self.assertEqual((m2['a'], m2['c']), (1, 8)) def testCallingMapRep(self): m = CallingMap({ |