asyncserver.py

Go to the documentation of this file.

# 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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

# Author: Cyril Jaquier
# 
# $Revision: 567 $

__author__ = "Cyril Jaquier"
__version__ = "$Revision: 567 $"
__date__ = "$Date: 2007-03-26 23:17:31 +0200 (Mon, 26 Mar 2007) $"
__copyright__ = "Copyright (c) 2004 Cyril Jaquier"
__license__ = "GPL"

from pickle import dumps, loads, HIGHEST_PROTOCOL
import asyncore, asynchat, socket, os, logging

# Gets the instance of the logger.
logSys = logging.getLogger("fail2ban.server")

##
# Request handler class.
#
# This class extends asynchat in order to provide a request handler for
# incoming query.

class RequestHandler(asynchat.async_chat):
    
    END_STRING = "<F2B_END_COMMAND>"

    def __init__(self, conn, transmitter):
        asynchat.async_chat.__init__(self, conn)
        self.__transmitter = transmitter
        self.__buffer = []
        # Sets the terminator.
        self.set_terminator(RequestHandler.END_STRING)

    def collect_incoming_data(self, data):
        #logSys.debug("Received raw data: " + str(data))
        self.__buffer.append(data)

    ##
    # Handles a new request.
    #
    # This method is called once we have a complete request.

    def found_terminator(self):
        # Joins the buffer items.
        message = loads("".join(self.__buffer))
        # Gives the message to the transmitter.
        message = self.__transmitter.proceed(message)
        # Serializes the response.
        message = dumps(message, HIGHEST_PROTOCOL)
        # Sends the response to the client.
        self.send(message + RequestHandler.END_STRING)
        # Closes the channel.
        self.close_when_done()
        
    def handle_error(self):
        logSys.error("Unexpected communication error")
        self.close()
        
##
# 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

    ##
    # Returns False as we only read the socket first.

    def writable(self):
        return False

    def handle_accept(self):
        try:
            conn, addr = self.accept()
        except socket.error:
            logSys.warning("Socket error")
            return
        except TypeError:
            logSys.warning("Type error")
            return
        # 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):
        self.__sock = sock
        # Remove socket
        if os.path.exists(sock):
            logSys.error("Fail2ban seems to be already running")
            if force:
                logSys.warn("Forcing execution of the server")
                os.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:
            raise AsyncServerException("Unable to bind socket %s" % self.__sock)
        self.listen(1)
        # Sets the init flag.
        self.__init = True
        # TODO Add try..catch
        asyncore.loop()
    
    ##
    # Stops the communication server.
    
    def stop(self):
        if self.__init:
            # Only closes the socket if it was initialized first.
            self.close()
        # Remove socket
        if os.path.exists(self.__sock):
            logSys.debug("Removed socket file " + self.__sock)
            os.remove(self.__sock)
        logSys.debug("Socket shutdown")


##
# AsyncServerException is used to wrap communication exceptions.

class AsyncServerException(Exception):
    pass