MessageServer.h

/*
 *  Phusion Passenger - http://www.modrails.com/
 *  Copyright (c) 2010 Phusion
 *
 *  "Phusion Passenger" is a trademark of Hongli Lai & Ninh Bui.
 *
 *  Permission is hereby granted, free of charge, to any person obtaining a copy
 *  of this software and associated documentation files (the "Software"), to deal
 *  in the Software without restriction, including without limitation the rights
 *  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 *  copies of the Software, and to permit persons to whom the Software is
 *  furnished to do so, subject to the following conditions:
 *
 *  The above copyright notice and this permission notice shall be included in
 *  all copies or substantial portions of the Software.
 *
 *  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 *  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 *  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 *  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 *  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 *  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 *  THE SOFTWARE.
 */
#ifndef _PASSENGER_MESSAGE_SERVER_H_
#define _PASSENGER_MESSAGE_SERVER_H_

#include <string>
#include <vector>

#include <boost/shared_ptr.hpp>
#include <boost/thread.hpp>
#include <oxt/system_calls.hpp>
#include <oxt/dynamic_thread_group.hpp>

#include <sys/types.h>
#include <sys/stat.h>
#include <sys/un.h>
#include <unistd.h>
#include <cerrno>
#include <cassert>

#include "Account.h"
#include "AccountsDatabase.h"
#include "Constants.h"
#include "FileDescriptor.h"
#include "MessageChannel.h"
#include "Logging.h"
#include "Exceptions.h"
#include "Utils.h"

namespace Passenger {

using namespace std;
using namespace boost;
using namespace oxt;


/* This source file follows the security guidelines written in Account.h. */

/**
 * Simple pluggable request/response messaging server framework.
 *
 * MessageServer implements a server with the following properties:
 * - It listens on a Unix socket. Socket creation and destruction is automatically handled.
 *   The socket is world-writable because a username/password authentication scheme is
 *   used to enforce security.
 * - Multithreaded: 1 thread per client.
 * - Designed for simple request/response cycles. That is, a client sends a request, and
 *   the server may respond with arbitrary data. The server does not respond sporadically,
 *   i.e. it only responds after a request.
 * - Requests are discrete MessageChannel array messages, not byte streams.
 * - Connections are authenticated. Connecting clients must send a username and password,
 *   which are then checked against an accounts database. The associated account is known
 *   throughout the entire connection life time so that it's possible to implement
 *   authorization features.
 *
 * MessageServer does not process messages by itself. Instead, one registers handlers
 * which handle message processing. This framework allows one to seperate message
 * handling code by function, while allowing everything to listen on the same socket and
 * to use a common request parsing and dispatching codebase.
 *
 * A username/password authentication scheme was chosen over a file permission scheme because
 * experience has shown that the latter is inadequate. For example, the web server may
 * consist of multiple worker processes, each running as a different user. Although ACLs
 * can solve this problem as well, not every platform supports ACLs by default.
 *
 * <h2>Writing handlers</h2>
 * Handlers must inherit from MessageServer::Handler. They may implement newClient()
 * and must implement processMessage().
 *
 * When a new client is accepted, MessageServer will call newClient() on all handlers.
 * This method accepts one argument: a common client context object. This context object
 * contains client-specific information, such as its file descriptor. It cannot be
 * extended to store more information, but it is passed to every handler anyway,
 * hence the word "common" in its name.
 * newClient() is supposed to return a handler-specific client context object for storing
 * its own information, or a null pointer if it doesn't need to store anything.
 *
 * When a client sends a request, MessageServer iterates through all handlers and calls
 * processMessage() on each one, passing it the common client context and the
 * handler-specific client context. processMessage() may return either true or false;
 * true indicates that the handler processed the message, false indicates that
 * it did not. Iteration stops at the first handler that returns true.
 * If all handlers return false, i.e. the client sent a message that no handler recognizes,
 * then MessageServer will close the connection with the client.
 *
 * Handlers do not need to be thread-safe as long as they only operate on data in the
 * context objects. MessageServer ensures that context objects are not shared with other
 * threads.
 *
 * <h2>Usage example</h2>
 * This implements a simple ping server. Every time a "ping" request is sent, the
 * server responds with "pong" along with the number of times it had already sent
 * pong to the same client in the past.
 *
 * @code
 *   class PingHandler: public MessageServer::Handler {
 *   public:
 *       struct MyContext: public MessageServer::ClientContext {
 *           int count;
 *           
 *           MyContext() {
 *               count = 0;
 *           }
 *       };
 *       
 *       MessageServer::ClientContextPtr newClient(MessageServer::CommonClientContext &commonContext) {
 *           return MessageServer::ClientContextPtr(new MyContext());
 *       }
 *       
 *       bool processMessage(MessageServer::CommonClientContext &commonContext,
 *                           MessageServer::ClientContextPtr &specificContext,
 *                           const vector<string> &args)
 *       {
 *           if (args[0] == "ping") {
 *               MyContext *myContext = (MyContext *) specificContext.get();
 *               commonContext.channel.write("pong", toString(specificContext->count).c_str(), NULL);
 *               specificContext->count++;
 *               return true;
 *           } else {
 *               return false;
 *           }
 *       }
 *   };
 *   
 *   ...
 *   
 *   MessageServer server("server.sock");
 *   server.addHandler(MessageServer::HandlerPtr(new PingHandler()));
 *   server.addHandler(MessageServer::HandlerPtr(new PingHandler()));
 *   server.mainLoop();
 * @endcode
 *
 * @ingroup Support
 */
class MessageServer {
public:
        static const unsigned int CLIENT_THREAD_STACK_SIZE = 64 * 1024;
        
