Add fail2ban sensor (#9975)
* Initial revision of fail2ban sensor * Verified working, added tests * Re-factored code so that log reading isn't called for each sensor * Lint fixes * Removed errant reset of last ban, added test to verify bans persist through update * Removed for loop in read_log and replaced with regex per review request * Refactored update to use current ban array for last ban state - also was missing return False in timer for default behavior * Removed CONF_SCAN_INTERVAL from PLATFORM_SCHEMA.extend - renamed DEFAULT_SCAN_INTERVAL to SCAN_INTERVAL * SCAN_INTERVAL changed to timedelta * Force travis rebuild (last build timed out) * Using compiled regex nowpull/10064/head
Kevin Fronczak
Pascal Vizeli
parent
4fa0119245
commit
f9d89a016e
|
|
@ -0,0 +1,145 @@
|
|||
"""
|
||||
Support for displaying IPs banned by fail2ban.
|
||||
|
||||
For more details about this platform, please refer to the documentation at
|
||||
https://home-assistant.io/components/sensor.fail2ban/
|
||||
"""
|
||||
import os
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
from datetime import timedelta
|
||||
|
||||
import re
|
||||
import voluptuous as vol
|
||||
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
import homeassistant.util.dt as dt_util
|
||||
from homeassistant.components.sensor import PLATFORM_SCHEMA
|
||||
from homeassistant.const import (
|
||||
CONF_NAME, CONF_SCAN_INTERVAL, CONF_FILE_PATH
|
||||
)
|
||||
from homeassistant.helpers.entity import Entity
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
CONF_JAILS = 'jails'
|
||||
|
||||
DEFAULT_NAME = 'fail2ban'
|
||||
DEFAULT_LOG = '/var/log/fail2ban.log'
|
||||
SCAN_INTERVAL = timedelta(seconds=120)
|
||||
|
||||
STATE_CURRENT_BANS = 'current_bans'
|
||||
STATE_ALL_BANS = 'total_bans'
|
||||
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
|
||||
vol.Required(CONF_JAILS, default=[]):
|
||||
vol.All(cv.ensure_list, vol.Length(min=1)),
|
||||
vol.Optional(CONF_FILE_PATH, default=DEFAULT_LOG): cv.isfile,
|
||||
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
|
||||
})
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def async_setup_platform(hass, config, async_add_devices, discovery_info=None):
|
||||
"""Set up the fail2ban sensor."""
|
||||
name = config.get(CONF_NAME)
|
||||
jails = config.get(CONF_JAILS)
|
||||
scan_interval = config.get(CONF_SCAN_INTERVAL)
|
||||
log_file = config.get(CONF_FILE_PATH)
|
||||
|
||||
device_list = []
|
||||
log_parser = BanLogParser(scan_interval, log_file)
|
||||
for jail in jails:
|
||||
device_list.append(BanSensor(name, jail, log_parser))
|
||||
|
||||
async_add_devices(device_list, True)
|
||||
|
||||
|
||||
class BanSensor(Entity):
|
||||
"""Implementation of a fail2ban sensor."""
|
||||
|
||||
def __init__(self, name, jail, log_parser):
|
||||
"""Initialize the sensor."""
|
||||
self._name = '{} {}'.format(name, jail)
|
||||
self.jail = jail
|
||||
self.ban_dict = {STATE_CURRENT_BANS: [], STATE_ALL_BANS: []}
|
||||
self.last_ban = None
|
||||
self.log_parser = log_parser
|
||||
self.log_parser.ip_regex[self.jail] = re.compile(
|
||||
r"\[{}\].(Ban|Unban) ([\w+\.]{{3,}})".format(re.escape(self.jail))
|
||||
)
|
||||
_LOGGER.debug("Setting up jail %s", self.jail)
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the sensor."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def state_attributes(self):
|
||||
"""Return the state attributes of the fail2ban sensor."""
|
||||
return self.ban_dict
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
"""Return the most recently banned IP Address."""
|
||||
return self.last_ban
|
||||
|
||||
def update(self):
|
||||
"""Update the list of banned ips."""
|
||||
if self.log_parser.timer():
|
||||
self.log_parser.read_log(self.jail)
|
||||
|
||||
if self.log_parser.data:
|
||||
for entry in self.log_parser.data:
|
||||
_LOGGER.debug(entry)
|
||||
current_ip = entry[1]
|
||||
if entry[0] == 'Ban':
|
||||
if current_ip not in self.ban_dict[STATE_CURRENT_BANS]:
|
||||
self.ban_dict[STATE_CURRENT_BANS].append(current_ip)
|
||||
if current_ip not in self.ban_dict[STATE_ALL_BANS]:
|
||||
self.ban_dict[STATE_ALL_BANS].append(current_ip)
|
||||
if len(self.ban_dict[STATE_ALL_BANS]) > 10:
|
||||
self.ban_dict[STATE_ALL_BANS].pop(0)
|
||||
|
||||
elif entry[0] == 'Unban':
|
||||
if current_ip in self.ban_dict[STATE_CURRENT_BANS]:
|
||||
self.ban_dict[STATE_CURRENT_BANS].remove(current_ip)
|
||||
|
||||
if self.ban_dict[STATE_CURRENT_BANS]:
|
||||
self.last_ban = self.ban_dict[STATE_CURRENT_BANS][-1]
|
||||
else:
|
||||
self.last_ban = 'None'
|
||||
|
||||
|
||||
class BanLogParser(object):
|
||||
"""Class to parse fail2ban logs."""
|
||||
|
||||
def __init__(self, interval, log_file):
|
||||
"""Initialize the parser."""
|
||||
self.interval = interval
|
||||
self.log_file = log_file
|
||||
self.data = list()
|
||||
self.last_update = dt_util.now()
|
||||
self.ip_regex = dict()
|
||||
|
||||
def timer(self):
|
||||
"""Check if we are allowed to update."""
|
||||
boundary = dt_util.now() - self.interval
|
||||
if boundary > self.last_update:
|
||||
self.last_update = dt_util.now()
|
||||
return True
|
||||
return False
|
||||
|
||||
def read_log(self, jail):
|
||||
"""Read the fail2ban log and find entries for jail."""
|
||||
self.data = list()
|
||||
try:
|
||||
with open(self.log_file, 'r', encoding='utf-8') as file_data:
|
||||
self.data = self.ip_regex[jail].findall(file_data.read())
|
||||
|
||||
except (IndexError, FileNotFoundError, IsADirectoryError,
|
||||
UnboundLocalError):
|
||||
_LOGGER.warning("File not present: %s",
|
||||
os.path.basename(self.log_file))
|
||||
|
|
@ -0,0 +1,220 @@
|
|||
"""The tests for local file sensor platform."""
|
||||
import unittest
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
from datetime import timedelta
|
||||
from mock_open import MockOpen
|
||||
|
||||
from homeassistant.setup import setup_component
|
||||
from homeassistant.components.sensor.fail2ban import (
|
||||
BanSensor, BanLogParser, STATE_CURRENT_BANS, STATE_ALL_BANS
|
||||
)
|
||||
|
||||
from tests.common import get_test_home_assistant, assert_setup_component
|
||||
|
||||
|
||||
def fake_log(log_key):
|
||||
"""Return a fake fail2ban log."""
|
||||
fake_log_dict = {
|
||||
'single_ban': (
|
||||
'2017-01-01 12:23:35 fail2ban.actions [111]: '
|
||||
'NOTICE [jail_one] Ban 111.111.111.111'
|
||||
),
|
||||
'multi_ban': (
|
||||
'2017-01-01 12:23:35 fail2ban.actions [111]: '
|
||||
'NOTICE [jail_one] Ban 111.111.111.111\n'
|
||||
'2017-01-01 12:23:35 fail2ban.actions [111]: '
|
||||
'NOTICE [jail_one] Ban 222.222.222.222'
|
||||
),
|
||||
'multi_jail': (
|
||||
'2017-01-01 12:23:35 fail2ban.actions [111]: '
|
||||
'NOTICE [jail_one] Ban 111.111.111.111\n'
|
||||
'2017-01-01 12:23:35 fail2ban.actions [111]: '
|
||||
'NOTICE [jail_two] Ban 222.222.222.222'
|
||||
),
|
||||
'unban_all': (
|
||||
'2017-01-01 12:23:35 fail2ban.actions [111]: '
|
||||
'NOTICE [jail_one] Ban 111.111.111.111\n'
|
||||
'2017-01-01 12:23:35 fail2ban.actions [111]: '
|
||||
'NOTICE [jail_one] Unban 111.111.111.111\n'
|
||||
'2017-01-01 12:23:35 fail2ban.actions [111]: '
|
||||
'NOTICE [jail_one] Ban 222.222.222.222\n'
|
||||
'2017-01-01 12:23:35 fail2ban.actions [111]: '
|
||||
'NOTICE [jail_one] Unban 222.222.222.222'
|
||||
),
|
||||
'unban_one': (
|
||||
'2017-01-01 12:23:35 fail2ban.actions [111]: '
|
||||
'NOTICE [jail_one] Ban 111.111.111.111\n'
|
||||
'2017-01-01 12:23:35 fail2ban.actions [111]: '
|
||||
'NOTICE [jail_one] Ban 222.222.222.222\n'
|
||||
'2017-01-01 12:23:35 fail2ban.actions [111]: '
|
||||
'NOTICE [jail_one] Unban 111.111.111.111'
|
||||
)
|
||||
}
|
||||
return fake_log_dict[log_key]
|
||||
|
||||
|
||||
class TestBanSensor(unittest.TestCase):
|
||||
"""Test the fail2ban sensor."""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up things to be run when tests are started."""
|
||||
self.hass = get_test_home_assistant()
|
||||
|
||||
def tearDown(self):
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
|
||||
@patch('os.path.isfile', Mock(return_value=True))
|
||||
def test_setup(self):
|
||||
"""Test that sensor can be setup."""
|
||||
config = {
|
||||
'sensor': {
|
||||
'platform': 'fail2ban',
|
||||
'jails': ['jail_one']
|
||||
}
|
||||
}
|
||||
mock_fh = MockOpen()
|
||||
with patch('homeassistant.components.sensor.fail2ban.open', mock_fh,
|
||||
create=True):
|
||||
assert setup_component(self.hass, 'sensor', config)
|
||||
self.hass.block_till_done()
|
||||
assert_setup_component(1, 'sensor')
|
||||
|
||||
@patch('os.path.isfile', Mock(return_value=True))
|
||||
def test_multi_jails(self):
|
||||
"""Test that multiple jails can be set up as sensors.."""
|
||||
config = {
|
||||
'sensor': {
|
||||
'platform': 'fail2ban',
|
||||
'jails': ['jail_one', 'jail_two']
|
||||
}
|
||||
}
|
||||
mock_fh = MockOpen()
|
||||
with patch('homeassistant.components.sensor.fail2ban.open', mock_fh,
|
||||
create=True):
|
||||
assert setup_component(self.hass, 'sensor', config)
|
||||
self.hass.block_till_done()
|
||||
assert_setup_component(2, 'sensor')
|
||||
|
||||
def test_single_ban(self):
|
||||
"""Test that log is parsed correctly for single ban."""
|
||||
log_parser = BanLogParser(timedelta(seconds=-1), '/tmp')
|
||||
sensor = BanSensor('fail2ban', 'jail_one', log_parser)
|
||||
self.assertEqual(sensor.name, 'fail2ban jail_one')
|
||||
mock_fh = MockOpen(read_data=fake_log('single_ban'))
|
||||
with patch('homeassistant.components.sensor.fail2ban.open', mock_fh,
|
||||
create=True):
|
||||
sensor.update()
|
||||
|
||||
self.assertEqual(sensor.state, '111.111.111.111')
|
||||
self.assertEqual(
|
||||
sensor.state_attributes[STATE_CURRENT_BANS], ['111.111.111.111']
|
||||
)
|
||||
self.assertEqual(
|
||||
sensor.state_attributes[STATE_ALL_BANS], ['111.111.111.111']
|
||||
)
|
||||
|
||||
def test_multiple_ban(self):
|
||||
"""Test that log is parsed correctly for multiple ban."""
|
||||
log_parser = BanLogParser(timedelta(seconds=-1), '/tmp')
|
||||
sensor = BanSensor('fail2ban', 'jail_one', log_parser)
|
||||
self.assertEqual(sensor.name, 'fail2ban jail_one')
|
||||
mock_fh = MockOpen(read_data=fake_log('multi_ban'))
|
||||
with patch('homeassistant.components.sensor.fail2ban.open', mock_fh,
|
||||
create=True):
|
||||
sensor.update()
|
||||
|
||||
self.assertEqual(sensor.state, '222.222.222.222')
|
||||
self.assertEqual(
|
||||
sensor.state_attributes[STATE_CURRENT_BANS],
|
||||
['111.111.111.111', '222.222.222.222']
|
||||
)
|
||||
self.assertEqual(
|
||||
sensor.state_attributes[STATE_ALL_BANS],
|
||||
['111.111.111.111', '222.222.222.222']
|
||||
)
|
||||
|
||||
def test_unban_all(self):
|
||||
"""Test that log is parsed correctly when unbanning."""
|
||||
log_parser = BanLogParser(timedelta(seconds=-1), '/tmp')
|
||||
sensor = BanSensor('fail2ban', 'jail_one', log_parser)
|
||||
self.assertEqual(sensor.name, 'fail2ban jail_one')
|
||||
mock_fh = MockOpen(read_data=fake_log('unban_all'))
|
||||
with patch('homeassistant.components.sensor.fail2ban.open', mock_fh,
|
||||
create=True):
|
||||
sensor.update()
|
||||
|
||||
self.assertEqual(sensor.state, 'None')
|
||||
self.assertEqual(sensor.state_attributes[STATE_CURRENT_BANS], [])
|
||||
self.assertEqual(
|
||||
sensor.state_attributes[STATE_ALL_BANS],
|
||||
['111.111.111.111', '222.222.222.222']
|
||||
)
|
||||
|
||||
def test_unban_one(self):
|
||||
"""Test that log is parsed correctly when unbanning one ip."""
|
||||
log_parser = BanLogParser(timedelta(seconds=-1), '/tmp')
|
||||
sensor = BanSensor('fail2ban', 'jail_one', log_parser)
|
||||
self.assertEqual(sensor.name, 'fail2ban jail_one')
|
||||
mock_fh = MockOpen(read_data=fake_log('unban_one'))
|
||||
with patch('homeassistant.components.sensor.fail2ban.open', mock_fh,
|
||||
create=True):
|
||||
sensor.update()
|
||||
|
||||
self.assertEqual(sensor.state, '222.222.222.222')
|
||||
self.assertEqual(
|
||||
sensor.state_attributes[STATE_CURRENT_BANS],
|
||||
['222.222.222.222']
|
||||
)
|
||||
self.assertEqual(
|
||||
sensor.state_attributes[STATE_ALL_BANS],
|
||||
['111.111.111.111', '222.222.222.222']
|
||||
)
|
||||
|
||||
def test_multi_jail(self):
|
||||
"""Test that log is parsed correctly when using multiple jails."""
|
||||
log_parser = BanLogParser(timedelta(seconds=-1), '/tmp')
|
||||
sensor1 = BanSensor('fail2ban', 'jail_one', log_parser)
|
||||
sensor2 = BanSensor('fail2ban', 'jail_two', log_parser)
|
||||
self.assertEqual(sensor1.name, 'fail2ban jail_one')
|
||||
self.assertEqual(sensor2.name, 'fail2ban jail_two')
|
||||
mock_fh = MockOpen(read_data=fake_log('multi_jail'))
|
||||
with patch('homeassistant.components.sensor.fail2ban.open', mock_fh,
|
||||
create=True):
|
||||
sensor1.update()
|
||||
sensor2.update()
|
||||
|
||||
self.assertEqual(sensor1.state, '111.111.111.111')
|
||||
self.assertEqual(
|
||||
sensor1.state_attributes[STATE_CURRENT_BANS], ['111.111.111.111']
|
||||
)
|
||||
self.assertEqual(
|
||||
sensor1.state_attributes[STATE_ALL_BANS], ['111.111.111.111']
|
||||
)
|
||||
self.assertEqual(sensor2.state, '222.222.222.222')
|
||||
self.assertEqual(
|
||||
sensor2.state_attributes[STATE_CURRENT_BANS], ['222.222.222.222']
|
||||
)
|
||||
self.assertEqual(
|
||||
sensor2.state_attributes[STATE_ALL_BANS], ['222.222.222.222']
|
||||
)
|
||||
|
||||
def test_ban_active_after_update(self):
|
||||
"""Test that ban persists after subsequent update."""
|
||||
log_parser = BanLogParser(timedelta(seconds=-1), '/tmp')
|
||||
sensor = BanSensor('fail2ban', 'jail_one', log_parser)
|
||||
self.assertEqual(sensor.name, 'fail2ban jail_one')
|
||||
mock_fh = MockOpen(read_data=fake_log('single_ban'))
|
||||
with patch('homeassistant.components.sensor.fail2ban.open', mock_fh,
|
||||
create=True):
|
||||
sensor.update()
|
||||
self.assertEqual(sensor.state, '111.111.111.111')
|
||||
sensor.update()
|
||||
self.assertEqual(sensor.state, '111.111.111.111')
|
||||
self.assertEqual(
|
||||
sensor.state_attributes[STATE_CURRENT_BANS], ['111.111.111.111']
|
||||
)
|
||||
self.assertEqual(
|
||||
sensor.state_attributes[STATE_ALL_BANS], ['111.111.111.111']
|
||||
)
|
||||
Loading…
Reference in New Issue