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

improve adapter/converter handlers working on invalid characters in sense of json and/or sqlite-database; 

both should be additionally exception-safe, so avoid possible errors in log-handlers (concat, str. conversion, etc);
test cases extended to cover any possible variants (invalid chars in unicode, bytes, str + unterminated char-sequence) with both cases (with replace of chars, with and without errors inside adapter-handlers).
pull/2171/head
sebres 2018-07-04 16:54:05 +02:00
parent 85fd1854bc
commit 930cc6c8f1
2 changed files with 81 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
fail2ban/server/database.py
View File

@ -33,32 +33,39 @@ from threading import RLock
from .mytime import MyTime
from .ticket import FailTicket
from .utils import Utils
from ..helpers import getLogger, PREFER_ENC
from ..helpers import getLogger, uni_string, PREFER_ENC
# Gets the instance of the logger.
logSys = getLogger(__name__)
if sys.version_info >= (3,):
def _json_default(x):
if isinstance(x, set):
x = list(x)
return x
def _json_default(x):
"""Avoid errors on types unknow in json-adapters."""
if isinstance(x, set):
x = list(x)
return uni_string(x)
if sys.version_info >= (3,):
def _json_dumps_safe(x):
try:
x = json.dumps(x, ensure_ascii=False, default=_json_default).encode(
PREFER_ENC, 'replace')
except Exception as e: # pragma: no cover
logSys.error('json dumps failed: %s', e)
# adapter handler should be exception-safe, so avoid possible errors in log-handlers (concat, str. conversion, etc)
try:
logSys.error('json dumps failed: %r', e, exc_info=logSys.getEffectiveLevel() <= 4)
except: pass
x = '{}'
return x
def _json_loads_safe(x):
try:
x = json.loads(x.decode(
PREFER_ENC, 'replace'))
x = json.loads(x.decode(PREFER_ENC, 'replace'))
except Exception as e: # pragma: no cover
logSys.error('json loads failed: %s', e)
# converter handler should be exception-safe, so avoid possible errors in log-handlers (concat, str. conversion, etc)
try:
logSys.error('json loads failed: %r', e, exc_info=logSys.getEffectiveLevel() <= 4)
except: pass
x = {}
return x
else:
@ -68,25 +75,31 @@ else:
elif isinstance(x, (list, set)):
return [_normalize(element) for element in x]
elif isinstance(x, unicode):
return x.encode(PREFER_ENC)
else:
return x
# in 2.x default text_factory is unicode - so return proper unicode here:
return x.encode(PREFER_ENC, 'replace').decode(PREFER_ENC)
elif isinstance(x, basestring):
return x.decode(PREFER_ENC, 'replace')
return x
def _json_dumps_safe(x):
try:
x = json.dumps(_normalize(x), ensure_ascii=False).decode(
PREFER_ENC, 'replace')
x = json.dumps(_normalize(x), ensure_ascii=False, default=_json_default)
except Exception as e: # pragma: no cover
logSys.error('json dumps failed: %s', e)
# adapter handler should be exception-safe, so avoid possible errors in log-handlers (concat, str. conversion, etc)
try:
logSys.error('json dumps failed: %r', e, exc_info=logSys.getEffectiveLevel() <= 4)
except: pass
x = '{}'
return x
def _json_loads_safe(x):
try:
x = json.loads(x.decode(
PREFER_ENC, 'replace'))
x = json.loads(x.decode(PREFER_ENC, 'replace'))
except Exception as e: # pragma: no cover
logSys.error('json loads failed: %s', e)
# converter handler should be exception-safe, so avoid possible errors in log-handlers (concat, str. conversion, etc)
try:
logSys.error('json loads failed: %r', e, exc_info=logSys.getEffectiveLevel() <= 4)
except: pass
x = {}
return x
@ -184,6 +197,8 @@ class Fail2BanDb(object):
self._db = sqlite3.connect(
filename, check_same_thread=False,
detect_types=sqlite3.PARSE_DECLTYPES)
# # to allow use multi-byte utf-8
# self._db.text_factory = str
self._bansMergedCache = {}

70
fail2ban/tests/databasetestcase.py
View File