        /** Interface for client context objects. */
        class ClientContext {
        public:
                virtual ~ClientContext() { }
        };
        
        typedef shared_ptr<ClientContext> ClientContextPtr;
        
        /**
         * A common client context, containing client-specific information
         * used by MessageServer itself.
         */
        class CommonClientContext: public ClientContext {
        public:
                /** The client's socket file descriptor. */
                FileDescriptor fd;
                
                /** The channel that's associated with the client's socket. */
                MessageChannel channel;
                
                /** The account with which the client authenticated. */
                AccountPtr account;
                
                
                CommonClientContext(FileDescriptor &theFd, AccountPtr &theAccount)
                        : fd(theFd), channel(theFd), account(theAccount)
                { }
                
                /** Returns a string representation for this client context. */
                string name() {
                        return toString(channel.filenum());
                }
                
                /**
                 * Checks whether this client has all of the rights in <tt>rights</tt>. The
                 * client will be notified about the result of this check, by sending it a
                 * message.
                 *
                 * @throws SecurityException The client doesn't have one of the required rights.
                 * @throws SystemException Something went wrong while communicating with the client.
                 * @throws boost::thread_interrupted
                 */
                void requireRights(Account::Rights rights) {
                        if (!account->hasRights(rights)) {
                                P_TRACE(2, "Security error: insufficient rights to execute this command.");
                                channel.write("SecurityException", "Insufficient rights to execute this command.", NULL);
                                throw SecurityException("Insufficient rights to execute this command.");
                        } else {
                                channel.write("Passed security", NULL);
                        }
                }
        };
        
        /**
         * An abstract message handler class.
         *
         * The methods defined in this class are allowed to throw arbitrary exceptions.
         * Such exceptions are caught and logged, after which the connection to the
         * client is closed.
         */
        class Handler {
        public:
                virtual ~Handler() { }
                
                /**
                 * Called when a new client has connected to the MessageServer.
                 *
                 * This method is called after the client has authenticated itself.
                 *
                 * @param context Contains common client-specific information.
                 * @return A client context object for storing handler-specific client
                 *         information, or null. The default implementation returns null.
                 */
                virtual ClientContextPtr newClient(CommonClientContext &context) {
                        return ClientContextPtr();
                }
                
