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

change actioncheck behavior (on error only, gh-488)

pull/2607/head
sebres 2020-01-15 13:06:21 +01:00
parent 94b64055e1
commit ed4b5b8bdc
4 changed files with 95 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
fail2ban/server/action.py
View File

@ -407,7 +407,7 @@ class CommandAction(ActionBase):
cmd = self.replaceTag(tag, self._properties,
conditional=('family='+family if family else ''),
cache=self.__substCache)
if '<' not in cmd or not family: return cmd
if not family or '<' not in cmd: return cmd
# replace family as dynamic tags, important - don't cache, no recursion and auto-escape here:
cmd = self.replaceDynamicTags(cmd, {'family':family})
return cmd
@ -962,31 +962,38 @@ class CommandAction(ActionBase):
except (KeyError, TypeError):
family = ''
# invariant check:
if self.actioncheck:
# don't repair/restore if unban (no matter):
def _beforeRepair():
if cmd == '<actionunban>' and not self._properties.get('actionrepair_on_unban'):
self._logSys.error("Invariant check failed. Unban is impossible.")
repcnt = 0
while True:
# got some error, do invariant check:
if repcnt and self.actioncheck:
# don't repair/restore if unban (no matter):
def _beforeRepair():
if cmd == '<actionunban>' and not self._properties.get('actionrepair_on_unban'):
self._logSys.error("Invariant check failed. Unban is impossible.")
return False
return True
# check and repair if broken:
ret = self._invariantCheck(family, _beforeRepair, forceStart=(cmd != '<actionunban>'))
# if not sane (and not restored) return:
if ret != 1:
return False
return True
# check and repair if broken:
ret = self._invariantCheck(family, _beforeRepair, forceStart=(cmd != '<actionunban>'))
# if not sane (and not restored) return:
if ret != 1:
return False
# Replace static fields
realCmd = self.replaceTag(cmd, self._properties,
conditional=('family='+family if family else ''), cache=self.__substCache)
# Replace static fields
realCmd = self.replaceTag(cmd, self._properties,
conditional=('family='+family if family else ''), cache=self.__substCache)
# Replace dynamical tags, important - don't cache, no recursion and auto-escape here
if aInfo is not None:
realCmd = self.replaceDynamicTags(realCmd, aInfo)
else:
realCmd = cmd
# Replace dynamical tags, important - don't cache, no recursion and auto-escape here
if aInfo is not None:
realCmd = self.replaceDynamicTags(realCmd, aInfo)
else:
realCmd = cmd
return self.executeCmd(realCmd, self.timeout)
# try execute command:
ret = self.executeCmd(realCmd, self.timeout)
repcnt += 1
if ret or repcnt > 1:
return ret
@staticmethod
def executeCmd(realCmd, timeout=60, **kwargs):

6
fail2ban/tests/actionstestcase.py
View File

@ -215,6 +215,9 @@ class ExecuteActions(LogCaptureTestCase):
# flush for inet6 is intentionally "broken" here - test no unhandled except and invariant check:
act['actionflush?family=inet6'] = act.actionflush + '; exit 1'
act.actionstart_on_demand = True
# force errors via check in ban/unban:
act.actionban = "<actioncheck> ; " + act.actionban
act.actionunban = "<actioncheck> ; " + act.actionunban
self.__actions.start()
self.assertNotLogged("stdout: %r" % 'ip start')
@ -292,6 +295,9 @@ class ExecuteActions(LogCaptureTestCase):
act['actionflush?family=inet6'] = act.actionflush + '; exit 1'
act.actionstart_on_demand = True
act.actionrepair_on_unban = True
# force errors via check in ban/unban:
act.actionban = "<actioncheck> ; " + act.actionban
act.actionunban = "<actioncheck> ; " + act.actionunban
self.__actions.start()
self.assertNotLogged("stdout: %r" % 'ip start')

46
fail2ban/tests/actiontestcase.py
View File

@ -305,8 +305,8 @@ class CommandActionTest(LogCaptureTestCase):
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 = "echo -n"
self.assertEqual(self.__action.actionban, 'echo -n')
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"
@ -316,6 +316,7 @@ class CommandActionTest(LogCaptureTestCase):
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')
@ -365,12 +366,51 @@ class CommandActionTest(LogCaptureTestCase):
self.pruneLog('[phase 2]')
self.__action.actionstart = "touch '%s'" % tmp
self.__action.actionstop = "rm '%s'" % tmp
self.__action.actionban = """printf "%%%%b\n" <ip> >> '%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')
@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'

21
fail2ban/tests/servertestcase.py
View File

@ -2030,25 +2030,38 @@ class ServerConfigReaderTests(LogCaptureTestCase):
action.ban(aInfos['ipv4'])
if tests.get('ip4-start'): self.assertLogged(*tests.get('*-start', ())+tests['ip4-start'], all=True)
if tests.get('ip6-start'): self.assertNotLogged(*tests['ip6-start'], all=True)
self.assertLogged(*tests.get('ip4-check',())+tests['ip4-ban'], all=True)
self.assertLogged(*tests['ip4-ban'], all=True)
self.assertNotLogged(*tests['ip6'], all=True)
# test unban ip4 :
self.pruneLog('# === unban ipv4 ===')
action.unban(aInfos['ipv4'])
self.assertLogged(*tests.get('ip4-check',())+tests['ip4-unban'], all=True)
self.assertLogged(*tests['ip4-unban'], all=True)
self.assertNotLogged(*tests['ip6'], all=True)
# test ban ip6 :
self.pruneLog('# === ban ipv6 ===')
action.ban(aInfos['ipv6'])
if tests.get('ip6-start'): self.assertLogged(*tests.get('*-start', ())+tests['ip6-start'], all=True)
if tests.get('ip4-start'): self.assertNotLogged(*tests['ip4-start'], all=True)
self.assertLogged(*tests.get('ip6-check',())+tests['ip6-ban'], all=True)
self.assertLogged(*tests['ip6-ban'], all=True)
self.assertNotLogged(*tests['ip4'], all=True)
# test unban ip6 :
self.pruneLog('# === unban ipv6 ===')
action.unban(aInfos['ipv6'])
self.assertLogged(*tests.get('ip6-check',())+tests['ip6-unban'], all=True)
self.assertLogged(*tests['ip6-unban'], all=True)
self.assertNotLogged(*tests['ip4'], all=True)
# test invariant check (normally on demand in error case only):
if tests.get('ip4-check'):
self.pruneLog('# === check ipv4 ===')
action._invariantCheck(aInfos['ipv4']['family'])
self.assertLogged(*tests['ip4-check'], all=True)
if tests.get('ip6-check') and tests['ip6-check'] != tests['ip4-check']:
self.assertNotLogged(*tests['ip6-check'], all=True)
if tests.get('ip6-check'):
self.pruneLog('# === check ipv6 ===')
action._invariantCheck(aInfos['ipv6']['family'])
self.assertLogged(*tests['ip6-check'], all=True)
if tests.get('ip4-check') and tests['ip4-check'] != tests['ip6-check']:
self.assertNotLogged(*tests['ip4-check'], all=True)
# test flush for actions should supported this:
if tests.get('flush'):
self.pruneLog('# === flush ===')