HTTP.cpp

/*
** libutil++
** $Id: HTTP.cpp 1664 2016-04-04 03:57:40Z sella $
** Copyright (c) 2011-2016 Digital Genesis, LLC. All Rights Reserved.
** Released under the LGPL Version 2.1 License.
** http://www.digitalgenesis.com
*/

static const char rcsid[] __attribute__((used)) = "$Id: HTTP.cpp 1664 2016-04-04 03:57:40Z sella $";

#include "HTTP.h"
#include "Code.h"
#include "callbacks.h"
#include "../util/CommonMacro.h"
#include "../util/String.h"
#include "../util/File.h"

#include <cstddef>
#include <string>
#include <algorithm>

using namespace sella::server;
using namespace sella::net;
using namespace sella::util;

HTTP::HTTP() throw (ServerException) :
        libutilxx__log(::libutilxx__log),
        running(false),
        listeners(ListenersDefault()),
        secureListeners(SecureListenersDefault()),
        protocols(SecureTCPSocket::DefaultProtocols),
        preferServerCiphers(false),
        requestBodyLimit(RequestBodyLimitDefault),
        connectionLimit(ConnectionLimitDefault),
        root(RootDefault()),
        indexes(DirectoryIndexesDefault()),
        banner(BannerDefault()),
        allow(AllowDefault()),
        fallbackAcceptType(FallbackAcceptTypeDefault),
        compressionThreshold(CompressionThresholdDefault),
        compressionLevel(CompressionLevelDefault),
        aclCallback(NULL),
        authCallback(NULL),
        reqCallback(http::request_callback),
        errorCallback(http::error_callback)
{ }

HTTP::HTTP(Log::shared log) throw (ServerException) :
        libutilxx__log(log),
        running(false),
        listeners(ListenersDefault()),
        secureListeners(SecureListenersDefault()),
        protocols(SecureTCPSocket::DefaultProtocols),
        preferServerCiphers(false),
        requestBodyLimit(RequestBodyLimitDefault),
        connectionLimit(ConnectionLimitDefault),
        root(RootDefault()),
        indexes(DirectoryIndexesDefault()),
        banner(BannerDefault()),
        allow(AllowDefault()),
        fallbackAcceptType(FallbackAcceptTypeDefault),
        compressionThreshold(CompressionThresholdDefault),
        compressionLevel(CompressionLevelDefault),
        aclCallback(NULL),
        authCallback(NULL),
        reqCallback(http::request_callback),
        errorCallback(http::error_callback)
{ }

HTTP::~HTTP() throw () {
        if (isRunning()) {
                stop();
        } else {
                shutdown();
        }
}

bool HTTP::receive(void) throw (ServerException) {
        timeval tv = { 0, 0 };

        return receive(&tv);
}

bool HTTP::receive(timeval *tv) throw (ServerException) {
        int c;

        if (servers.empty()) {
                startup();
        }

        if ((c = select(tv)) > 0) {
                while (!pending.empty()) {
                        HTTPConnection::shared connection = pending.front();
                        pending.pop();

                        worker(connection);
                }
        }

        return (c > 0);
}

bool HTTP::run(size_t workers) throw (ServerException) {
        try {
                ThreadPool::shared pool = ThreadPool::shared_ptr(workers);

                start(pool);
        } catch (ThreadPoolException &e) {
                RETHROW(ServerException, e);
        }

        ::select(1, NULL, NULL, NULL, NULL);

        return stop();
}

bool HTTP::run(size_t maximumWorkers, size_t minimumWorkers, size_t spareWorkers) throw (ServerException) {
        try {
                ThreadPool::shared pool = ThreadPool::shared_ptr(maximumWorkers, minimumWorkers, spareWorkers);

                start(pool);
        } catch (ThreadPoolException &e) {
                RETHROW(ServerException, e);
        }

        ::select(1, NULL, NULL, NULL, NULL);

        return stop();
}

