Viewing file:  asyncserver.py (9.97 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
# 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.
# Author: Cyril Jaquier
#
__author__ = "Cyril Jaquier"
__copyright__ = "Copyright (c) 2004 Cyril Jaquier"
__license__ = "GPL"
from pickle import dumps, loads, HIGHEST_PROTOCOL
import asynchat
import asyncore
import errno
import fcntl
import os
import socket
import sys
import threading
import traceback
from .utils import Utils
from ..protocol import CSPROTO
from ..helpers import logging, getLogger, formatExceptionInfo
# Gets the instance of the logger.
logSys = getLogger(__name__)
##
# Request handler class.
#
# This class extends asynchat in order to provide a request handler for
# incoming query.
class RequestHandler(asynchat.async_chat):
def __init__(self, conn, transmitter):
asynchat.async_chat.__init__(self, conn)
self.__conn = conn
self.__transmitter = transmitter
self.__buffer = []
# Sets the terminator.
self.set_terminator(CSPROTO.END)
def __close(self):
if self.__conn:
conn = self.__conn
self.__conn = None
try:
conn.shutdown(socket.SHUT_RDWR)
conn.close()
except socket.error: # pragma: no cover - normally unreachable
pass
def handle_close(self):
self.__close()
asynchat.async_chat.handle_close(self)
def collect_incoming_data(self, data):
#logSys.debug("Received raw data: " + str(data))
self.__buffer.append(data)
# exception identifies deserialization errors (exception by load in pickle):
class LoadError(Exception):
pass
##
# Handles a new request.
#
# This method is called once we have a complete request.
def found_terminator(self):
try:
# Pop whole buffer
message = self.__buffer
self.__buffer = []
# Joins the buffer items.
message = CSPROTO.EMPTY.join(message)
# Closes the channel if close was received
if message == CSPROTO.CLOSE:
self.close_when_done()
return
# Deserialize
try:
message = loads(message)
except Exception as e:
logSys.error('PROTO-error: load message failed: %s', e,
exc_info=logSys.getEffectiveLevel()<logging.DEBUG)
raise RequestHandler.LoadError(e)
# Gives the message to the transmitter.
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:
if not isinstance(e, RequestHandler.LoadError): # pragma: no cover - normally unreachable
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):
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, err_count=None):
"""Custom event loop implementation
Uses poll instead of loop to respect `active` flag,
to avoid loop timeout mistake: different in poll and poll2 (sec vs ms),
and to prevent sporadic errors like EBADF 'Bad file descriptor' etc. (see gh-161)
"""
if not err_count: err_count={}
err_count['listen'] = 0
if timeout is None:
timeout = Utils.DEFAULT_SLEEP_TIME
poll = asyncore.poll
if callable(use_poll):
poll = use_poll
elif use_poll and asyncore.poll2 and hasattr(asyncore.select, 'poll'): # pragma: no cover
logSys.debug('Server listener (select) uses poll')
# poll2 expected a timeout in milliseconds (but poll and loop in seconds):
timeout = float(timeout) / 1000
poll = asyncore.poll2
# Poll as long as active:
while active():
try:
poll(timeout)
if err_count['listen']:
err_count['listen'] -= 1
except Exception as e:
if not active():
break
err_count['listen'] += 1
if err_count['listen'] < 20:
# 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 err_count['listen'] == 20:
logSys.exception(e)
logSys.error('Too many errors - stop logging connection errors')
elif err_count['listen'] > 100: # pragma: no cover - normally unreachable
if (
e.args[0] == errno.EMFILE # [Errno 24] Too many open files
or sum(err_count.values()) > 1000
):
logSys.critical("Too many errors - critical count reached %r", err_count)
break
##
# Asynchronous server class.
#
# This class extends asyncore and dispatches connection requests to
# RequestHandler.
class AsyncServer(asyncore.dispatcher):
def __init__(self, transmitter):
asyncore.dispatcher.__init__(self)
self.__transmitter = transmitter
self.__sock = "/var/run/fail2ban/fail2ban.sock"
self.__init = False
self.__active = False
self.__errCount = {'accept': 0, 'listen': 0}
self.onstart = None
##
# Returns False as we only read the socket first.
def writable(self):
return False
def handle_accept(self):
try:
conn, addr = self.accept()
except Exception as e: # pragma: no cover
self.__errCount['accept'] += 1
if self.__errCount['accept'] < 20:
logSys.warning("Accept socket error: %s", e,
exc_info=(self.__errCount['accept'] <= 1))
elif self.__errCount['accept'] == 20:
logSys.error("Too many acceptor errors - stop logging errors")
elif self.__errCount['accept'] > 100:
if (
(isinstance(e, socket.error) and e.args[0] == errno.EMFILE) # [Errno 24] Too many open files
or sum(self.__errCount.values()) > 1000
):
logSys.critical("Too many errors - critical count reached %r", self.__errCount)
self.stop()
return
if self.__errCount['accept']:
self.__errCount['accept'] -= 1;
AsyncServer.__markCloseOnExec(conn)
# Creates an instance of the handler class to handle the
# request/response on the incoming connection.
RequestHandler(conn, self.__transmitter)
##
# Starts the communication server.
#
# @param sock: socket file.
# @param force: remove the socket file if exists.
def start(self, sock, force, timeout=None, use_poll=False):
self.__worker = threading.current_thread()
self.__sock = sock
# Remove socket
if os.path.exists(sock):
logSys.error("Fail2ban seems to be already running")
if force:
logSys.warning("Forcing execution of the server")
self._remove_sock()
else:
raise AsyncServerException("Server already running")
# Creates the socket.
self.create_socket(socket.AF_UNIX, socket.SOCK_STREAM)
self.set_reuse_addr()
try:
self.bind(sock)
except Exception: # pragma: no cover
raise AsyncServerException("Unable to bind socket %s" % self.__sock)
AsyncServer.__markCloseOnExec(self.socket)
self.listen(1)
# Sets the init flag.
self.__init = self.__loop = self.__active = True
# Execute on start event (server ready):
if self.onstart:
self.onstart()
# Event loop as long as active:
loop(lambda: self.__loop, timeout=timeout, use_poll=use_poll, err_count=self.__errCount)
self.__active = False
# Cleanup all
self.stop()
def close(self):
stopflg = False
if self.__active:
self.__loop = False
# shutdown socket here:
if self.socket:
try:
self.socket.shutdown(socket.SHUT_RDWR)
except socket.error: # pragma: no cover - normally unreachable
pass
# close connection:
asyncore.dispatcher.close(self)
# If not the loop thread (stops self in handler), wait (a little bit)
# for the server leaves loop, before remove socket
if threading.current_thread() != self.__worker:
Utils.wait_for(lambda: not self.__active, 1)
stopflg = True
# Remove socket (file) only if it was created:
if self.__init and os.path.exists(self.__sock):
self._remove_sock()
logSys.debug("Removed socket file " + self.__sock)
if stopflg:
logSys.debug("Socket shutdown")
self.__active = False
##
# Stops the communication server.
def stop_communication(self):
if self.__transmitter:
logSys.debug("Stop communication, shutdown")
self.__transmitter = None
##
# Stops the server.
def stop(self):
self.stop_communication()
self.close()
# better remains a method (not a property) since used as a callable for wait_for
def isActive(self):
return self.__active
##
# Safe remove (in multithreaded mode):
def _remove_sock(self):
try:
os.remove(self.__sock)
except OSError as e: # pragma: no cover
if e.errno != errno.ENOENT:
raise
##
# Marks socket as close-on-exec to avoid leaking file descriptors when
# running actions involving command execution.
# @param sock: socket file.
@staticmethod
def __markCloseOnExec(sock):
fd = sock.fileno()
flags = fcntl.fcntl(fd, fcntl.F_GETFD)
fcntl.fcntl(fd, fcntl.F_SETFD, flags|fcntl.FD_CLOEXEC)
##
# AsyncServerException is used to wrap communication exceptions.
class AsyncServerException(Exception):
pass
|