- Main Page
- Modules
- Namespaces
- Classes
- Files

- File List

# MessageServer.h

    00001 /*
    00002  *  Phusion Passenger - http://www.modrails.com/
    00003  *  Copyright (c) 2010 Phusion
    00004  *
    00005  *  "Phusion Passenger" is a trademark of Hongli Lai & Ninh Bui.
    00006  *
    00007  *  Permission is hereby granted, free of charge, to any person obtaining a copy
    00008  *  of this software and associated documentation files (the "Software"), to deal
    00009  *  in the Software without restriction, including without limitation the rights
    00010  *  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
    00011  *  copies of the Software, and to permit persons to whom the Software is
    00012  *  furnished to do so, subject to the following conditions:
    00013  *
    00014  *  The above copyright notice and this permission notice shall be included in
    00015  *  all copies or substantial portions of the Software.
    00016  *
    00017  *  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
    00018  *  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
    00019  *  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
    00020  *  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
    00021  *  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
    00022  *  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
    00023  *  THE SOFTWARE.
    00024  */
    00025 #ifndef _PASSENGER_MESSAGE_SERVER_H_
    00026 #define _PASSENGER_MESSAGE_SERVER_H_
    00027
    00028 #include <string>
    00029 #include <vector>
    00030
    00031 #include <boost/shared_ptr.hpp>
    00032 #include <boost/thread.hpp>
    00033 #include <oxt/system_calls.hpp>
    00034 #include <oxt/dynamic_thread_group.hpp>
    00035
    00036 #include <sys/types.h>
    00037 #include <sys/stat.h>
    00038 #include <sys/un.h>
    00039 #include <unistd.h>
    00040 #include <cerrno>
    00041 #include <cassert>
    00042
    00043 #include "Account.h"
    00044 #include "AccountsDatabase.h"
    00045 #include "Constants.h"
    00046 #include "FileDescriptor.h"
    00047 #include "MessageChannel.h"
    00048 #include "Logging.h"
    00049 #include "Exceptions.h"
    00050 #include "Utils.h"
    00051
    00052 namespace Passenger {
    00053
    00054 using namespace std;
    00055 using namespace boost;
    00056 using namespace oxt;
    00057
    00058
    00059 /* This source file follows the security guidelines written in Account.h. */
    00060
    00061 /**
    00062  * Simple pluggable request/response messaging server framework.
    00063  *
    00064  * MessageServer implements a server with the following properties:
    00065  * - It listens on a Unix socket. Socket creation and destruction is automatically handled.
    00066  *   The socket is world-writable because a username/password authentication scheme is
    00067  *   used to enforce security.
    00068  * - Multithreaded: 1 thread per client.
    00069  * - Designed for simple request/response cycles. That is, a client sends a request, and
    00070  *   the server may respond with arbitrary data. The server does not respond sporadically,
    00071  *   i.e. it only responds after a request.
    00072  * - Requests are discrete MessageChannel array messages, not byte streams.
    00073  * - Connections are authenticated. Connecting clients must send a username and password,
    00074  *   which are then checked against an accounts database. The associated account is known
    00075  *   throughout the entire connection life time so that it's possible to implement
    00076  *   authorization features.
    00077  *
    00078  * MessageServer does not process messages by itself. Instead, one registers handlers
    00079  * which handle message processing. This framework allows one to seperate message
    00080  * handling code by function, while allowing everything to listen on the same socket and
    00081  * to use a common request parsing and dispatching codebase.
    00082  *
    00083  * A username/password authentication scheme was chosen over a file permission scheme because
    00084  * experience has shown that the latter is inadequate. For example, the web server may
    00085  * consist of multiple worker processes, each running as a different user. Although ACLs
    00086  * can solve this problem as well, not every platform supports ACLs by default.
    00087  *
    00088  * <h2>Writing handlers</h2>
    00089  * Handlers must inherit from MessageServer::Handler. They may implement newClient()
    00090  * and must implement processMessage().
    00091  *
    00092  * When a new client is accepted, MessageServer will call newClient() on all handlers.
    00093  * This method accepts one argument: a common client context object. This context object
    00094  * contains client-specific information, such as its file descriptor. It cannot be
    00095  * extended to store more information, but it is passed to every handler anyway,
    00096  * hence the word "common" in its name.
    00097  * newClient() is supposed to return a handler-specific client context object for storing
    00098  * its own information, or a null pointer if it doesn't need to store anything.
    00099  *
    00100  * When a client sends a request, MessageServer iterates through all handlers and calls
    00101  * processMessage() on each one, passing it the common client context and the
    00102  * handler-specific client context. processMessage() may return either true or false;
    00103  * true indicates that the handler processed the message, false indicates that
    00104  * it did not. Iteration stops at the first handler that returns true.
    00105  * If all handlers return false, i.e. the client sent a message that no handler recognizes,
    00106  * then MessageServer will close the connection with the client.
    00107  *
    00108  * Handlers do not need to be thread-safe as long as they only operate on data in the
    00109  * context objects. MessageServer ensures that context objects are not shared with other
    00110  * threads.
    00111  *
    00112  * <h2>Usage example</h2>
    00113  * This implements a simple ping server. Every time a "ping" request is sent, the
    00114  * server responds with "pong" along with the number of times it had already sent
    00115  * pong to the same client in the past.
    00116  *
    00117  * @code
    00118  *   class PingHandler: public MessageServer::Handler {
    00119  *   public:
    00120  *       struct MyContext: public MessageServer::ClientContext {
    00121  *           int count;
    00122  *
    00123  *           MyContext() {
    00124  *               count = 0;
    00125  *           }
    00126  *       };
    00127  *
    00128  *       MessageServer::ClientContextPtr newClient(MessageServer::CommonClientContext &commonContext) {
    00129  *           return MessageServer::ClientContextPtr(new MyContext());
    00130  *       }
    00131  *
    00132  *       bool processMessage(MessageServer::CommonClientContext &commonContext,
    00133  *                           MessageServer::ClientContextPtr &specificContext,
    00134  *                           const vector<string> &args)
    00135  *       {
    00136  *           if (args[0] == "ping") {
    00137  *               MyContext *myContext = (MyContext *) specificContext.get();
    00138  *               commonContext.channel.write("pong", toString(specificContext->count).c_str(), NULL);
    00139  *               specificContext->count++;
    00140  *               return true;
    00141  *           } else {
    00142  *               return false;
    00143  *           }
    00144  *       }
    00145  *   };
    00146  *
    00147  *   ...
    00148  *
    00149  *   MessageServer server("server.sock");
    00150  *   server.addHandler(MessageServer::HandlerPtr(new PingHandler()));
    00151  *   server.addHandler(MessageServer::HandlerPtr(new PingHandler()));
    00152  *   server.mainLoop();
    00153  * @endcode
    00154  *
    00155  * @ingroup Support
    00156  */
    00157 class MessageServer {
    00158 public:
    00159         static const unsigned int CLIENT_THREAD_STACK_SIZE = 64 * 1024;
    00160
    00161         /** Interface for client context objects. */
    00162         class ClientContext {
    00163         public:
    00164                 virtual ~ClientContext() { }
    00165         };
    00166
    00167         typedef shared_ptr<ClientContext> ClientContextPtr;
    00168
    00169         /**
    00170          * A common client context, containing client-specific information
    00171          * used by MessageServer itself.
    00172          */
    00173         class CommonClientContext: public ClientContext {
    00174         public:
    00175                 /** The client's socket file descriptor. */
    00176                 FileDescriptor fd;
    00177
    00178                 /** The channel that's associated with the client's socket. */
    00179                 MessageChannel channel;
    00180
    00181                 /** The account with which the client authenticated. */
    00182                 AccountPtr account;
    00183
    00184
    00185                 CommonClientContext(FileDescriptor &theFd, AccountPtr &theAccount)
    00186                         : fd(theFd), channel(theFd), account(theAccount)
    00187                 { }
    00188
    00189                 /** Returns a string representation for this client context. */
    00190                 string name() {
    00191                         return toString(channel.filenum());
    00192                 }
    00193
    00194                 /**
    00195                  * Checks whether this client has all of the rights in <tt>rights</tt>. The
    00196                  * client will be notified about the result of this check, by sending it a
    00197                  * message.
    00198                  *
    00199                  * @throws SecurityException The client doesn't have one of the required rights.
    00200                  * @throws SystemException Something went wrong while communicating with the client.
    00201                  * @throws boost::thread_interrupted
    00202                  */
    00203                 void requireRights(Account::Rights rights) {
    00204                         if (!account->hasRights(rights)) {
    00205                                 P_TRACE(2, "Security error: insufficient rights to execute this command.");
    00206                                 channel.write("SecurityException", "Insufficient rights to execute this command.", NULL);
    00207                                 throw SecurityException("Insufficient rights to execute this command.");
    00208                         } else {
    00209                                 channel.write("Passed security", NULL);
    00210                         }
    00211                 }
    00212         };
    00213
    00214         /**
    00215          * An abstract message handler class.
    00216          *
    00217          * The methods defined in this class are allowed to throw arbitrary exceptions.
    00218          * Such exceptions are caught and logged, after which the connection to the
    00219          * client is closed.
    00220          */
    00221         class Handler {
    00222         public:
    00223                 virtual ~Handler() { }
    00224
    00225                 /**
    00226                  * Called when a new client has connected to the MessageServer.
    00227                  *
    00228                  * This method is called after the client has authenticated itself.
    00229                  *
    00230                  * @param context Contains common client-specific information.
    00231                  * @return A client context object for storing handler-specific client
    00232                  *         information, or null. The default implementation returns null.
    00233                  */
    00234                 virtual ClientContextPtr newClient(CommonClientContext &context) {
    00235                         return ClientContextPtr();
    00236                 }
    00237
    00238                 /**
    00239                  * Called when a client has disconnected from the MessageServer. The
    00240                  * default implementation does nothing.
    00241                  *
    00242                  * This method is called even if processMessage() throws an exception.
    00243                  * It is however not called if newClient() throws an exception.
    00244                  *
    00245                  * @param commonContext Contains common client-specific information.
    00246                  * @param handlerSpecificContext The client context object as was returned
    00247                  *     earlier by newClient().
    00248                  */
    00249                 virtual void clientDisconnected(MessageServer::CommonClientContext &context,
    00250                                                 MessageServer::ClientContextPtr &handlerSpecificContext)
    00251                 { }
    00252
    00253                 /**
    00254                  * Called then a client has sent a request message.
    00255                  *
    00256                  * This method is called after newClient() is called.
    00257                  *
    00258                  * @param commonContext Contains common client-specific information.
    00259                  * @param handlerSpecificContext The client context object as was returned
    00260                  *     earlier by newClient().
    00261                  * @param args The request message's contents.
    00262                  * @return Whether this handler has processed the message. Return false
    00263                  *         if the message is unrecognized.
    00264                  */
    00265                 virtual bool processMessage(CommonClientContext &commonContext,
    00266                                             ClientContextPtr &handlerSpecificContext,
    00267                                             const vector<string> &args) = 0;
    00268         };
    00269
    00270         typedef shared_ptr<Handler> HandlerPtr;
    00271
    00272 protected:
    00273         /** The filename of the server socket on which this MessageServer is listening. */
    00274         string socketFilename;
    00275
    00276         /** An accounts database, used for authenticating clients. */
    00277         AccountsDatabasePtr accountsDatabase;
    00278
    00279         /** The registered message handlers. */
    00280         vector<HandlerPtr> handlers;
    00281
    00282         /** The maximum number of milliseconds that client may spend on logging in.
    00283          * Clients that take longer are disconnected.
    00284          *
    00285          * @invariant loginTimeout != 0
    00286          */
    00287         unsigned long long loginTimeout;
    00288
    00289         /** The client threads. */
    00290         dynamic_thread_group threadGroup;
    00291
    00292         /** The server socket's file descriptor.
    00293          * @invariant serverFd >= 0
    00294          */
    00295         int serverFd;
    00296
    00297
    00298         /** Calls clientDisconnected() on all handlers when destroyed. */
    00299         struct DisconnectEventBroadcastGuard {
    00300                 vector<HandlerPtr> &handlers;
    00301                 CommonClientContext &commonContext;
    00302                 vector<ClientContextPtr> &handlerSpecificContexts;
    00303
    00304                 DisconnectEventBroadcastGuard(vector<HandlerPtr> &_handlers,
    00305                                               CommonClientContext &_commonContext,
    00306                                               vector<ClientContextPtr> &_handlerSpecificContexts)
    00307                 : handlers(_handlers),
    00308                   commonContext(_commonContext),
    00309                   handlerSpecificContexts(_handlerSpecificContexts)
    00310                 { }
    00311
    00312                 ~DisconnectEventBroadcastGuard() {
    00313                         vector<HandlerPtr>::iterator handler_iter;
    00314                         vector<ClientContextPtr>::iterator context_iter;
    00315
    00316                         for (handler_iter = handlers.begin(), context_iter = handlerSpecificContexts.begin();
    00317                              handler_iter != handlers.end();
    00318                              handler_iter++, context_iter++) {
    00319                                 (*handler_iter)->clientDisconnected(commonContext, *context_iter);
    00320                         }
    00321                 }
    00322         };
    00323
    00324
    00325         /**
    00326          * Create a server socket and set it up for listening. This socket will
    00327          * be world-writable.
    00328          *
    00329          * @throws RuntimeException
    00330          * @throws SystemException
    00331          * @throws boost::thread_interrupted
    00332          */
    00333         void startListening() {
    00334                 TRACE_POINT();
    00335                 int ret;
    00336
    00337                 serverFd = createUnixServer(socketFilename.c_str());
    00338                 do {
    00339                         ret = chmod(socketFilename.c_str(),
    00340                                 S_ISVTX |
    00341                                 S_IRUSR | S_IWUSR | S_IXUSR |
    00342                                 S_IRGRP | S_IWGRP | S_IXGRP |
    00343                                 S_IROTH | S_IWOTH | S_IXOTH);
    00344                 } while (ret == -1 && errno == EINTR);
    00345         }
    00346
    00347         /**
    00348          * Authenticate the given client and returns its account information.
    00349          *
    00350          * @return A smart pointer to an Account object, or NULL if authentication failed.
    00351          */
    00352         AccountPtr authenticate(FileDescriptor &client) {
    00353                 MessageChannel channel(client);
    00354                 string username, password;
    00355                 MemZeroGuard passwordGuard(password);
    00356                 unsigned long long timeout = loginTimeout;
    00357
    00358                 try {
    00359                         try {
    00360                                 if (!channel.readScalar(username, 50, &timeout)) {
    00361                                         return AccountPtr();
    00362                                 }
    00363                         } catch (const SecurityException &) {
    00364                                 channel.write("The supplied username is too long.", NULL);
    00365                                 return AccountPtr();
    00366                         }
    00367
    00368                         try {
    00369                                 if (!channel.readScalar(password, MESSAGE_SERVER_MAX_PASSWORD_SIZE, &timeout)) {
    00370                                         return AccountPtr();
    00371                                 }
    00372                         } catch (const SecurityException &) {
    00373                                 channel.write("The supplied password is too long.", NULL);
    00374                                 return AccountPtr();
    00375                         }
    00376
    00377                         AccountPtr account = accountsDatabase->authenticate(username, password);
    00378                         passwordGuard.zeroNow();
    00379                         if (account == NULL) {
    00380                                 channel.write("Invalid username or password.", NULL);
    00381                                 return AccountPtr();
    00382                         } else {
    00383                                 channel.write("ok", NULL);
    00384                                 return account;
    00385                         }
    00386                 } catch (const SystemException &) {
    00387                         return AccountPtr();
    00388                 } catch (const TimeoutException &) {
    00389                         return AccountPtr();
    00390                 }
    00391         }
    00392
    00393         void broadcastNewClientEvent(CommonClientContext &context,
    00394                                      vector<ClientContextPtr> &handlerSpecificContexts) {
    00395                 vector<HandlerPtr>::iterator it;
    00396
    00397                 for (it = handlers.begin(); it != handlers.end(); it++) {
    00398                         handlerSpecificContexts.push_back((*it)->newClient(context));
    00399                 }
    00400         }
    00401
    00402         bool processMessage(CommonClientContext &commonContext,
    00403                             vector<ClientContextPtr> &handlerSpecificContexts,
    00404                             const vector<string> &args) {
    00405                 vector<HandlerPtr>::iterator handler_iter;
    00406                 vector<ClientContextPtr>::iterator context_iter;
    00407
    00408                 for (handler_iter = handlers.begin(), context_iter = handlerSpecificContexts.begin();
    00409                      handler_iter != handlers.end();
    00410                      handler_iter++, context_iter++) {
    00411                         if ((*handler_iter)->processMessage(commonContext, *context_iter, args)) {
    00412                                 return true;
    00413                         }
    00414                 }
    00415                 return false;
    00416         }
    00417
    00418         void processUnknownMessage(CommonClientContext &commonContext, const vector<string> &args) {
    00419                 TRACE_POINT();
    00420                 string name;
    00421                 if (args.empty()) {
    00422                         name = "(null)";
    00423                 } else {
    00424                         name = args[0];
    00425                 }
    00426                 P_TRACE(2, "A MessageServer client sent an invalid command: "
    00427                         << name << " (" << args.size() << " elements)");
    00428         }
    00429
    00430         /**
    00431          * The main function for a thread which handles a client.
    00432          */
    00433         void clientHandlingMainLoop(FileDescriptor &client) {
    00434                 TRACE_POINT();
    00435                 vector<string> args;
    00436
    00437                 P_TRACE(4, "MessageServer client thread " << (int) client << " started.");
    00438
    00439                 try {
    00440                         AccountPtr account(authenticate(client));
    00441                         if (account == NULL) {
    00442                                 P_TRACE(4, "MessageServer client thread " << (int) client << " exited.");
    00443                                 return;
    00444                         }
    00445
    00446                         CommonClientContext commonContext(client, account);
    00447                         vector<ClientContextPtr> handlerSpecificContexts;
    00448                         broadcastNewClientEvent(commonContext, handlerSpecificContexts);
    00449                         DisconnectEventBroadcastGuard dguard(handlers, commonContext, handlerSpecificContexts);
    00450
    00451                         while (!this_thread::interruption_requested()) {
    00452                                 UPDATE_TRACE_POINT();
    00453                                 if (!commonContext.channel.read(args)) {
    00454                                         // Client closed connection.
    00455                                         break;
    00456                                 }
    00457
    00458                                 P_TRACE(4, "MessageServer client " << commonContext.name() <<
    00459                                         ": received message: " << toString(args));
    00460
    00461                                 UPDATE_TRACE_POINT();
    00462                                 if (!processMessage(commonContext, handlerSpecificContexts, args)) {
    00463                                         processUnknownMessage(commonContext, args);
    00464                                         break;
    00465                                 }
    00466                                 args.clear();
    00467                         }
    00468
    00469                         client.close();
    00470                         P_TRACE(4, "MessageServer client thread " << (int) client << " exited.");
    00471                 } catch (const boost::thread_interrupted &) {
    00472                         P_TRACE(2, "MessageServer client thread " << (int) client << " interrupted.");
    00473                 } catch (const tracable_exception &e) {
    00474                         P_TRACE(2, "An error occurred in a MessageServer client thread " << (int) client << ":\n"
    00475                                 << "   message: " << toString(args) << "\n"
    00476                                 << "   exception: " << e.what() << "\n"
    00477                                 << "   backtrace:\n" << e.backtrace());
    00478                 } catch (const std::exception &e) {
    00479                         P_TRACE(2, "An error occurred in a MessageServer client thread " << (int) client <<":\n"
    00480                                 << "   message: " << toString(args) << "\n"
    00481                                 << "   exception: " << e.what() << "\n"
    00482                                 << "   backtrace: not available");
    00483                 } catch (...) {
    00484                         P_TRACE(2, "An unknown exception occurred in a MessageServer client thread.");
    00485                 }
    00486         }
    00487
    00488 public:
    00489         /**
    00490          * Creates a new MessageServer object.
    00491          * The actual server main loop is not started until you call mainLoop().
    00492          *
    00493          * @param socketFilename The socket filename on which this MessageServer
    00494          *                       should be listening.
    00495          * @param accountsDatabase An accounts database for this server, used for
    00496          *                         authenticating clients.
    00497          * @throws RuntimeException Something went wrong while setting up the server socket.
    00498          * @throws SystemException Something went wrong while setting up the server socket.
    00499          * @throws boost::thread_interrupted
    00500          */
    00501         MessageServer(const string &socketFilename, AccountsDatabasePtr accountsDatabase) {
    00502                 this->socketFilename   = socketFilename;
    00503                 this->accountsDatabase = accountsDatabase;
    00504                 loginTimeout = 2000;
    00505                 startListening();
    00506         }
    00507
    00508         ~MessageServer() {
    00509                 this_thread::disable_syscall_interruption dsi;
    00510                 syscalls::close(serverFd);
    00511                 syscalls::unlink(socketFilename.c_str());
    00512         }
    00513
    00514         string getSocketFilename() const {
    00515                 return socketFilename;
    00516         }
    00517
    00518         /**
    00519          * Starts the server main loop. This method will loop forever until some
    00520          * other thread interrupts the calling thread, or until an exception is raised.
    00521          *
    00522          * @throws SystemException Unable to accept a new connection. If this is a
    00523          *                         non-fatal error then you may call mainLoop() again
    00524          *                         to restart the server main loop.
    00525          * @throws boost::thread_interrupted The calling thread has been interrupted.
    00526          */
    00527         void mainLoop() {
    00528                 TRACE_POINT();
    00529                 while (true) {
    00530                         this_thread::interruption_point();
    00531                         sockaddr_un addr;
    00532                         socklen_t len = sizeof(addr);
    00533                         FileDescriptor fd;
    00534
    00535                         UPDATE_TRACE_POINT();
    00536                         fd = syscalls::accept(serverFd, (struct sockaddr *) &addr, &len);
    00537                         if (fd == -1) {
    00538                                 throw SystemException("Unable to accept a new client", errno);
    00539                         }
    00540
    00541                         UPDATE_TRACE_POINT();
    00542                         this_thread::disable_interruption di;
    00543                         this_thread::disable_syscall_interruption dsi;
    00544
    00545                         function<void ()> func(boost::bind(&MessageServer::clientHandlingMainLoop,
    00546                                 this, fd));
    00547                         string name = "MessageServer client thread ";
    00548                         name.append(toString(fd));
    00549                         threadGroup.create_thread(func, name, CLIENT_THREAD_STACK_SIZE);
    00550                 }
    00551         }
    00552
    00553         /**
    00554          * Registers a new handler.
    00555          *
    00556          * @pre The main loop isn't running.
    00557          */
    00558         void addHandler(HandlerPtr handler) {
    00559                 handlers.push_back(handler);
    00560         }
    00561
    00562         /**
    00563          * Sets the maximum number of milliseconds that clients may spend on logging in.
    00564          * Clients that take longer are disconnected.
    00565          *
    00566          * @pre timeout != 0
    00567          * @pre The main loop isn't running.
    00568          */
    00569         void setLoginTimeout(unsigned long long timeout) {
    00570                 assert(timeout != 0);
    00571                 loginTimeout = timeout;
    00572         }
    00573 };
    00574
    00575 typedef shared_ptr<MessageServer> MessageServerPtr;
    00576
    00577 } // namespace Passenger
    00578
    00579 #endif /* _PASSENGER_MESSAGE_SERVER_H_ */