void HTTP::start(size_t workers) throw (ServerException) {
        try {
                ThreadPool::shared pool = ThreadPool::shared_ptr(workers);

                start(pool);
        } catch (ThreadPoolException &e) {
                RETHROW(ServerException, e);
        }
}

void HTTP::start(size_t maximumWorkers, size_t minimumWorkers, size_t spareWorkers) throw (ServerException) {
        try {
                ThreadPool::shared pool = ThreadPool::shared_ptr(maximumWorkers, minimumWorkers, spareWorkers);

                start(pool);
        } catch (ThreadPoolException &e) {
                RETHROW(ServerException, e);
        }
}

void HTTP::start(ThreadPool::shared pool) throw (ServerException) {
        std::lock_guard<std::mutex> lg(mutex);

        if (running) {
                THROW(ServerException, "HTTP server is already running");
        } else if (pool == NULL) {
                THROW(ServerException, "ThreadPool is null");
        }

        startup();
        this->pool = pool;
        running = true;

        try {
                manager = std::thread(&HTTP::server, this);
        } catch (std::exception &e) {
                TRACE(PACKAGE, "HTTP server failed to start");

                this->pool.reset();
                running = false;

                RETHROW_CODE(ServerException, 0, e);
        }

        TRACE(PACKAGE, "HTTP server started");
}

bool HTTP::stop(void) throw () {
        TRACE(PACKAGE, "stop");

        mutex.lock();

        if (running) {
                running = false;

                trigger.set();

                mutex.unlock();

                if (manager.joinable()) {
                        manager.join();
                }

                pool.reset();

                return true;
        }

        mutex.unlock();

        return false;
}

void HTTP::setListeners(const sella::net::IPAddressPort::vector &listeners) throw () {
        if (!listeners.empty()) {
                this->listeners = listeners;
        } else {
                this->listeners = ListenersDefault();
        }
}

void HTTP::setListeners(const std::string &listeners) throw () {
        this->listeners.clear();

        const auto &list = String::split(listeners, ",; ", false, false);

        for (auto l = list.begin(); l != list.end(); ++l) {
                try {
                        this->listeners.push_back(IPAddressPort(*l));
                } catch (Exception &e) {
                        TRACE(PACKAGE, "Bad listener: %s", e.what());
                }
        }

        if (this->listeners.empty()) {
                this->listeners = ListenersDefault();
        }
}

void HTTP::setSecureListeners(const sella::net::IPAddressPort::vector &secureListeners) throw () {
        if (!secureListeners.empty()) {
                this->secureListeners = secureListeners;
        } else {
                this->secureListeners = SecureListenersDefault();
        }
}

void HTTP::setSecureListeners(const std::string &secureListeners) throw () {
        this->secureListeners.clear();

        const auto &list = String::split(secureListeners, ",; ", false, false);

        for (auto l = list.begin(); l != list.end(); ++l) {
                try {
                        this->secureListeners.push_back(IPAddressPort(*l));
                } catch (Exception &e) {
                        TRACE(PACKAGE, "Bad listener: %s", e.what());
                }
        }

        if (this->secureListeners.empty()) {
                this->secureListeners = SecureListenersDefault();
        }
}

void HTTP::setSecureProtocols(const std::set<SecureTCPSocket::Protocol> &protocols) throw () {
        this->protocols = protocols;
}

void HTTP::setSecureProtocol(SecureTCPSocket::Protocol protocol, bool on) throw () {
        if (on) {
                this->protocols.insert(protocol);
        } else {
                this->protocols.erase(protocol);
        }
}

void HTTP::setSecureCipher(const std::string &ciphers) throw () {
        this->ciphers = ciphers;
}

void HTTP::setSecurePreferServerCipher(bool on) throw () {
        this->preferServerCiphers = on;
}

void HTTP::setCertificateFile(const std::string &certificateFile) throw () {
        this->certificateFile = certificateFile;
}

