unix_command_mgr.cc

Go to the documentation of this file.

// Copyright (C) 2015-2026 Internet Systems Consortium, Inc. ("ISC")
//
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
#include <config.h>
 
#include <asiolink/asio_wrapper.h>
#include <asiolink/interval_timer.h>
#include <asiolink/io_service.h>
#include <asiolink/unix_domain_socket.h>
#include <asiolink/unix_domain_socket_acceptor.h>
#include <asiolink/unix_domain_socket_endpoint.h>
#include <config/command_mgr.h>
#include <config/unix_command_config.h>
#include <config/unix_command_mgr.h>
#include <cc/data.h>
#include <cc/command_interpreter.h>
#include <cc/json_feed.h>
#include <dhcp/iface_mgr.h>
#include <config/config_log.h>
#include <config/timeouts.h>
#include <util/filesystem.h>
#include <util/watch_socket.h>
#include <boost/enable_shared_from_this.hpp>
#include <array>
#include <functional>
#include <unistd.h>
#include <sys/file.h>
 
using namespace isc;
using namespace isc::asiolink;
using namespace isc::config;
using namespace isc::data;
using namespace isc::dhcp;
namespace ph = std::placeholders;
 
namespace {

const size_t BUF_SIZE = 32768;
 
class ConnectionPool;
 
class Connection;

typedef boost::shared_ptr<Connection> ConnectionPtr;

class Connection : public boost::enable_shared_from_this<Connection> {
public:

    Connection(const IOServicePtr& io_service,
               const boost::shared_ptr<UnixDomainSocket>& socket,
               ConnectionPool& connection_pool,
               const long timeout,
               bool use_external)
        : io_service_(io_service), socket_(socket), timeout_timer_(io_service_),
          timeout_(timeout), buf_(), response_(), position_(0),
          connection_pool_(connection_pool), feed_(), watch_socket_(),
          use_external_(use_external), defer_shutdown_(false) {
 
        LOG_DEBUG(command_logger, DBG_COMMAND, COMMAND_SOCKET_CONNECTION_OPENED)
            .arg(socket_->getNative());
 
        // Callback value of 0 is used to indicate that callback function is
        // not installed.
        if (use_external_) {
            watch_socket_.reset(new util::WatchSocket());
            IfaceMgr::instance().addExternalSocket(watch_socket_->getSelectFd(), 0);
            IfaceMgr::instance().addExternalSocket(socket_->getNative(), 0);
        }
 
        // Initialize state model for receiving and preparsing commands.
        feed_.initModel();
 
        // Start timer for detecting timeouts.
        scheduleTimer();
    }

    ~Connection() {
        timeout_timer_.cancel();
    }

    void scheduleTimer() {
        timeout_timer_.setup(std::bind(&Connection::timeoutHandler, this),
                             timeout_, IntervalTimer::ONE_SHOT);
    }

    void stop() {
        if (defer_shutdown_) {
            io_service_->post(std::bind([](ConnectionPtr c) { c->stop(); }, shared_from_this()));
            return;
        }
 
        LOG_DEBUG(command_logger, DBG_COMMAND, COMMAND_SOCKET_CONNECTION_CLOSED)
            .arg(socket_->getNative());
 
        if (use_external_) {
            IfaceMgr::instance().deleteExternalSocket(watch_socket_->getSelectFd());
            IfaceMgr::instance().deleteExternalSocket(socket_->getNative());
 
            // Close watch socket and log errors if occur.
            std::string watch_error;
            if (!watch_socket_->closeSocket(watch_error)) {
                LOG_ERROR(command_logger, COMMAND_WATCH_SOCKET_CLOSE_ERROR)
                    .arg(watch_error);
            }
        }
 
        socket_->close();
        timeout_timer_.cancel();
    }

    void terminate();

    void doReceive() {
        socket_->asyncReceive(&buf_[0], sizeof(buf_),
                              std::bind(&Connection::receiveHandler,
                                        shared_from_this(), ph::_1, ph::_2));
    }