                /**
                 * Called when a client has disconnected from the MessageServer. The
                 * default implementation does nothing.
                 *
                 * This method is called even if processMessage() throws an exception.
                 * It is however not called if newClient() throws an exception.
                 *
                 * @param commonContext Contains common client-specific information.
                 * @param handlerSpecificContext The client context object as was returned
                 *     earlier by newClient().
                 */
                virtual void clientDisconnected(MessageServer::CommonClientContext &context,
                                                MessageServer::ClientContextPtr &handlerSpecificContext)
                { }
                
                /**
                 * Called then a client has sent a request message.
                 *
                 * This method is called after newClient() is called.
                 *
                 * @param commonContext Contains common client-specific information.
                 * @param handlerSpecificContext The client context object as was returned
                 *     earlier by newClient().
                 * @param args The request message's contents.
                 * @return Whether this handler has processed the message. Return false
                 *         if the message is unrecognized.
                 */
                virtual bool processMessage(CommonClientContext &commonContext,
                                            ClientContextPtr &handlerSpecificContext,
                                            const vector<string> &args) = 0;
        };
        
        typedef shared_ptr<Handler> HandlerPtr;
        
protected:
        /** The filename of the server socket on which this MessageServer is listening. */
        string socketFilename;
        
        /** An accounts database, used for authenticating clients. */
        AccountsDatabasePtr accountsDatabase;
        
        /** The registered message handlers. */
        vector<HandlerPtr> handlers;
        
        /** The maximum number of milliseconds that client may spend on logging in.
         * Clients that take longer are disconnected.
         *
         * @invariant loginTimeout != 0
         */
        unsigned long long loginTimeout;
        
        /** The client threads. */
        dynamic_thread_group threadGroup;
        
        /** The server socket's file descriptor.
         * @invariant serverFd >= 0
         */
        int serverFd;
        
        
        /** Calls clientDisconnected() on all handlers when destroyed. */
        struct DisconnectEventBroadcastGuard {
                vector<HandlerPtr> &handlers;
                CommonClientContext &commonContext;
                vector<ClientContextPtr> &handlerSpecificContexts;
                
                DisconnectEventBroadcastGuard(vector<HandlerPtr> &_handlers,
                                              CommonClientContext &_commonContext,
                                              vector<ClientContextPtr> &_handlerSpecificContexts)
                : handlers(_handlers),
                  commonContext(_commonContext),
                  handlerSpecificContexts(_handlerSpecificContexts)
                { }
                
                ~DisconnectEventBroadcastGuard() {
                        vector<HandlerPtr>::iterator handler_iter;
                        vector<ClientContextPtr>::iterator context_iter;

                        for (handler_iter = handlers.begin(), context_iter = handlerSpecificContexts.begin();
                             handler_iter != handlers.end();
                             handler_iter++, context_iter++) {
                                (*handler_iter)->clientDisconnected(commonContext, *context_iter);
                        }
                }
        };
        
        
        /**
         * Create a server socket and set it up for listening. This socket will
         * be world-writable.
         *
         * @throws RuntimeException
         * @throws SystemException
         * @throws boost::thread_interrupted
         */
        void startListening() {
                TRACE_POINT();
                int ret;
                
                serverFd = createUnixServer(socketFilename.c_str());
                do {
                        ret = chmod(socketFilename.c_str(),
                                S_ISVTX |
                                S_IRUSR | S_IWUSR | S_IXUSR |
                                S_IRGRP | S_IWGRP | S_IXGRP |
                                S_IROTH | S_IWOTH | S_IXOTH);
                } while (ret == -1 && errno == EINTR);
        }
        