void HTTP::setPrivateKeyFile(const std::string &privateKeyFile) throw () {
        this->privateKeyFile = privateKeyFile;
}

void HTTP::setPasswordFile(const std::string &passwordFile) throw () {
        this->passwordFile = passwordFile;
}

void HTTP::setPassword(const std::string &password) throw () {
        /* Scrub previous buffer before release back to stack. */
        for (size_t i = 0; i < this->password.size(); i++) {
                this->password.at(i) = 0;
        }

        this->password = password;
}

void HTTP::setCertificateChainFile(const std::string &certificateChainFile) throw () {
        this->certificateChainFile = certificateChainFile;
}

void HTTP::setCACertificateFile(const std::string &caCertificateFile) throw () {
        this->caCertificateFile = caCertificateFile;
}

void HTTP::setRequestBodyLimit(size_t requestBodyLimit) throw () {
        if (requestBodyLimit > 0) {
                this->requestBodyLimit = requestBodyLimit;
        } else {
                this->requestBodyLimit = RequestBodyLimitDefault;
        }
}

void HTTP::setConnectionLimit(size_t connectionLimit) throw () {
        if (connectionLimit > 0) {
                this->connectionLimit = connectionLimit;
        } else {
                this->connectionLimit = ConnectionLimitDefault;
        }
}

void HTTP::setRoot(const std::string &root) throw (ServerException) {
        try {
                this->root = File::getRealPath(root);
        } catch (FileException &e) {
                RETHROW(ServerException, e);
        }
}

void HTTP::setDirectoryIndex(const std::vector<std::string> &indexes) throw () {
        this->indexes = indexes;
}

void HTTP::addDirectoryIndex(const std::string &index) throw () {
        if (std::find(indexes.begin(), indexes.end(), index) != indexes.end()) {
                this->indexes.push_back(index);
        }
}

void HTTP::setBanner(const std::string &banner) throw () {
        this->banner = banner;
}

void HTTP::setCustomHeaders(const std::vector<std::string> &headers) throw () {
        this->headers = headers;
}

void HTTP::setAllows(const std::set<HTTPRequest::Method> &methods) throw () {
        allow = methods;
}

void HTTP::setAllow(HTTPRequest::Method method) throw () {
        allow.insert(method);
}

void HTTP::setFallbackAcceptType(HTTPRequest::AcceptType fallbackAcceptType) throw () {
        this->fallbackAcceptType = fallbackAcceptType;
}

void HTTP::setCompressionThreshold(size_t threshold) throw () {
        if (threshold < 1) {
                this->compressionThreshold = 0;
        } else {
                this->compressionThreshold = threshold;
        }
}

void HTTP::setCompressionLevel(int level) throw () {
        if (level < 0) {
                this->compressionLevel = 0;
        } else if (level > 9) {
                this->compressionLevel = 9;
        } else {
                this->compressionLevel = level;
        }
}

void HTTP::setACLCallback(const acl_callback &callback) throw () {
        if (callback) {
                this->aclCallback = callback;
        } else {
                this->aclCallback = NULL;
        }
}

void HTTP::setAuthenticationCallback(const authentication_callback &callback) throw () {
        if (callback) {
                this->authCallback = callback;
        } else {
                this->authCallback = NULL;
        }
}

void HTTP::setRequestCallback(const request_callback &callback) throw () {
        if (callback) {
                this->reqCallback = callback;
        } else {
                this->reqCallback = http::request_callback;
        }
}

void HTTP::setErrorCallback(const error_callback &callback) throw () {
        if (callback) {
                this->errorCallback = callback;
        } else {
                this->errorCallback = http::error_callback;
        }
}

IPAddressPort::vector HTTP::getListeners(void) const throw () {
        return listeners;
}

