RemoteControl.cpp

Go to the documentation of this file.

/*
    Copyright (C) 2014-2016 Leosac

    This file is part of Leosac.

    Leosac is free software: you can redistribute it and/or modify
    it under the terms of the GNU Affero General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    Leosac is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
    GNU Affero General Public License for more details.

    You should have received a copy of the GNU Affero General Public License
    along with this program. If not, see <http://www.gnu.org/licenses/>.
*/

#include "RemoteControl.hpp"
#include "core/CoreUtils.hpp"
#include "core/config/RemoteConfigCollector.hpp"
#include "core/tasks/FetchRemoteConfig.hpp"
#include "core/tasks/RemoteControlAsyncResponse.hpp"
#include "core/tasks/SyncConfig.hpp"
#include "kernel.hpp"
#include "tools/XmlPropertyTree.hpp"
#include "tools/log.hpp"
#include <boost/archive/binary_iarchive.hpp>
#include <boost/archive/text_iarchive.hpp>
#include <boost/archive/text_oarchive.hpp>
#include <boost/property_tree/ptree_serialization.hpp>
#include <boost/regex.hpp>
#include <cassert>
#include <zmqpp/curve.hpp>

using namespace Leosac;

RemoteControl::RemoteControl(zmqpp::context &ctx, Kernel &kernel,
                             const boost::property_tree::ptree &cfg)
    : kernel_(kernel)
    , socket_(ctx, zmqpp::socket_type::router)
    , auth_(ctx)
    , context_(ctx)
    , security_(cfg)
{
    auth_.configure_curve("CURVE_ALLOW_ANY");
    process_config(cfg);
}

void RemoteControl::process_config(const boost::property_tree::ptree &cfg)
{
    int port = cfg.get<int>("port");

    secret_key_ = cfg.get<std::string>("secret_key");
    public_key_ = cfg.get<std::string>("public_key");

    INFO("Enabling Remote Control:"
         << "\n\t "
         << "Port: " << port << "\n\t "
         << "Public Key: " << public_key_ << "\n\t "
         << "Private Key: " << secret_key_);

    command_handlers_["MODULE_CONFIG"] =
        std::bind(&RemoteControl::handle_module_config, this, std::placeholders::_1,
                  std::placeholders::_2);

    command_handlers_["MODULE_LIST"] =
        std::bind(&RemoteControl::handle_module_list, this, std::placeholders::_1,
                  std::placeholders::_2);

    command_handlers_["SYNC_FROM"] =
        std::bind(&RemoteControl::handle_sync_from, this, std::placeholders::_1,
                  std::placeholders::_2);

    command_handlers_["SAVE"] =
        std::bind(&RemoteControl::handle_save, this, std::placeholders::_1,
                  std::placeholders::_2);

    command_handlers_["GENERAL_CONFIG"] =
        std::bind(&RemoteControl::handle_general_config, this, std::placeholders::_1,
                  std::placeholders::_2);

    command_handlers_["CONFIG_VERSION"] =
        std::bind(&RemoteControl::handle_config_version, this, std::placeholders::_1,
                  std::placeholders::_2);

    socket_.set(zmqpp::socket_option::curve_server, true);
    socket_.set(zmqpp::socket_option::curve_secret_key, secret_key_);
    socket_.set(zmqpp::socket_option::curve_public_key, public_key_);
    socket_.bind("tcp://*:" + std::to_string(port));
}

void RemoteControl::handle_msg()
{
    zmqpp::message msg;
    zmqpp::message rep;
    std::string frame1;
    socket_.receive(msg);

    assert(msg.parts() > 1);
    msg >> current_client_idt_;
    rep << current_client_idt_;

    msg.pop_front();         // otherwise getting the "User-Id" property
    msg.reset_read_cursor(); // wont work.

    msg >> frame1;
    DEBUG("Remote Control command: " << frame1 << " with " << msg.parts()
                                     << " parts");

    std::string user_pubkey;
    bool ret = msg.get_property("User-Id", user_pubkey);
    assert(ret);

    if (command_handlers_.find(frame1) != command_handlers_.end())
    {

        if (security_.allow_request(user_pubkey, frame1))
        {
            auto h = command_handlers_[frame1];

            ret = h(&msg, &rep);
            if (!ret)
            {
                // if the handler fails, that means the source message was incorrect.
                // therefore, the response shall be empty (except for the zmq id)
                assert(rep.parts() == 1);
                rep << "KO"
                    << "Malformed message: " << frame1;
                WARN("Received malformed message on Remote Control Interface. "
                     "Message type was "
                     << frame1);
            }
        }
        else
        {
            WARN("Request denied. Insuficient permission for user "
                 << user_pubkey << ". Command was: " << frame1);
            rep << "KO"
                << "Insuficient permission";
        }
    }
    else
    {
        rep << "KO"
            << "UNKOWN MESSAGE";
        WARN("Unknown message on Remote Control interface");
    }

    socket_.send(rep);
}

void RemoteControl::module_list(zmqpp::message *message_out)
{
    assert(message_out);
    for (const std::string &s : kernel_.module_manager().modules_names())
    {
        *message_out << s;
    }
}

