From e8ee3ba544c7eefd7b41e34ec3a44550c1280fbb Mon Sep 17 00:00:00 2001
From: sebres <serg.brester@sebres.de>
Date: Tue, 8 Sep 2020 11:36:54 +0200
Subject: [PATCH] resolves a bottleneck within transmitting of large data
 between server and client: speedup search of communications end-marker and
 increase max buffer size (up to 32KB)

---
 fail2ban/client/csocket.py       | 14 +++++++++-----
 fail2ban/tests/sockettestcase.py |  4 ++--
 2 files changed, 11 insertions(+), 7 deletions(-)

diff --git a/fail2ban/client/csocket.py b/fail2ban/client/csocket.py
index ab3e294b..88795674 100644
--- a/fail2ban/client/csocket.py
+++ b/fail2ban/client/csocket.py
@@ -48,7 +48,8 @@ class CSocket:
 	def send(self, msg, nonblocking=False, timeout=None):
 		# Convert every list member to string
 		obj = dumps(map(CSocket.convert, msg), HIGHEST_PROTOCOL)
-		self.__csock.send(obj + CSPROTO.END)
+		self.__csock.send(obj)
+		self.__csock.send(CSPROTO.END)
 		return self.receive(self.__csock, nonblocking, timeout)
 
 	def settimeout(self, timeout):
@@ -81,9 +82,12 @@ class CSocket:
 		msg = CSPROTO.EMPTY
 		if nonblocking: sock.setblocking(0)
 		if timeout: sock.settimeout(timeout)
-		while msg.rfind(CSPROTO.END) == -1:
-			chunk = sock.recv(512)
-			if chunk in ('', b''): # python 3.x may return b'' instead of ''
-				raise RuntimeError("socket connection broken")
+		bufsize = 1024
+		while msg.rfind(CSPROTO.END, -32) == -1:
+			chunk = sock.recv(bufsize)
+			if not len(chunk):
+				raise socket.error(104, 'Connection reset by peer')
+			if chunk == CSPROTO.END: break
 			msg = msg + chunk
+			if bufsize < 32768: bufsize <<= 1
 		return loads(msg)
diff --git a/fail2ban/tests/sockettestcase.py b/fail2ban/tests/sockettestcase.py
index 8cd22a41..2d414e5c 100644
--- a/fail2ban/tests/sockettestcase.py
+++ b/fail2ban/tests/sockettestcase.py
@@ -152,7 +152,7 @@ class Socket(LogCaptureTestCase):
 		org_handler = RequestHandler.found_terminator
 		try:
 			RequestHandler.found_terminator = lambda self: self.close()
-			self.assertRaisesRegexp(RuntimeError, r"socket connection broken", 
+			self.assertRaisesRegexp(Exception, r"reset by peer|Broken pipe",
 				lambda: client.send(testMessage, timeout=unittest.F2B.maxWaitTime(10)))
 		finally:
 			RequestHandler.found_terminator = org_handler
@@ -168,7 +168,7 @@ class Socket(LogCaptureTestCase):
 		org_handler = RequestHandler.found_terminator
 		try:
 			RequestHandler.found_terminator = lambda self: TestMsgError()
-			#self.assertRaisesRegexp(RuntimeError, r"socket connection broken", client.send, testMessage)
+			#self.assertRaisesRegexp(Exception, r"reset by peer|Broken pipe", client.send, testMessage)
 			self.assertEqual(client.send(testMessage), 'ERROR: test unpickle error')
 		finally:
 			RequestHandler.found_terminator = org_handler