std::string HTTP::getListenersString(void) const throw () {
        std::string buf;

        if (!listeners.empty()) {
                for (auto l = listeners.begin(); l != listeners.end(); ++l) {
                        buf += l->str() + ",";
                }

                return buf.erase(buf.size() - 1);
        }

        return buf;
}

IPAddressPort::vector HTTP::getSecureListeners(void) const throw () {
        return secureListeners;
}

std::string HTTP::getSecureListenersString(void) const throw () {
        std::string buf;

        if (!secureListeners.empty()) {
                for (auto l = secureListeners.begin(); l != secureListeners.end(); ++l) {
                        buf += l->str() + ",";
                }

                return buf.erase(buf.size() - 1);
        }

        return buf;
}

size_t HTTP::getRequestBodyLimit(void) const throw () {
        return requestBodyLimit;
}

size_t HTTP::getConnectionLimit(void) const throw () {
        return connectionLimit;
}

const std::string& HTTP::getRoot(void) const throw () {
        return root;
}

const std::vector<std::string>& HTTP::getDirectoryIndexes(void) const throw () {
        return indexes;
}

const std::string& HTTP::getBanner(void) const throw () {
        return banner;
}

const std::vector<std::string>& HTTP::getCustomHeaders(void) const throw () {
        return headers;
}

std::set<HTTPRequest::Method> HTTP::getAllow(void) const throw () {
        return allow;
}

HTTPRequest::AcceptType HTTP::getFallbackAcceptType(void) const throw () {
        return fallbackAcceptType;
}

size_t HTTP::getCompressionThreshold(void) const throw () {
        return compressionThreshold;
}

int HTTP::getCompressionLevel(void) const throw () {
        return compressionLevel;
}

void HTTP::clearAllow(HTTPRequest::Method method) throw () {
        allow.erase(method);
}

bool HTTP::isRunning(void) const throw () {
        return running;
}

void HTTP::clear(void) {
        if (isRunning()) {
                stop();
        }

        pool.reset();
        mux.clear();
        servers.clear();
        listeners = ListenersDefault();
        secureListeners = SecureListenersDefault();
        protocols = SecureTCPSocket::DefaultProtocols;
        ciphers.clear();
        preferServerCiphers = false;
        certificateFile.clear();
        privateKeyFile.clear();

        /* Scrub buffer before release back to stack. */
        for (size_t i = 0; i < password.size(); i++) {
                password.at(i) = 0;
        }

        passwordFile.clear();
        password.clear();
        certificateChainFile.clear();
        caCertificateFile.clear();
        running = false;
        std::queue<HTTPConnection::shared>().swap(pending);
        requestBodyLimit = RequestBodyLimitDefault;
        connectionLimit = ConnectionLimitDefault;
        root = RootDefault();
        indexes = DirectoryIndexesDefault();
        banner = BannerDefault();
        headers.clear();
        allow = AllowDefault();
        compressionThreshold = CompressionThresholdDefault;
        compressionLevel = CompressionLevelDefault;
        aclCallback = NULL;
        authCallback = NULL;
        reqCallback = http::request_callback;
        errorCallback = http::error_callback;
}