void RemoteControl::module_config(const std::string &module,
                                  ConfigManager::ConfigFormat cfg_format,
                                  zmqpp::message *message_out)
{
    assert(message_out);

    // we need to make sure the module's name exist.
    std::vector<std::string> modules_names =
        kernel_.module_manager().modules_names();
    if (std::find(modules_names.begin(), modules_names.end(), module) !=
        modules_names.end())
    {
        zmqpp::socket sock(context_, zmqpp::socket_type::req);
        sock.connect("inproc://module-" + module);

        bool ret = sock.send(zmqpp::message() << "DUMP_CONFIG" << cfg_format);
        ASSERT_LOG(ret, "Failed to send");

        zmqpp::message rep;

        sock.receive(rep);
        *message_out << "OK";
        *message_out << module;

        // extract data from received configuration
        // fixme this is poor code and involve lots of copying.
        while (rep.remaining())
        {
            std::string tmp;
            rep >> tmp;
            *message_out << tmp;
        }
    }
    else
    {
        // if module with this name is not found
        ERROR("RemoteControl: Cannot retrieve local module configuration for {"
              << module << "}"
              << "The module appears to not be loaded.");
        *message_out << "KO"
                     << "Module not loaded, so config not available";
    }
}

void RemoteControl::general_config(ConfigManager::ConfigFormat cfg_format,
                                   zmqpp::message *msg_out)
{
    assert(msg_out);

    auto cfg = kernel_.config_manager().get_exportable_general_config();

    if (cfg_format == ConfigManager::ConfigFormat::BOOST_ARCHIVE)
    {
        std::ostringstream oss;
        boost::archive::text_oarchive archive(oss);
        boost::property_tree::save(archive, cfg, 1);
        msg_out->add("OK");
        msg_out->add(oss.str());
    }
    else
    {
        msg_out->add("OK");
        msg_out->add(Tools::propertyTreeToXml(cfg));
    }
}

bool RemoteControl::handle_module_config(zmqpp::message *msg_in,
                                         zmqpp::message *msg_out)
{
    assert(msg_in);
    assert(msg_out);

    if (msg_in->remaining() >= 2)
    {
        std::string module_name;
        ConfigManager::ConfigFormat format;
        *msg_in >> module_name >> format;
        module_config(module_name, format, msg_out);
        return true;
    }
    return false;
}

bool RemoteControl::handle_module_list(zmqpp::message *msg_in,
                                       zmqpp::message *msg_out)
{
    assert(msg_in);
    assert(msg_out);

    if (msg_in->remaining() == 0)
    {
        module_list(msg_out);
        return true;
    }
    return false;
}

bool RemoteControl::handle_sync_from(zmqpp::message *msg_in, zmqpp::message *msg_out)
{
    assert(msg_in);
    assert(msg_out);

    uint8_t autocommit;

    if (msg_in->remaining() == 4)
    {
        std::string endpoint;
        std::string remote_server_pubkey;
        uint8_t sync_general_config;

        *msg_in >> endpoint;
        *msg_in >> autocommit;
        *msg_in >> remote_server_pubkey;
        *msg_in >> sync_general_config;


        auto fetch_task = std::make_shared<Tasks::FetchRemoteConfig>(
            endpoint, remote_server_pubkey);

        auto sync_task = std::make_shared<Tasks::SyncConfig>(
            kernel_, fetch_task, sync_general_config, autocommit);
        auto *sched = &kernel_.core_utils()->scheduler();

        auto success_response_task =
            std::make_shared<Tasks::RemoteControlAsyncResponse>(
                current_client_idt_,
                zmqpp::message() << "Success" << sync_task->get_guid(), socket_);

        auto failure_response_task =
            std::make_shared<Tasks::RemoteControlAsyncResponse>(
                current_client_idt_,
                zmqpp::message() << "Failed" << sync_task->get_guid(), socket_);

        auto abort_response_task =
            std::make_shared<Tasks::RemoteControlAsyncResponse>(
                current_client_idt_,
                zmqpp::message() << "Aborted" << sync_task->get_guid(), socket_);

        sync_task->set_on_success([=]() {
            success_response_task->run();
            ASSERT_LOG(success_response_task->succeed(), "TASK FAILED");
        });
        sync_task->set_on_failure([=]() {
            failure_response_task->run();
            ASSERT_LOG(failure_response_task->succeed(), "TASK FAILED");
        });
        fetch_task->set_on_failure([=]() {
            // not run on main thread, need to be queued.
            sched->enqueue(abort_response_task, TargetThread::MAIN);
        });

        fetch_task->set_on_success([=]() {
            DEBUG("FETCH TASK COMPLETE. WILL QUEUE SYNC_CONFIG");
            sched->enqueue(sync_task, TargetThread::MAIN);
        });

        kernel_.core_utils()->scheduler().enqueue(fetch_task, TargetThread::POOL);

        *msg_out << "DELAYED" << sync_task->get_guid();
        return true;
    }
    return false;
}

bool RemoteControl::handle_save(zmqpp::message *msg_in, zmqpp::message *msg_out)
{
    assert(msg_in);
    assert(msg_out);

    if (msg_in->remaining() == 0)
    {
        if (kernel_.save_config())
            *msg_out << "OK";
        else
            *msg_out << "KO"
                     << "Saving config failed for some unkown reason.";

        return true;
    }
    return false;
}

bool RemoteControl::handle_general_config(zmqpp::message *msg_in,
                                          zmqpp::message *msg_out)
{
    assert(msg_in);
    assert(msg_out);

    if (msg_in->remaining() == 1)
    {
        ConfigManager::ConfigFormat format;
        *msg_in >> format;
        general_config(format, msg_out);
        return true;
    }
    return false;
}

bool RemoteControl::handle_config_version(zmqpp::message *msg_in,
                                          zmqpp::message *msg_out)
{
    assert(msg_in);
    assert(msg_out);

    if (msg_in->remaining() == 0)
    {
        *msg_out << static_cast<uint64_t>(kernel_.config_manager().config_version());
        return true;
    }
    return false;
}

void RemoteControl::update()
{
}