    void doSend() {
        // doSend is called only when there is something to send so
        // remaining is guaranteed to be greater than 0 and position_
        // to point inside the buffer.
        size_t remaining = response_.size() - position_;
        socket_->asyncSend(&response_[position_], remaining,
           std::bind(&Connection::sendHandler, shared_from_this(), ph::_1, ph::_2));
 
        if (use_external_) {
            // Asynchronous send has been scheduled and we need to indicate this
            // to break the synchronous select(). The handler should clear this
            // status when invoked.
            try {
                watch_socket_->markReady();
            } catch (const std::exception& ex) {
                LOG_ERROR(command_logger, COMMAND_WATCH_SOCKET_MARK_READY_ERROR)
                    .arg(ex.what());
            }
        }
    }

    //
    void receiveHandler(const boost::system::error_code& ec,
                        size_t bytes_transferred);

    void sendHandler(const boost::system::error_code& ec,
                     size_t bytes_transferred);

    void timeoutHandler();
 
private:

    IOServicePtr io_service_;

    boost::shared_ptr<UnixDomainSocket> socket_;

    IntervalTimer timeout_timer_;

    long timeout_;

    std::array<char, BUF_SIZE> buf_;

    std::string response_;

    size_t position_;

    ConnectionPool& connection_pool_;

    JSONFeed feed_;

    util::WatchSocketPtr watch_socket_;

    bool use_external_;

    bool defer_shutdown_;
};

class ConnectionPool {
public:

    void start(const ConnectionPtr& connection) {
        connection->doReceive();
        connections_.insert(connection);
    }

    void stop(const ConnectionPtr& connection) {
        try {
            connection->stop();
            connections_.erase(connection);
        } catch (const std::exception& ex) {
            LOG_ERROR(command_logger, COMMAND_SOCKET_CONNECTION_CLOSE_FAIL)
                .arg(ex.what());
        }
    }

    void stopAll() {
        for (auto const& conn : connections_) {
            conn->stop();
        }
        connections_.clear();
    }
 
private:

