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