mirror of https://github.com/fail2ban/fail2ban
sebres
4 years ago
parent
e3d43d1241
commit
6f4b6ec8cc
@ -1,19 +0,0 @@
|
|||||||
# Fail2ban reporting to badips.com
|
|
||||||
#
|
|
||||||
# Note: This reports an IP only and does not actually ban traffic. Use
|
|
||||||
# another action in the same jail if you want bans to occur.
|
|
||||||
#
|
|
||||||
# Set the category to the appropriate value before use.
|
|
||||||
#
|
|
||||||
# To get see register and optional key to get personalised graphs see:
|
|
||||||
# http://www.badips.com/blog/personalized-statistics-track-the-attackers-of-all-your-servers-with-one-key
|
|
||||||
|
|
||||||
[Definition]
|
|
||||||
|
|
||||||
actionban = curl --fail --user-agent "<agent>" http://www.badips.com/add/<category>/<ip>
|
|
||||||
|
|
||||||
[Init]
|
|
||||||
|
|
||||||
# Option: category
|
|
||||||
# Notes.: Values are from the list here: http://www.badips.com/get/categories
|
|
||||||
category =
|
|
||||||
@ -1,391 +0,0 @@
|
|||||||
# emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: t -*-
|
|
||||||
# vi: set ft=python sts=4 ts=4 sw=4 noet :
|
|
||||||
|
|
||||||
# This file is part of Fail2Ban.
|
|
||||||
#
|
|
||||||
# Fail2Ban is free software; you can redistribute it and/or modify
|
|
||||||
# it under the terms of the GNU General Public License as published by
|
|
||||||
# the Free Software Foundation; either version 2 of the License, or
|
|
||||||
# (at your option) any later version.
|
|
||||||
#
|
|
||||||
# Fail2Ban is distributed in the hope that it will be useful,
|
|
||||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
||||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
||||||
# GNU General Public License for more details.
|
|
||||||
#
|
|
||||||
# You should have received a copy of the GNU General Public License
|
|
||||||
# along with Fail2Ban; if not, write to the Free Software
|
|
||||||
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
|
|
||||||
|
|
||||||
import sys
|
|
||||||
if sys.version_info < (2, 7): # pragma: no cover
|
|
||||||
raise ImportError("badips.py action requires Python >= 2.7")
|
|
||||||
import json
|
|
||||||
import threading
|
|
||||||
import logging
|
|
||||||
if sys.version_info >= (3, ): # pragma: 2.x no cover
|
|
||||||
from urllib.request import Request, urlopen
|
|
||||||
from urllib.parse import urlencode
|
|
||||||
from urllib.error import HTTPError
|
|
||||||
else: # pragma: 3.x no cover
|
|
||||||
from urllib2 import Request, urlopen, HTTPError
|
|
||||||
from urllib import urlencode
|
|
||||||
|
|
||||||
from fail2ban.server.actions import Actions, ActionBase, BanTicket
|
|
||||||
from fail2ban.helpers import splitwords, str2LogLevel
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
class BadIPsAction(ActionBase): # pragma: no cover - may be unavailable
|
|
||||||
"""Fail2Ban action which reports bans to badips.com, and also
|
|
||||||
blacklist bad IPs listed on badips.com by using another action's
|
|
||||||
ban method.
|
|
||||||
|
|
||||||
Parameters
|
|
||||||
----------
|
|
||||||
jail : Jail
|
|
||||||
The jail which the action belongs to.
|
|
||||||
name : str
|
|
||||||
Name assigned to the action.
|
|
||||||
category : str
|
|
||||||
Valid badips.com category for reporting failures.
|
|
||||||
score : int, optional
|
|
||||||
Minimum score for bad IPs. Default 3.
|
|
||||||
age : str, optional
|
|
||||||
Age of last report for bad IPs, per badips.com syntax.
|
|
||||||
Default "24h" (24 hours)
|
|
||||||
banaction : str, optional
|
|
||||||
Name of banaction to use for blacklisting bad IPs. If `None`,
|
|
||||||
no blacklist of IPs will take place.
|
|
||||||
Default `None`.
|
|
||||||
bancategory : str, optional
|
|
||||||
Name of category to use for blacklisting, which can differ
|
|
||||||
from category used for reporting. e.g. may want to report
|
|
||||||
"postfix", but want to use whole "mail" category for blacklist.
|
|
||||||
Default `category`.
|
|
||||||
bankey : str, optional
|
|
||||||
Key issued by badips.com to retrieve personal list
|
|
||||||
of blacklist IPs.
|
|
||||||
updateperiod : int, optional
|
|
||||||
Time in seconds between updating bad IPs blacklist.
|
|
||||||
Default 900 (15 minutes)
|
|
||||||
loglevel : int/str, optional
|
|
||||||
Log level of the message when an IP is (un)banned.
|
|
||||||
Default `DEBUG`.
|
|
||||||
Can be also supplied as two-value list (comma- or space separated) to
|
|
||||||
provide level of the summary message when a group of IPs is (un)banned.
|
|
||||||
Example `DEBUG,INFO`.
|
|
||||||
agent : str, optional
|
|
||||||
User agent transmitted to server.
|
|
||||||
Default `Fail2Ban/ver.`
|
|
||||||
|
|
||||||
Raises
|
|
||||||
------
|
|
||||||
ValueError
|
|
||||||
If invalid `category`, `score`, `banaction` or `updateperiod`.
|
|
||||||
"""
|
|
||||||
|
|
||||||
TIMEOUT = 10
|
|
||||||
_badips = "https://www.badips.com"
|
|
||||||
def _Request(self, url, **argv):
|
|
||||||
return Request(url, headers={'User-Agent': self.agent}, **argv)
|
|
||||||
|
|
||||||
def __init__(self, jail, name, category, score=3, age="24h",
|
|
||||||
banaction=None, bancategory=None, bankey=None, updateperiod=900,
|
|
||||||
loglevel='DEBUG', agent="Fail2Ban", timeout=TIMEOUT):
|
|
||||||
super(BadIPsAction, self).__init__(jail, name)
|
|
||||||
|
|
||||||
self.timeout = timeout
|
|
||||||
self.agent = agent
|
|
||||||
self.category = category
|
|
||||||
self.score = score
|
|
||||||
self.age = age
|
|
||||||
self.banaction = banaction
|
|
||||||
self.bancategory = bancategory or category
|
|
||||||
self.bankey = bankey
|
|
||||||
loglevel = splitwords(loglevel)
|
|
||||||
self.sumloglevel = str2LogLevel(loglevel[-1])
|
|
||||||
self.loglevel = str2LogLevel(loglevel[0])
|
|
||||||
self.updateperiod = updateperiod
|
|
||||||
|
|
||||||
self._bannedips = set()
|
|
||||||
# Used later for threading.Timer for updating badips
|
|
||||||
self._timer = None
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def isAvailable(timeout=1):
|
|
||||||
try:
|
|
||||||
response = urlopen(Request("/".join([BadIPsAction._badips]),
|
|
||||||
headers={'User-Agent': "Fail2Ban"}), timeout=timeout)
|
|
||||||
return True, ''
|
|
||||||
except Exception as e: # pragma: no cover
|
|
||||||
return False, e
|
|
||||||
|
|
||||||
def logError(self, response, what=''): # pragma: no cover - sporadical (502: Bad Gateway, etc)
|
|
||||||
messages = {}
|
|
||||||
try:
|
|
||||||
messages = json.loads(response.read().decode('utf-8'))
|
|
||||||
except:
|
|
||||||
pass
|
|
||||||
self._logSys.error(
|
|
||||||
"%s. badips.com response: '%s'", what,
|
|
||||||
messages.get('err', 'Unknown'))
|
|
||||||
|
|
||||||
def getCategories(self, incParents=False):
|
|
||||||
"""Get badips.com categories.
|
|
||||||
|
|
||||||
Returns
|
|
||||||
-------
|
|
||||||
set
|
|
||||||
Set of categories.
|
|
||||||
|
|
||||||
Raises
|
|
||||||
------
|
|
||||||
HTTPError
|
|
||||||
Any issues with badips.com request.
|
|
||||||
ValueError
|
|
||||||
If badips.com response didn't contain necessary information
|
|
||||||
"""
|
|
||||||
try:
|
|
||||||
response = urlopen(
|
|
||||||
self._Request("/".join([self._badips, "get", "categories"])), timeout=self.timeout)
|
|
||||||
except HTTPError as response: # pragma: no cover
|
|
||||||
self.logError(response, "Failed to fetch categories")
|
|
||||||
raise
|
|
||||||
else:
|
|
||||||
response_json = json.loads(response.read().decode('utf-8'))
|
|
||||||
if not 'categories' in response_json:
|
|
||||||
err = "badips.com response lacked categories specification. Response was: %s" \
|
|
||||||
% (response_json,)
|
|
||||||
self._logSys.error(err)
|
|
||||||
raise ValueError(err)
|
|
||||||
categories = response_json['categories']
|
|
||||||
categories_names = set(
|
|
||||||
value['Name'] for value in categories)
|
|
||||||
if incParents:
|
|
||||||
categories_names.update(set(
|
|
||||||
value['Parent'] for value in categories
|
|
||||||
if "Parent" in value))
|
|
||||||
return categories_names
|
|
||||||
|
|
||||||
def getList(self, category, score, age, key=None):
|
|
||||||
"""Get badips.com list of bad IPs.
|
|
||||||
|
|
||||||
Parameters
|
|
||||||
----------
|
|
||||||
category : str
|
|
||||||
Valid badips.com category.
|
|
||||||
score : int
|
|
||||||
Minimum score for bad IPs.
|
|
||||||
age : str
|
|
||||||
Age of last report for bad IPs, per badips.com syntax.
|
|
||||||
key : str, optional
|
|
||||||
Key issued by badips.com to fetch IPs reported with the
|
|
||||||
associated key.
|
|
||||||
|
|
||||||
Returns
|
|
||||||
-------
|
|
||||||
set
|
|
||||||
Set of bad IPs.
|
|
||||||
|
|
||||||
Raises
|
|
||||||
------
|
|
||||||
HTTPError
|
|
||||||
Any issues with badips.com request.
|
|
||||||
"""
|
|
||||||
try:
|
|
||||||
url = "?".join([
|
|
||||||
"/".join([self._badips, "get", "list", category, str(score)]),
|
|
||||||
urlencode({'age': age})])
|
|
||||||
if key:
|
|
||||||
url = "&".join([url, urlencode({'key': key})])
|
|
||||||
self._logSys.debug('badips.com: get list, url: %r', url)
|
|
||||||
response = urlopen(self._Request(url), timeout=self.timeout)
|
|
||||||
except HTTPError as response: # pragma: no cover
|
|
||||||
self.logError(response, "Failed to fetch bad IP list")
|
|
||||||
raise
|
|
||||||
else:
|
|
||||||
return set(response.read().decode('utf-8').split())
|
|
||||||
|
|
||||||
@property
|
|
||||||
def category(self):
|
|
||||||
"""badips.com category for reporting IPs.
|
|
||||||
"""
|
|
||||||
return self._category
|
|
||||||
|
|
||||||
@category.setter
|
|
||||||
def category(self, category):
|
|
||||||
if category not in self.getCategories():
|
|
||||||
self._logSys.error("Category name '%s' not valid. "
|
|
||||||
"see badips.com for list of valid categories",
|
|
||||||
category)
|
|
||||||
raise ValueError("Invalid category: %s" % category)
|
|
||||||
self._category = category
|
|
||||||
|
|
||||||
@property
|
|
||||||
def bancategory(self):
|
|
||||||
"""badips.com bancategory for fetching IPs.
|
|
||||||
"""
|
|
||||||
return self._bancategory
|
|
||||||
|
|
||||||
@bancategory.setter
|
|
||||||
def bancategory(self, bancategory):
|
|
||||||
if bancategory != "any" and bancategory not in self.getCategories(incParents=True):
|
|
||||||
self._logSys.error("Category name '%s' not valid. "
|
|
||||||
"see badips.com for list of valid categories",
|
|
||||||
bancategory)
|
|
||||||
raise ValueError("Invalid bancategory: %s" % bancategory)
|
|
||||||
self._bancategory = bancategory
|
|
||||||
|
|
||||||
@property
|
|
||||||
def score(self):
|
|
||||||
"""badips.com minimum score for fetching IPs.
|
|
||||||
"""
|
|
||||||
return self._score
|
|
||||||
|
|
||||||
@score.setter
|
|
||||||
def score(self, score):
|
|
||||||
score = int(score)
|
|
||||||
if 0 <= score <= 5:
|
|
||||||
self._score = score
|
|
||||||
else:
|
|
||||||
raise ValueError("Score must be 0-5")
|
|
||||||
|
|
||||||
@property
|
|
||||||
def banaction(self):
|
|
||||||
"""Jail action to use for banning/unbanning.
|
|
||||||
"""
|
|
||||||
return self._banaction
|
|
||||||
|
|
||||||
@banaction.setter
|
|
||||||
def banaction(self, banaction):
|
|
||||||
if banaction is not None and banaction not in self._jail.actions:
|
|
||||||
self._logSys.error("Action name '%s' not in jail '%s'",
|
|
||||||
banaction, self._jail.name)
|
|
||||||
raise ValueError("Invalid banaction")
|
|
||||||
self._banaction = banaction
|
|
||||||
|
|
||||||
@property
|
|
||||||
def updateperiod(self):
|
|
||||||
"""Period in seconds between banned bad IPs will be updated.
|
|
||||||
"""
|
|
||||||
return self._updateperiod
|
|
||||||
|
|
||||||
@updateperiod.setter
|
|
||||||
def updateperiod(self, updateperiod):
|
|
||||||
updateperiod = int(updateperiod)
|
|
||||||
if updateperiod > 0:
|
|
||||||
self._updateperiod = updateperiod
|
|
||||||
else:
|
|
||||||
raise ValueError("Update period must be integer greater than 0")
|
|
||||||
|
|
||||||
def _banIPs(self, ips):
|
|
||||||
for ip in ips:
|
|
||||||
try:
|
|
||||||
ai = Actions.ActionInfo(BanTicket(ip), self._jail)
|
|
||||||
self._jail.actions[self.banaction].ban(ai)
|
|
||||||
except Exception as e:
|
|
||||||
self._logSys.error(
|
|
||||||
"Error banning IP %s for jail '%s' with action '%s': %s",
|
|
||||||
ip, self._jail.name, self.banaction, e,
|
|
||||||
exc_info=self._logSys.getEffectiveLevel()<=logging.DEBUG)
|
|
||||||
else:
|
|
||||||
self._bannedips.add(ip)
|
|
||||||
self._logSys.log(self.loglevel,
|
|
||||||
"Banned IP %s for jail '%s' with action '%s'",
|
|
||||||
ip, self._jail.name, self.banaction)
|
|
||||||
|
|
||||||
def _unbanIPs(self, ips):
|
|
||||||
for ip in ips:
|
|
||||||
try:
|
|
||||||
ai = Actions.ActionInfo(BanTicket(ip), self._jail)
|
|
||||||
self._jail.actions[self.banaction].unban(ai)
|
|
||||||
except Exception as e:
|
|
||||||
self._logSys.error(
|
|
||||||
"Error unbanning IP %s for jail '%s' with action '%s': %s",
|
|
||||||
ip, self._jail.name, self.banaction, e,
|
|
||||||
exc_info=self._logSys.getEffectiveLevel()<=logging.DEBUG)
|
|
||||||
else:
|
|
||||||
self._logSys.log(self.loglevel,
|
|
||||||
"Unbanned IP %s for jail '%s' with action '%s'",
|
|
||||||
ip, self._jail.name, self.banaction)
|
|
||||||
finally:
|
|
||||||
self._bannedips.remove(ip)
|
|
||||||
|
|
||||||
def start(self):
|
|
||||||
"""If `banaction` set, blacklists bad IPs.
|
|
||||||
"""
|
|
||||||
if self.banaction is not None:
|
|
||||||
self.update()
|
|
||||||
|
|
||||||
def update(self):
|
|
||||||
"""If `banaction` set, updates blacklisted IPs.
|
|
||||||
|
|
||||||
Queries badips.com for list of bad IPs, removing IPs from the
|
|
||||||
blacklist if no longer present, and adds new bad IPs to the
|
|
||||||
blacklist.
|
|
||||||
"""
|
|
||||||
if self.banaction is not None:
|
|
||||||
if self._timer:
|
|
||||||
self._timer.cancel()
|
|
||||||
self._timer = None
|
|
||||||
|
|
||||||
try:
|
|
||||||
ips = self.getList(
|
|
||||||
self.bancategory, self.score, self.age, self.bankey)
|
|
||||||
# Remove old IPs no longer listed
|
|
||||||
s = self._bannedips - ips
|
|
||||||
m = len(s)
|
|
||||||
self._unbanIPs(s)
|
|
||||||
# Add new IPs which are now listed
|
|
||||||
s = ips - self._bannedips
|
|
||||||
p = len(s)
|
|
||||||
self._banIPs(s)
|
|
||||||
if m != 0 or p != 0:
|
|
||||||
self._logSys.log(self.sumloglevel,
|
|
||||||
"Updated IPs for jail '%s' (-%d/+%d)",
|
|
||||||
self._jail.name, m, p)
|
|
||||||
self._logSys.debug(
|
|
||||||
"Next update for jail '%' in %i seconds",
|
|
||||||
self._jail.name, self.updateperiod)
|
|
||||||
finally:
|
|
||||||
self._timer = threading.Timer(self.updateperiod, self.update)
|
|
||||||
self._timer.start()
|
|
||||||
|
|
||||||
def stop(self):
|
|
||||||
"""If `banaction` set, clears blacklisted IPs.
|
|
||||||
"""
|
|
||||||
if self.banaction is not None:
|
|
||||||
if self._timer:
|
|
||||||
self._timer.cancel()
|
|
||||||
self._timer = None
|
|
||||||
self._unbanIPs(self._bannedips.copy())
|
|
||||||
|
|
||||||
def ban(self, aInfo):
|
|
||||||
"""Reports banned IP to badips.com.
|
|
||||||
|
|
||||||
Parameters
|
|
||||||
----------
|
|
||||||
aInfo : dict
|
|
||||||
Dictionary which includes information in relation to
|
|
||||||
the ban.
|
|
||||||
|
|
||||||
Raises
|
|
||||||
------
|
|
||||||
HTTPError
|
|
||||||
Any issues with badips.com request.
|
|
||||||
"""
|
|
||||||
try:
|
|
||||||
url = "/".join([self._badips, "add", self.category, str(aInfo['ip'])])
|
|
||||||
self._logSys.debug('badips.com: ban, url: %r', url)
|
|
||||||
response = urlopen(self._Request(url), timeout=self.timeout)
|
|
||||||
except HTTPError as response: # pragma: no cover
|
|
||||||
self.logError(response, "Failed to ban")
|
|
||||||
raise
|
|
||||||
else:
|
|
||||||
messages = json.loads(response.read().decode('utf-8'))
|
|
||||||
self._logSys.debug(
|
|
||||||
"Response from badips.com report: '%s'",
|
|
||||||
messages['suc'])
|
|
||||||
|
|
||||||
Action = BadIPsAction
|
|
||||||
@ -1,157 +0,0 @@
|
|||||||
# emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: t -*-
|
|
||||||
# vi: set ft=python sts=4 ts=4 sw=4 noet :
|
|
||||||
|
|
||||||
# This file is part of Fail2Ban.
|
|
||||||
#
|
|
||||||
# Fail2Ban is free software; you can redistribute it and/or modify
|
|
||||||
# it under the terms of the GNU General Public License as published by
|
|
||||||
# the Free Software Foundation; either version 2 of the License, or
|
|
||||||
# (at your option) any later version.
|
|
||||||
#
|
|
||||||
# Fail2Ban is distributed in the hope that it will be useful,
|
|
||||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
||||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
||||||
# GNU General Public License for more details.
|
|
||||||
#
|
|
||||||
# You should have received a copy of the GNU General Public License
|
|
||||||
# along with Fail2Ban; if not, write to the Free Software
|
|
||||||
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
|
|
||||||
|
|
||||||
import os
|
|
||||||
import unittest
|
|
||||||
import sys
|
|
||||||
from functools import wraps
|
|
||||||
from socket import timeout
|
|
||||||
from ssl import SSLError
|
|
||||||
|
|
||||||
from ..actiontestcase import CallingMap
|
|
||||||
from ..dummyjail import DummyJail
|
|
||||||
from ..servertestcase import IPAddr
|
|
||||||
from ..utils import LogCaptureTestCase, CONFIG_DIR
|
|
||||||
|
|
||||||
if sys.version_info >= (3, ): # pragma: 2.x no cover
|
|
||||||
from urllib.error import HTTPError, URLError
|
|
||||||
else: # pragma: 3.x no cover
|
|
||||||
from urllib2 import HTTPError, URLError
|
|
||||||
|
|
||||||
def skip_if_not_available(f):
|
|
||||||
"""Helper to decorate tests to skip in case of timeout/http-errors like "502 bad gateway".
|
|
||||||
"""
|
|
||||||
@wraps(f)
|
|
||||||
def wrapper(self, *args):
|
|
||||||
try:
|
|
||||||
return f(self, *args)
|
|
||||||
except (SSLError, HTTPError, URLError, timeout) as e: # pragma: no cover - timeout/availability issues
|
|
||||||
if not isinstance(e, timeout) and 'timed out' not in str(e):
|
|
||||||
if not hasattr(e, 'code') or e.code > 200 and e.code <= 404:
|
|
||||||
raise
|
|
||||||
raise unittest.SkipTest('Skip test because of %s' % e)
|
|
||||||
return wrapper
|
|
||||||
|
|
||||||
if sys.version_info >= (2,7): # pragma: no cover - may be unavailable
|
|
||||||
class BadIPsActionTest(LogCaptureTestCase):
|
|
||||||
|
|
||||||
available = True, None
|
|
||||||
pythonModule = None
|
|
||||||
modAction = None
|
|
||||||
|
|
||||||
@skip_if_not_available
|
|
||||||
def setUp(self):
|
|
||||||
"""Call before every test case."""
|
|
||||||
super(BadIPsActionTest, self).setUp()
|
|
||||||
unittest.F2B.SkipIfNoNetwork()
|
|
||||||
|
|
||||||
self.jail = DummyJail()
|
|
||||||
|
|
||||||
self.jail.actions.add("test")
|
|
||||||
|
|
||||||
pythonModuleName = os.path.join(CONFIG_DIR, "action.d", "badips.py")
|
|
||||||
|
|
||||||
# check availability (once if not alive, used shorter timeout as in test cases):
|
|
||||||
if BadIPsActionTest.available[0]:
|
|
||||||
if not BadIPsActionTest.modAction:
|
|
||||||
if not BadIPsActionTest.pythonModule:
|
|
||||||
BadIPsActionTest.pythonModule = self.jail.actions._load_python_module(pythonModuleName)
|
|
||||||
BadIPsActionTest.modAction = BadIPsActionTest.pythonModule.Action
|
|
||||||
self.jail.actions._load_python_module(pythonModuleName)
|
|
||||||
BadIPsActionTest.available = BadIPsActionTest.modAction.isAvailable(timeout=2 if unittest.F2B.fast else 30)
|
|
||||||
if not BadIPsActionTest.available[0]:
|
|
||||||
raise unittest.SkipTest('Skip test because service is not available: %s' % BadIPsActionTest.available[1])
|
|
||||||
|
|
||||||
self.jail.actions.add("badips", pythonModuleName, initOpts={
|
|
||||||
'category': "ssh",
|
|
||||||
'banaction': "test",
|
|
||||||
'age': "2w",
|
|
||||||
'score': 5,
|
|
||||||
#'key': "fail2ban-test-suite",
|
|
||||||
#'bankey': "fail2ban-test-suite",
|
|
||||||
'timeout': (3 if unittest.F2B.fast else 60),
|
|
||||||
})
|
|
||||||
self.action = self.jail.actions["badips"]
|
|
||||||
|
|
||||||
def tearDown(self):
|
|
||||||
"""Call after every test case."""
|
|
||||||
# Must cancel timer!
|
|
||||||
if self.action._timer:
|
|
||||||
self.action._timer.cancel()
|
|
||||||
super(BadIPsActionTest, self).tearDown()
|
|
||||||
|
|
||||||
@skip_if_not_available
|
|
||||||
def testCategory(self):
|
|
||||||
categories = self.action.getCategories()
|
|
||||||
self.assertIn("ssh", categories)
|
|
||||||
self.assertTrue(len(categories) >= 10)
|
|
||||||
|
|
||||||
self.assertRaises(
|
|
||||||
ValueError, setattr, self.action, "category",
|
|
||||||
"invalid-category")
|
|
||||||
|
|
||||||
# Not valid for reporting category...
|
|
||||||
self.assertRaises(
|
|
||||||
ValueError, setattr, self.action, "category", "mail")
|
|
||||||
# but valid for blacklisting.
|
|
||||||
self.action.bancategory = "mail"
|
|
||||||
|
|
||||||
@skip_if_not_available
|
|
||||||
def testScore(self):
|
|
||||||
self.assertRaises(ValueError, setattr, self.action, "score", -5)
|
|
||||||
self.action.score = 3
|
|
||||||
self.action.score = "3"
|
|
||||||
|
|
||||||
@skip_if_not_available
|
|
||||||
def testBanaction(self):
|
|
||||||
self.assertRaises(
|
|
||||||
ValueError, setattr, self.action, "banaction",
|
|
||||||
"invalid-action")
|
|
||||||
self.action.banaction = "test"
|
|
||||||
|
|
||||||
@skip_if_not_available
|
|
||||||
def testUpdateperiod(self):
|
|
||||||
self.assertRaises(
|
|
||||||
ValueError, setattr, self.action, "updateperiod", -50)
|
|
||||||
self.assertRaises(
|
|
||||||
ValueError, setattr, self.action, "updateperiod", 0)
|
|
||||||
self.action.updateperiod = 900
|
|
||||||
self.action.updateperiod = "900"
|
|
||||||
|
|
||||||
@skip_if_not_available
|
|
||||||
def testStartStop(self):
|
|
||||||
self.action.start()
|
|
||||||
self.assertTrue(len(self.action._bannedips) > 10,
|
|
||||||
"%s is fewer as 10: %r" % (len(self.action._bannedips), self.action._bannedips))
|
|
||||||
self.action.stop()
|
|
||||||
self.assertTrue(len(self.action._bannedips) == 0)
|
|
||||||
|
|
||||||
@skip_if_not_available
|
|
||||||
def testBanIP(self):
|
|
||||||
aInfo = CallingMap({
|
|
||||||
'ip': IPAddr('192.0.2.1')
|
|
||||||
})
|
|
||||||
self.action.ban(aInfo)
|
|
||||||
self.assertLogged('badips.com: ban', wait=True)
|
|
||||||
self.pruneLog()
|
|
||||||
# produce an error using wrong category/IP:
|
|
||||||
self.action._category = 'f2b-this-category-dont-available-test-suite-only'
|
|
||||||
aInfo['ip'] = ''
|
|
||||||
self.assertRaises(BadIPsActionTest.pythonModule.HTTPError, self.action.ban, aInfo)
|
|
||||||
self.assertLogged('IP is invalid', 'invalid category', wait=True, all=False)
|
|
||||||
Loading…
Reference in new issue