    std::set<ConnectionPtr> connections_;
 
};
 
void
Connection::terminate() {
    try {
        socket_->shutdown();
 
    } catch (const std::exception& ex) {
        LOG_ERROR(command_logger, COMMAND_SOCKET_CONNECTION_SHUTDOWN_FAIL)
            .arg(ex.what());
    }
}
 
void
Connection::receiveHandler(const boost::system::error_code& ec,
                           size_t bytes_transferred) {
    if (ec) {
        if (ec.value() == boost::asio::error::eof) {
            std::stringstream os;
            if (feed_.getProcessedText().empty()) {
               os << "no input data to discard";
            } else {
               os << "discarding partial command of "
                  << feed_.getProcessedText().size() << " bytes";
            }
 
            // Foreign host has closed the connection. We should remove it from the
            // connection pool.
            LOG_INFO(command_logger, COMMAND_SOCKET_CLOSED_BY_FOREIGN_HOST)
                .arg(socket_->getNative()).arg(os.str());
        } else if (ec.value() != boost::asio::error::operation_aborted) {
            LOG_ERROR(command_logger, COMMAND_SOCKET_READ_FAIL)
                .arg(ec.value()).arg(socket_->getNative());
        }
 
        connection_pool_.stop(shared_from_this());
        return;
 
    } else if (bytes_transferred == 0) {
        // Nothing received. Close the connection.
        connection_pool_.stop(shared_from_this());
        return;
    }
 
    LOG_DEBUG(command_logger, DBG_COMMAND, COMMAND_SOCKET_READ)
        .arg(bytes_transferred).arg(socket_->getNative());
 
    // Reschedule the timer because the transaction is ongoing.
    scheduleTimer();
 
    ConstElementPtr cmd;
    ConstElementPtr rsp;
 
    try {
        // Received some data over the socket. Append them to the JSON feed
        // to see if we have reached the end of command.
        feed_.postBuffer(&buf_[0], bytes_transferred);
        feed_.poll();
        // If we haven't yet received the full command, continue receiving.
        if (feed_.needData()) {
            doReceive();
            return;
        }
 
        // Received entire command. Parse the command into JSON.
        if (feed_.feedOk()) {
            cmd = feed_.toElement();
 
            defer_shutdown_ = true;
 
            std::unique_ptr<Connection, void (*)(Connection*)> p(this, [](Connection* c) {
                c->defer_shutdown_ = false;
            });
 
            // Cancel the timer to make sure that long lasting command
            // processing doesn't cause the timeout.
            timeout_timer_.cancel();
 
            // If successful, then process it as a command.
            rsp = CommandMgr::instance().processCommand(cmd);
 
        } else {
            // Failed to parse command as JSON or process the received command.
            // This exception will be caught below and the error response will
            // be sent.
            isc_throw(BadValue, feed_.getErrorMessage());
        }
 
    } catch (const Exception& ex) {
        LOG_WARN(command_logger, COMMAND_PROCESS_ERROR1).arg(ex.what());
        rsp = createAnswer(CONTROL_RESULT_ERROR, std::string(ex.what()));
    }
 
    // No response generated. Connection will be closed.
    if (!rsp || rsp->str().empty()) {
        LOG_WARN(command_logger, COMMAND_RESPONSE_ERROR)
            .arg(cmd ? cmd->str() : "unknown");
        rsp = createAnswer(CONTROL_RESULT_ERROR,
                           "internal server error: no response generated");
 
    } else {
 
        // Reschedule the timer as it may be either canceled or need to be
        // updated to not timeout before we manage to the send the reply.
        scheduleTimer();
 
        // Let's convert JSON response to text. Note that at this stage
        // the rsp pointer is always set and its rendering not empty.
        response_ = rsp->str();
 
        doSend();
        return;
    }
 
    // Close the connection if we have sent the entire response.
    connection_pool_.stop(shared_from_this());
}
 
void
Connection::sendHandler(const boost::system::error_code& ec,
                        size_t bytes_transferred) {
    if (use_external_) {
        // Clear the watch socket so as the future send operation can mark it
        // again to interrupt the synchronous select() call.
        try {
            watch_socket_->clearReady();
        } catch (const std::exception& ex) {
            LOG_ERROR(command_logger, COMMAND_WATCH_SOCKET_CLEAR_ERROR)
                .arg(ex.what());
        }
    }
 
    if (ec) {
        // If an error occurred, log this error and stop the connection.
        if (ec.value() != boost::asio::error::operation_aborted) {
            LOG_ERROR(command_logger, COMMAND_SOCKET_WRITE_FAIL)
                .arg(socket_->getNative()).arg(ec.message());
        }
 
    } else {
 
        // Reschedule the timer because the transaction is ongoing.
        scheduleTimer();
 
        // No error. We are in a process of sending a response. Need to
        // advance the to send position in the response.
        if (bytes_transferred >= response_.size() - position_) {
            position_ = response_.size();
        } else {
            position_ += bytes_transferred;
        }
 
        LOG_DEBUG(command_logger, DBG_COMMAND, COMMAND_SOCKET_WRITE)
            .arg(bytes_transferred)
            .arg(response_.size() - position_)
            .arg(socket_->getNative());
 
        // Check if there is any data left to be sent and sent it.
        if (position_ < response_.size()) {
            doSend();
            return;
        }
 
        // Gracefully shutdown the connection and close the socket if
        // we have sent the whole response.
        terminate();
    }
 
    // All data sent or an error has occurred. Close the connection.
    connection_pool_.stop(shared_from_this());
}
 
void
Connection::timeoutHandler() {
    LOG_INFO(command_logger, COMMAND_SOCKET_CONNECTION_TIMEOUT)
        .arg(socket_->getNative());
 
    try {
        socket_->cancel();
 
    } catch (const std::exception& ex) {
        LOG_ERROR(command_logger, COMMAND_SOCKET_CONNECTION_CANCEL_FAIL)
            .arg(socket_->getNative())
            .arg(ex.what());
    }
 
    std::stringstream os;
    os << "Connection over control channel timed out";
    if (!feed_.getProcessedText().empty()) {
        os << ", discarded partial command of "
           << feed_.getProcessedText().size() << " bytes";
    }
 
    ConstElementPtr rsp = createAnswer(CONTROL_RESULT_ERROR, os.str());
    response_ = rsp->str();
    doSend();
}
 
}
 
namespace isc {
namespace config {

class UnixCommandMgrImpl {
public:

    UnixCommandMgrImpl()
        : timeout_(TIMEOUT_DHCP_SERVER_RECEIVE_COMMAND), use_external_(true) {
    }

    void openCommandSockets(const isc::data::ConstElementPtr config);

    void openCommandSocket(const isc::data::ConstElementPtr config);