        /**
         * Authenticate the given client and returns its account information.
         *
         * @return A smart pointer to an Account object, or NULL if authentication failed.
         */
        AccountPtr authenticate(FileDescriptor &client) {
                MessageChannel channel(client);
                string username, password;
                MemZeroGuard passwordGuard(password);
                unsigned long long timeout = loginTimeout;
                
                try {
                        try {
                                if (!channel.readScalar(username, 50, &timeout)) {
                                        return AccountPtr();
                                }
                        } catch (const SecurityException &) {
                                channel.write("The supplied username is too long.", NULL);
                                return AccountPtr();
                        }
                        
                        try {
                                if (!channel.readScalar(password, MESSAGE_SERVER_MAX_PASSWORD_SIZE, &timeout)) {
                                        return AccountPtr();
                                }
                        } catch (const SecurityException &) {
                                channel.write("The supplied password is too long.", NULL);
                                return AccountPtr();
                        }
                        
                        AccountPtr account = accountsDatabase->authenticate(username, password);
                        passwordGuard.zeroNow();
                        if (account == NULL) {
                                channel.write("Invalid username or password.", NULL);
                                return AccountPtr();
                        } else {
                                channel.write("ok", NULL);
                                return account;
                        }
                } catch (const SystemException &) {
                        return AccountPtr();
                } catch (const TimeoutException &) {
                        return AccountPtr();
                }
        }
        
        void broadcastNewClientEvent(CommonClientContext &context,
                                     vector<ClientContextPtr> &handlerSpecificContexts) {
                vector<HandlerPtr>::iterator it;
                
                for (it = handlers.begin(); it != handlers.end(); it++) {
                        handlerSpecificContexts.push_back((*it)->newClient(context));
                }
        }
        
        bool processMessage(CommonClientContext &commonContext,
                            vector<ClientContextPtr> &handlerSpecificContexts,
                            const vector<string> &args) {
                vector<HandlerPtr>::iterator handler_iter;
                vector<ClientContextPtr>::iterator context_iter;
                
                for (handler_iter = handlers.begin(), context_iter = handlerSpecificContexts.begin();
                     handler_iter != handlers.end();
                     handler_iter++, context_iter++) {
                        if ((*handler_iter)->processMessage(commonContext, *context_iter, args)) {
                                return true;
                        }
                }
                return false;
        }
        
        void processUnknownMessage(CommonClientContext &commonContext, const vector<string> &args) {
                TRACE_POINT();
                string name;
                if (args.empty()) {
                        name = "(null)";
                } else {
                        name = args[0];
                }
                P_TRACE(2, "A MessageServer client sent an invalid command: "
                        << name << " (" << args.size() << " elements)");
        }
        
