From bf6667d4da83a2c012d5fa245ca9002677aacec6 Mon Sep 17 00:00:00 2001 From: sebres Date: Wed, 6 Dec 2017 01:38:39 +0100 Subject: [PATCH] better (sane) stop server handling, AsyncServer.stop_communication back-ported to 0.10 (cherry-picked from 0.11); --- fail2ban/server/asyncserver.py | 58 +++++++++++++++++++++++----------- fail2ban/server/server.py | 30 ++++++++++-------- 2 files changed, 56 insertions(+), 32 deletions(-) diff --git a/fail2ban/server/asyncserver.py b/fail2ban/server/asyncserver.py index 9cc74658..11c81649 100644 --- a/fail2ban/server/asyncserver.py +++ b/fail2ban/server/asyncserver.py @@ -37,7 +37,7 @@ import traceback from .utils import Utils from ..protocol import CSPROTO -from ..helpers import getLogger,formatExceptionInfo +from ..helpers import logging, getLogger, formatExceptionInfo # Gets the instance of the logger. logSys = getLogger(__name__) @@ -80,22 +80,36 @@ class RequestHandler(asynchat.async_chat): # Deserialize message = loads(message) # Gives the message to the transmitter. - message = self.__transmitter.proceed(message) + if self.__transmitter: + message = self.__transmitter.proceed(message) + else: + message = ['SHUTDOWN'] # Serializes the response. message = dumps(message, HIGHEST_PROTOCOL) # Sends the response to the client. self.push(message + CSPROTO.END) - except Exception as e: # pragma: no cover + except Exception as e: logSys.error("Caught unhandled exception: %r", e, exc_info=logSys.getEffectiveLevel()<=logging.DEBUG) + # Sends the response to the client. + message = dumps("ERROR: %s" % e, HIGHEST_PROTOCOL) + self.push(message + CSPROTO.END) - + ## + # Handles an communication errors in request. + # def handle_error(self): - e1, e2 = formatExceptionInfo() - logSys.error("Unexpected communication error: %s" % str(e2)) - logSys.error(traceback.format_exc().splitlines()) - self.close() - + try: + e1, e2 = formatExceptionInfo() + logSys.error("Unexpected communication error: %s" % str(e2)) + logSys.error(traceback.format_exc().splitlines()) + # Sends the response to the client. + message = dumps("ERROR: %s" % e2, HIGHEST_PROTOCOL) + self.push(message + CSPROTO.END) + except Exception as e: # pragma: no cover - normally unreachable + pass + self.close_when_done() + def loop(active, timeout=None, use_poll=False): """Custom event loop implementation @@ -119,18 +133,20 @@ def loop(active, timeout=None, use_poll=False): poll(timeout) if errCount: errCount -= 1 - except Exception as e: # pragma: no cover + except Exception as e: if not active(): break errCount += 1 if errCount < 20: - if e.args[0] in (errno.ENOTCONN, errno.EBADF): # (errno.EBADF, 'Bad file descriptor') + # errno.ENOTCONN - 'Socket is not connected' + # errno.EBADF - 'Bad file descriptor' + if e.args[0] in (errno.ENOTCONN, errno.EBADF): # pragma: no cover (too sporadic) logSys.info('Server connection was closed: %s', str(e)) else: logSys.error('Server connection was closed: %s', str(e)) elif errCount == 20: - logSys.info('Too many errors - stop logging connection errors') logSys.exception(e) + logSys.error('Too many errors - stop logging connection errors') ## @@ -158,10 +174,10 @@ class AsyncServer(asyncore.dispatcher): def handle_accept(self): try: conn, addr = self.accept() - except socket.error: + except socket.error: # pragma: no cover logSys.warning("Socket error") return - except TypeError: + except TypeError: # pragma: no cover logSys.warning("Type error") return AsyncServer.__markCloseOnExec(conn) @@ -175,7 +191,7 @@ class AsyncServer(asyncore.dispatcher): # @param sock: socket file. # @param force: remove the socket file if exists. - def start(self, sock, force, use_poll=False): + def start(self, sock, force, timeout=None, use_poll=False): self.__worker = threading.current_thread() self.__sock = sock # Remove socket @@ -191,7 +207,7 @@ class AsyncServer(asyncore.dispatcher): self.set_reuse_addr() try: self.bind(sock) - except Exception: + except Exception: # pragma: no cover raise AsyncServerException("Unable to bind socket %s" % self.__sock) AsyncServer.__markCloseOnExec(self.socket) self.listen(1) @@ -201,12 +217,11 @@ class AsyncServer(asyncore.dispatcher): if self.onstart: self.onstart() # Event loop as long as active: - loop(lambda: self.__loop, use_poll=use_poll) + loop(lambda: self.__loop, timeout=timeout, use_poll=use_poll) self.__active = False # Cleanup all self.stop() - def close(self): stopflg = False if self.__active: @@ -228,6 +243,13 @@ class AsyncServer(asyncore.dispatcher): ## # Stops the communication server. + def stop_communication(self): + logSys.debug("Stop communication") + self.__transmitter = None + + ## + # Stops the server. + def stop(self): self.close() diff --git a/fail2ban/server/server.py b/fail2ban/server/server.py index de75bed6..b166b94a 100644 --- a/fail2ban/server/server.py +++ b/fail2ban/server/server.py @@ -82,7 +82,7 @@ class Server: def __sigTERMhandler(self, signum, frame): # pragma: no cover - indirect tested logSys.debug("Caught signal %d. Exiting", signum) - self.quit_signal() + self.quit() def __sigUSR1handler(self, signum, fname): # pragma: no cover - indirect tested logSys.debug("Caught signal %d. Flushing logs", signum) @@ -152,7 +152,7 @@ class Server: except AsyncServerException as e: logSys.error("Could not start server: %s", e) - # Stop server + # Stop (if not yet already executed): self.quit() # Removes the PID file. @@ -161,13 +161,22 @@ class Server: os.remove(pidfile) except (OSError, IOError) as e: # pragma: no cover logSys.error("Unable to remove PID file: %s", e) - logSys.info("Exiting Fail2ban") def quit(self): - self.quit_signal() + # Prevent to call quit twice: + self.quit = lambda: False logSys.info("Shutdown in progress...") + # Stop communication first because if jail's unban action + # tries to communicate via fail2ban-client we get a lockup + # among threads. So the simplest resolution is to stop all + # communications first (which should be ok anyways since we + # are exiting) + # See https://github.com/fail2ban/fail2ban/issues/7 + if self.__asyncServer is not None: + self.__asyncServer.stop_communication() + # Restore default signal handlers: if _thread_name() == '_MainThread': for s, sh in self.__prev_signals.iteritems(): @@ -182,18 +191,11 @@ class Server: self.__db.close() self.__db = None - def quit_signal(self): - # Prevent to call quit_signal twice: - self.quit_signal = lambda: False - # Stop communication first because if jail's unban action - # tries to communicate via fail2ban-client we get a lockup - # among threads. So the simplest resolution is to stop all - # communications first (which should be ok anyways since we - # are exiting) - # See https://github.com/fail2ban/fail2ban/issues/7 + # Stop async if self.__asyncServer is not None: self.__asyncServer.stop() self.__asyncServer = None + logSys.info("Exiting Fail2ban") def addJail(self, name, backend): addflg = True @@ -610,7 +612,7 @@ class Server: try: handler.flush() handler.close() - except (ValueError, KeyError): # pragma: no cover + except (ValueError, KeyError): # pragma: no cover # Is known to be thrown after logging was shutdown once # with older Pythons -- seems to be safe to ignore there # At least it was still failing on 2.6.2-0ubuntu1 (jaunty)