@ -35,10 +35,11 @@ from ..server.ticket import FailTicket
from ..server.actions import Actions, Utils
from .dummyjail import DummyJail
try:
from ..server.database import Fail2BanDb as Fail2BanDb
from ..server import database
Fail2BanDb = database.Fail2BanDb
except ImportError: # pragma: no cover
Fail2BanDb = None
from .utils import LogCaptureTestCase
from .utils import LogCaptureTestCase, logSys as DefLogSys
TEST_FILES_DIR = os.path.join(os.path.dirname(__file__), "files")
@ -238,30 +239,53 @@ class DatabaseTest(LogCaptureTestCase):
self.testAddJail()
# invalid + valid, invalid + valid unicode, invalid + valid dual converted (like in filter:readline by fallback) ...
tickets = [
FailTicket("127.0.0.1", 0, ['user "\xd1\xe2\xe5\xf2\xe0"', 'user "\xc3\xa4\xc3\xb6\xc3\xbc\xc3\x9f"']),
FailTicket("127.0.0.2", 0, ['user "\xd1\xe2\xe5\xf2\xe0"', u'user "\xc3\xa4\xc3\xb6\xc3\xbc\xc3\x9f"']),
FailTicket("127.0.0.3", 0, ['user "\xd1\xe2\xe5\xf2\xe0"', b'user "\xc3\xa4\xc3\xb6\xc3\xbc\xc3\x9f"'.decode('utf-8', 'replace')])
FailTicket("127.0.0.1", 0, ['user "test"', 'user "\xd1\xe2\xe5\xf2\xe0"', 'user "\xc3\xa4\xc3\xb6\xc3\xbc\xc3\x9f"']),
FailTicket("127.0.0.2", 0, ['user "test"', u'user "\xd1\xe2\xe5\xf2\xe0"', u'user "\xc3\xa4\xc3\xb6\xc3\xbc\xc3\x9f"']),
FailTicket("127.0.0.3", 0, ['user "test"', b'user "\xd1\xe2\xe5\xf2\xe0"', b'user "\xc3\xa4\xc3\xb6\xc3\xbc\xc3\x9f"']),
FailTicket("127.0.0.4", 0, ['user "test"', 'user "\xd1\xe2\xe5\xf2\xe0"', u'user "\xe4\xf6\xfc\xdf"']),
FailTicket("127.0.0.5", 0, ['user "test"', 'unterminated \xcf']),
FailTicket("127.0.0.6", 0, ['user "test"', u'unterminated \xcf']),
FailTicket("127.0.0.7", 0, ['user "test"', b'unterminated \xcf'])
]
self.db.addBan(self.jail, tickets[0])
self.db.addBan(self.jail, tickets[1])
self.db.addBan(self.jail, tickets[2])
for ticket in tickets:
self.db.addBan(self.jail, ticket)
self.assertNotLogged("json dumps failed")
readtickets = self.db.getBans(jail=self.jail)
self.assertEqual(len(readtickets), 3)
## python 2 or 3 :
invstr = u'user "\ufffd\ufffd\ufffd\ufffd\ufffd"'.encode('utf-8', 'replace')
self.assertTrue(
readtickets[0] == FailTicket("127.0.0.1", 0, [invstr, 'user "\xc3\xa4\xc3\xb6\xc3\xbc\xc3\x9f"'])
or readtickets[0] == tickets[0]
)
self.assertTrue(
readtickets[1] == FailTicket("127.0.0.2", 0, [invstr, u'user "\xc3\xa4\xc3\xb6\xc3\xbc\xc3\x9f"'.encode('utf-8', 'replace')])
or readtickets[1] == tickets[1]
)
self.assertTrue(
readtickets[2] == FailTicket("127.0.0.3", 0, [invstr, 'user "\xc3\xa4\xc3\xb6\xc3\xbc\xc3\x9f"'])
or readtickets[2] == tickets[2]
)
self.assertNotLogged("json loads failed")
## all tickets available
self.assertEqual(len(readtickets), 7)
## too different to cover all possible constellations for python 2 and 3,
## can replace/ignore some non-ascii chars by json dump/load (unicode/str),
## so check ip and matches count only:
for i, ticket in enumerate(tickets):
DefLogSys.debug('readtickets[%d]: %r', i, readtickets[i].getData())
DefLogSys.debug(' == tickets[%d]: %r', i, ticket.getData())
self.assertEqual(readtickets[i].getIP(), ticket.getIP())
self.assertEqual(len(readtickets[i].getMatches()), len(ticket.getMatches()))
## simulate errors in dumps/loads:
priorEnc = database.PREFER_ENC
try:
database.PREFER_ENC = 'f2b-test::non-existing-encoding'
for ticket in tickets:
self.db.addBan(self.jail, ticket)
self.assertLogged("json dumps failed")
readtickets = self.db.getBans(jail=self.jail)
self.assertLogged("json loads failed")
## despite errors all tickets written and loaded (check adapter-handlers are error-safe):
self.assertEqual(len(readtickets), 14)
finally:
database.PREFER_ENC = priorEnc
def _testAdd3Bans(self):
self.testAddJail()