        /**
         * The main function for a thread which handles a client.
         */
        void clientHandlingMainLoop(FileDescriptor &client) {
                TRACE_POINT();
                vector<string> args;
                
                P_TRACE(4, "MessageServer client thread " << (int) client << " started.");
                
                try {
                        AccountPtr account(authenticate(client));
                        if (account == NULL) {
                                P_TRACE(4, "MessageServer client thread " << (int) client << " exited.");
                                return;
                        }
                        
                        CommonClientContext commonContext(client, account);
                        vector<ClientContextPtr> handlerSpecificContexts;
                        broadcastNewClientEvent(commonContext, handlerSpecificContexts);
                        DisconnectEventBroadcastGuard dguard(handlers, commonContext, handlerSpecificContexts);
                        
                        while (!this_thread::interruption_requested()) {
                                UPDATE_TRACE_POINT();
                                if (!commonContext.channel.read(args)) {
                                        // Client closed connection.
                                        break;
                                }
                                
                                P_TRACE(4, "MessageServer client " << commonContext.name() <<
                                        ": received message: " << toString(args));
                                
                                UPDATE_TRACE_POINT();
                                if (!processMessage(commonContext, handlerSpecificContexts, args)) {
                                        processUnknownMessage(commonContext, args);
                                        break;
                                }
                                args.clear();
                        }
                        
                        client.close();
                        P_TRACE(4, "MessageServer client thread " << (int) client << " exited.");
                } catch (const boost::thread_interrupted &) {
                        P_TRACE(2, "MessageServer client thread " << (int) client << " interrupted.");
                } catch (const tracable_exception &e) {
                        P_TRACE(2, "An error occurred in a MessageServer client thread " << (int) client << ":\n"
                                << "   message: " << toString(args) << "\n"
                                << "   exception: " << e.what() << "\n"
                                << "   backtrace:\n" << e.backtrace());
                } catch (const std::exception &e) {
                        P_TRACE(2, "An error occurred in a MessageServer client thread " << (int) client <<":\n"
                                << "   message: " << toString(args) << "\n"
                                << "   exception: " << e.what() << "\n"
                                << "   backtrace: not available");
                } catch (...) {
                        P_TRACE(2, "An unknown exception occurred in a MessageServer client thread.");
                }
        }
        
public:
        /**
         * Creates a new MessageServer object.
         * The actual server main loop is not started until you call mainLoop().
         *
         * @param socketFilename The socket filename on which this MessageServer
         *                       should be listening.
         * @param accountsDatabase An accounts database for this server, used for
         *                         authenticating clients.
         * @throws RuntimeException Something went wrong while setting up the server socket.
         * @throws SystemException Something went wrong while setting up the server socket.
         * @throws boost::thread_interrupted
         */
        MessageServer(const string &socketFilename, AccountsDatabasePtr accountsDatabase) {
                this->socketFilename   = socketFilename;
                this->accountsDatabase = accountsDatabase;
                loginTimeout = 2000;
                startListening();
        }
        
        ~MessageServer() {
                this_thread::disable_syscall_interruption dsi;
                syscalls::close(serverFd);
                syscalls::unlink(socketFilename.c_str());
        }
        
        string getSocketFilename() const {
                return socketFilename;
        }
        
        /**
         * Starts the server main loop. This method will loop forever until some
         * other thread interrupts the calling thread, or until an exception is raised.
         *
         * @throws SystemException Unable to accept a new connection. If this is a
         *                         non-fatal error then you may call mainLoop() again
         *                         to restart the server main loop.
         * @throws boost::thread_interrupted The calling thread has been interrupted.
         */
        void mainLoop() {
                TRACE_POINT();
                while (true) {
                        this_thread::interruption_point();
                        sockaddr_un addr;
                        socklen_t len = sizeof(addr);
                        FileDescriptor fd;
                        
                        UPDATE_TRACE_POINT();
                        fd = syscalls::accept(serverFd, (struct sockaddr *) &addr, &len);
                        if (fd == -1) {
                                throw SystemException("Unable to accept a new client", errno);
                        }
                        
                        UPDATE_TRACE_POINT();
                        this_thread::disable_interruption di;
                        this_thread::disable_syscall_interruption dsi;
                        
                        function<void ()> func(boost::bind(&MessageServer::clientHandlingMainLoop,
                                this, fd));
                        string name = "MessageServer client thread ";
                        name.append(toString(fd));
                        threadGroup.create_thread(func, name, CLIENT_THREAD_STACK_SIZE);
                }
        }
        
        /**
         * Registers a new handler.
         * 
         * @pre The main loop isn't running.
         */
        void addHandler(HandlerPtr handler) {
                handlers.push_back(handler);
        }
        
        /**
         * Sets the maximum number of milliseconds that clients may spend on logging in.
         * Clients that take longer are disconnected.
         *
         * @pre timeout != 0
         * @pre The main loop isn't running.
         */
        void setLoginTimeout(unsigned long long timeout) {
                assert(timeout != 0);
                loginTimeout = timeout;
        }
};

typedef shared_ptr<MessageServer> MessageServerPtr;

} // namespace Passenger

#endif /* _PASSENGER_MESSAGE_SERVER_H_ */

---