    void closeCommandSocket(UnixSocketInfoPtr info);

    void closeCommandSockets(bool remove = true);

    void doAccept(UnixSocketInfoPtr info);

    int getControlSocketFD(UnixSocketInfoPtr info);

    IOServicePtr io_service_;

    ConnectionPool connection_pool_;

    std::map<std::string, UnixSocketInfoPtr> sockets_;

    long timeout_;

    bool use_external_;
};
 
void
UnixCommandMgrImpl::openCommandSockets(const isc::data::ConstElementPtr config) {
    if (!config) {
        isc_throw(BadSocketInfo, "Missing config parameters, can't create socket.");
    }
 
    if (config->getType() != Element::list) {
        isc_throw(DhcpConfigError, "expected list type ("
                  << config->getPosition() << ")");
    }
 
    for (auto const& socket : config->listValue()) {
        openCommandSocket(socket);
    }
 
    auto copy = sockets_;
    for (auto const& data : copy) {
        if (data.second->usable_) {
            // If the connection can be used (just created) or reused, keep it
            // in the list and clear the flag. It will be marked again on next
            // configuration event if needed.
            data.second->usable_ = false;
        } else {
            // If the connection can not be reused, stop it and remove it from the list.
            closeCommandSocket(data.second);
        }
    }
}
 
void
UnixCommandMgrImpl::openCommandSocket(const isc::data::ConstElementPtr config) {
    if (!config) {
        isc_throw(BadSocketInfo, "Missing config parameters, can't create socket.");
    }
 
    UnixCommandConfigPtr cmd_config(new UnixCommandConfig(config));
 
    // Search for the specific connection and reuse the existing one if found.
    auto it = sockets_.find(cmd_config->getSocketName());
    if (it != sockets_.end()) {
        // If the connection can be reused, mark it as usable.
        it->second->usable_ = true;
        return;
    }
 
    // Connection not found so it needs to be created.
    // First let's open lock file.
    std::string lock_name = cmd_config->getLockName();
    int lock_fd = open(lock_name.c_str(), O_RDONLY | O_CREAT, 0600);
    if (lock_fd == -1) {
        std::string errmsg = strerror(errno);
        isc_throw(SocketError, "cannot create socket lockfile, "
                  << lock_name  << ", : " << errmsg);
    }
 
    // Try to acquire lock. If we can't somebody else is actively
    // using it.
    int ret = flock(lock_fd, LOCK_EX | LOCK_NB);
    if (ret != 0) {
        std::string errmsg = strerror(errno);
        close(lock_fd);
        isc_throw(SocketError, "cannot lock socket lockfile, "
                  << lock_name  << ", : " << errmsg);
    }
 
    // We have the lock, so let's remove the pre-existing socket
    // file if it exists.
    static_cast<void>(::remove(cmd_config->getSocketName().c_str()));
 
    LOG_INFO(command_logger, COMMAND_ACCEPTOR_START)
        .arg(cmd_config->getSocketName());
 
    UnixSocketInfoPtr socket_info(new UnixSocketInfo());
    socket_info->config_ = cmd_config;
    socket_info->lock_fd_ = lock_fd;
 
    try {
        // Start asynchronous acceptor service.
        socket_info->acceptor_.reset(new UnixDomainSocketAcceptor(io_service_));
        UnixDomainSocketEndpoint endpoint(cmd_config->getSocketName());
        socket_info->acceptor_->open(endpoint);
        // Kea umask (0027) is too restrictive: the bind() system call
        // creates the socket file with 0777 (or on some systems fchmod
        // provided value) masked by the current umask. The 'connect'
        // system call requires write access to socket file.
        // So relaxing the umask to allow group write access.
        {
            isc::util::file::RelaxUmask ru;
            socket_info->acceptor_->bind(endpoint);
        }
        socket_info->acceptor_->listen();
        if (use_external_) {
            // Install this socket in Interface Manager.
            IfaceMgr::instance().addExternalSocket(socket_info->acceptor_->getNative(), 0);
        }
 
        doAccept(socket_info);
 
    } catch (const std::exception& ex) {
        close(lock_fd);
        static_cast<void>(::remove(cmd_config->getLockName().c_str()));
        isc_throw(SocketError, ex.what());
    }
 
    sockets_[cmd_config->getSocketName()] = socket_info;
}
 
void
UnixCommandMgrImpl::closeCommandSocket(UnixSocketInfoPtr info) {
    if (info) {
        // Close acceptor if the acceptor is open.
        if (info->acceptor_ && info->acceptor_->isOpen()) {
            if (use_external_) {
                IfaceMgr::instance().deleteExternalSocket(info->acceptor_->getNative());
            }
            info->acceptor_->close();
            static_cast<void>(::remove(info->config_->getSocketName().c_str()));
            static_cast<void>(::remove(info->config_->getLockName().c_str()));
        }
 
        // Stop all connections which can be closed. The only connection that won't
        // be closed is the one over which we have received a request to reconfigure
        // the server. This connection will be held until the UnixCommandMgr
        // responds to such request.
        connection_pool_.stopAll();
        if (info->lock_fd_ != -1) {
            close(info->lock_fd_);
            info->lock_fd_ = -1;
        }
        auto it = sockets_.find(info->config_->getSocketName());
        if (it != sockets_.end()) {
            sockets_.erase(it);
        }
        try {
            io_service_->pollOne();
        } catch (...) {
        }
    } else {
        closeCommandSockets(false);
    }
}
 
void
UnixCommandMgrImpl::closeCommandSockets(bool remove) {
    auto copy = sockets_;
    for (auto const& data : copy) {
        closeCommandSocket(data.second);
    }
    if (remove) {
        sockets_.clear();
    }
}
 
void
UnixCommandMgrImpl::doAccept(UnixSocketInfoPtr info) {
    // Create a socket into which the acceptor will accept new connection.
    info->socket_.reset(new UnixDomainSocket(io_service_));
    info->acceptor_->asyncAccept(*info->socket_,
                                 [this, info](const boost::system::error_code& ec) {
        if (!ec) {
            // New connection is arriving. Start asynchronous transmission.
            ConnectionPtr connection(new Connection(io_service_, info->socket_,
                                                    connection_pool_,
                                                    timeout_,
                                                    use_external_));
            connection_pool_.start(connection);
 
        } else if (ec.value() != boost::asio::error::operation_aborted) {
            LOG_ERROR(command_logger, COMMAND_SOCKET_ACCEPT_FAIL)
                .arg(info->acceptor_->getNative()).arg(ec.message());
        }
 
        // Unless we're stopping the service, start accepting connections again.
        if (ec.value() != boost::asio::error::operation_aborted) {
            doAccept(info);
        }
    });
}
 
int
UnixCommandMgrImpl::getControlSocketFD(UnixSocketInfoPtr info) {
    // Return the most recent listener or null.
    if (info) {
        auto const& it = sockets_.find(info->config_->getSocketName());
        if (it != sockets_.end()) {
            return (it->second->acceptor_->getNative());
        }
    } else if (sockets_.size()) {
        return (sockets_.begin()->second->acceptor_->getNative());
    }
    return (-1);
}
 
UnixCommandMgr::UnixCommandMgr() : impl_(new UnixCommandMgrImpl()) {
}
 
void
UnixCommandMgr::openCommandSocket(const isc::data::ConstElementPtr config) {
    impl_->openCommandSocket(config);
}
 
void
UnixCommandMgr::openCommandSockets(const isc::data::ConstElementPtr config) {
    impl_->openCommandSockets(config);
}
 
void
UnixCommandMgr::closeCommandSocket(UnixSocketInfoPtr info) {
    impl_->closeCommandSocket(info);
}
 
void
UnixCommandMgr::closeCommandSockets() {
    impl_->closeCommandSockets();
}
 
int
UnixCommandMgr::getControlSocketFD(UnixSocketInfoPtr info) {
    return (impl_->getControlSocketFD(info));
}
 
UnixCommandMgr&
UnixCommandMgr::instance() {
    static UnixCommandMgr cmd_mgr;
    return (cmd_mgr);
}
 
void
UnixCommandMgr::setIOService(const IOServicePtr& io_service) {
    impl_->io_service_ = io_service;
}
 
void
UnixCommandMgr::setConnectionTimeout(const long timeout) {
    impl_->timeout_ = timeout;
}
 
void
UnixCommandMgr::addExternalSockets(bool use_external) {
    impl_->use_external_ = use_external;
}
 
} // end of isc::config
} // end of isc