void HTTP::print(void) {
        INFO("[HTTP]");
        INFO("  listeners: %s", getListenersString().c_str());
        INFO("  secureListeners: %s", getSecureListenersString().c_str());

        if (!protocols.empty()) {
                std::string buf;
                for (auto p = protocols.begin(); p != protocols.end(); ++p) {
                        switch (*p) {
                                case SecureTCPSocket::SSLv3:
                                        buf.append("SSLv3 ");
                                        break;

                                case SecureTCPSocket::TLSv1_0:
                                        buf.append("TLSv1 ");
                                        break;

                                case SecureTCPSocket::TLSv1_1:
                                        buf.append("TLSv1.1 ");
                                        break;

                                case SecureTCPSocket::TLSv1_2:
                                        buf.append("TLSv1.2 ");
                                        break;
                        }
                }

                INFO("  protocols: { %s}", buf.c_str());
        } else {
                INFO("  protocols: all");
        }

        INFO("  ciphers: %s", ciphers.c_str());
        INFO("  preferServerCiphers: %s", (preferServerCiphers) ? "true" : "false");
        INFO("  certificateFile: %s", certificateFile.c_str());
        INFO("  privateKeyFile: %s", privateKeyFile.c_str());
        INFO("  passwordFile: %s", passwordFile.c_str());
        INFO("  password: %s", password.c_str());
        INFO("  certificateChainFile: %s", certificateChainFile.c_str());
        INFO("  caCertificateFile: %s", caCertificateFile.c_str());
        INFO("  requestBodyLimit: %zd", requestBodyLimit);
        INFO("  connectionLimit: %zd (pending: %zd)", connectionLimit, pending.size());
        INFO("  root: %s", root.c_str());
        INFO("  indexes: %s", String::join(indexes, ",").c_str());
        INFO("  banner: %s", banner.c_str());
        INFO("  headers: %zd", headers.size());
//      INFO("  allow: %s", allow);
        INFO("  compressionThreshold: %zd", compressionThreshold);
        INFO("  compressionLevel: %zd", compressionLevel);
        INFO("  aclCallback: %s", (aclCallback) ? "set" : "null");
        INFO("  authCallback: %s", (authCallback) ? "set" : "null");
        INFO("  reqCallback: %s", (reqCallback) ? "set" : "null");
        INFO("  errorCallback: %s", (errorCallback) ? "set" : "null");
}

const HTTP::shared HTTP::shared_ptr() throw (ServerException) {
        return std::make_shared<HTTP>();
}

const HTTP::shared HTTP::shared_ptr(Log::shared log) throw (ServerException) {
        return std::make_shared<HTTP>(log);
}

void HTTP::startup(void) throw (ServerException) {
        mux.insertRead(trigger.getSocket());

        for (auto l = listeners.begin(); l != listeners.end(); ++l) {
                try {
                        TCPSocket::shared s = TCPSocket::shared_ptr(l->getFamily());
                        s->setSoReuseAddr();
                        s->setBlock(true);

                        if (s->getFamily() == AF_INET6) {
                                s->setIpv6V6Only(true);
                        }

                        s->bind(*l);

                        mux.insertRead(s);
                        servers.insert(s->getFD());

                        s->listen();
                } catch (SocketException &e) {
                        RETHROW(ServerException, e);
                }
        }

        for (auto l = secureListeners.begin(); l != secureListeners.end(); ++l) {
                try {
                        SecureTCPSocket::shared s = SecureTCPSocket::shared_ptr(l->getFamily());
                        s->setSoReuseAddr();
                        s->setBlock(true);

                        s->setProtocols(protocols);
                        s->setCiphers(ciphers);
                        s->setPreferServerCiphers(preferServerCiphers);
                        s->setCertificateFile(certificateFile);
                        s->setPrivateKeyFile(privateKeyFile);
                        s->setPasswordFile(passwordFile);
                        s->setPassword(password);
                        s->setCertificateChainFile(certificateChainFile);
                        s->setCACertificateFile(caCertificateFile);

                        if (s->getFamily() == AF_INET6) {
                                s->setIpv6V6Only(true);
                        }

                        s->bind(*l);

                        mux.insertRead(s);
                        servers.insert(s->getFD());

                        s->listen();
                } catch (SocketException &e) {
                        RETHROW(ServerException, e);
                }
        }
}

void HTTP::shutdown(void) throw () {
        mux.clear();
        servers.clear();
}

int HTTP::select(timeval *tv) throw (ServerException) {
        int c;

        if ((c = mux.rselect(tv)) > 0) {
                const SocketMux::shared_flist &sockets = mux.getReadList();
                for (auto s = sockets.cbegin(); s != sockets.cend(); ++s) {
                        if (*s == trigger.getSocket()) {
                                trigger.clear();

                                return 0;
                        }

                        HTTPConnection::shared connection = accept(*s);
                                
                        if (pending.size() < connectionLimit) {
                                pending.push(std::move(connection));
                        } else {
                                try {
                                        errorCallback(connection, Code::ServiceUnavailable, std::string()); /* Service Unavailable */
                                } catch (ServerException &e) {
                                        WARNING("HTTP::errorCallback() failed with code %d: %s\n", e.getCode(), e.what());
                                }

                                connection->clear();
                                c--; /* Reduce new socket count. */
                        }
                }
        }

        return c;
}

HTTPConnection::shared HTTP::accept(const Socket::shared &socket) throw (ServerException) {
   try {
                TCPSocket::shared server(std::dynamic_pointer_cast<TCPSocket>(socket));
                TCPSocket::shared client = server->accept();

                if (client->isSecure()) {
                        TRACE(PACKAGE "-detail", "Incoming HTTPS connection from %s", client->getRemoteIPAddressPort().c_str());
                } else {
                        TRACE(PACKAGE "-detail", "Incoming HTTP connection from %s", client->getRemoteIPAddressPort().c_str());
                }

                return HTTPConnection::shared_ptr(*this, client);
   } catch (SocketException &e) {
      RETHROW(ServerException, e);
   }
}

void HTTP::process(HTTPConnection::shared &connection) throw () {
        try {
                if (aclCallback && !aclCallback(connection)) {
                        try {   
                                errorCallback(connection, Code::Forbidden, std::string()); /* Forbidden */
                        } catch (ServerException &e) {
                                WARNING("HTTP::errorCallback() failed with code %d: %s\n", e.getCode(), e.what());
                        }
                } else if (authCallback && !authCallback(connection)) {
                        try {
                                errorCallback(connection, Code::Unauthorized, std::string()); /* Unauthorized */
                        } catch (ServerException &e) {
                                WARNING("HTTP::errorCallback() failed with code %d: %s\n", e.getCode(), e.what());
                        }
                } else {
                        reqCallback(connection);
                }
        } catch (ServerException &e) {
                try {
                        if (e.getCode() > 0) {
                                errorCallback(connection, e.getCode(), e.getMessage());
                        } else {
                                errorCallback(connection, Code::InternalServerError, std::string()); /* Internal Server Error */
                        }
                } catch (ServerException &e) {
                        WARNING("HTTP::errorCallback() failed with code %d: %s\n", e.getCode(), e.what());
                }
        } catch (std::exception &e) {
                try {
                        errorCallback(connection, Code::InternalServerError, std::string()); /* Internal Server Error */
                } catch (ServerException &e) {
                        WARNING("HTTP::errorCallback() failed with code %d: %s\n", e.getCode(), e.what());
                }
        }
}

void HTTP::server(void) throw () {
        int c;
        timeval tv;
        Task::shared task;
        Task::callback function;

        while (running) {
                try {
                        tv = { DelaySecondsDefault, 0 };

                        while ((c = select(&tv)) > 0) {
                                while (!pending.empty()) {
                                        HTTPConnection::shared connection = pending.front();
                                        pending.pop();

                                        task = Task::shared_ptr();
                                        function = std::bind(&HTTP::worker, this, connection);
                                        task->setCallback(function);

                                        pool->schedule(task);
                                }
                        }
                } catch (ServerException &e) {
                        WARNING("ServerException: %s", e.what());
                }
        }

        TRACE(PACKAGE, "HTTP server stopping...");

        std::lock_guard<std::mutex> lg(mutex);

        shutdown();
        std::queue<HTTPConnection::shared>().swap(pending);

        pool->clear();
        pool->wait();

        TRACE(PACKAGE, "HTTP server stopped");
}

void HTTP::worker(HTTPConnection::shared &connection) throw () {
        process(connection);

        connection->clear();
}

/*
** vim: noet ts=3 sw=3
*/