Fulcrum/RPC.cpp

#include "RPC.h"
#include <QtCore>

#include <type_traits>

namespace RPC {

    const QString jsonRpcVersion("2.0");

    /// As per JSON RPC spec, id must be integer (no fractional part), string, or null.
    /// Will throw if that's not the case.
    auto Message::Id::operator=(const QVariant &v) -> Id &
    {
        bool ok;
        if (v.isNull())
            *this = nullptr;
        else if (QMetaType::Type(v.type()) == QMetaType::QString)
            *this = v.toString();
        else if (qint64 id_ll = v.toLongLong(&ok); ok && v.toString() == QString::number(id_ll)) // this checks that fractional part not present
            *this = id_ll;
        else
            // if we get here, id is not a valid type as per JSON RPC 2.0
            throw InvalidError("id must be a string, a non-fractonal number, or null");
        return *this;
    }

    bool Message::Id::operator<(const Id & other) const
    {
        if (index() != other.index()) {
            return index() < other.index();
        } else {
            bool ret = false;

            std::visit(Overloaded {
                [&other, &ret](const auto & arg) {
                    // qint64 or QString
                    ret = arg < std::get< typename std::decay<decltype(arg)>::type >(other);
                },
                [](const std::nullptr_t &) { /* always same, leave ret at false */ },
            }, *this);
            return ret;
        }
    }


    // will return a QVariant whose type is either: qint64, QString, or isNull()
    Message::Id::operator QVariant() const
    {
        QVariant ret;
        std::visit(Overloaded {
            [&ret](const auto & arg) { ret = arg; }, // qint64 or QString
            [](const std::nullptr_t &) { /* noop already null */ },
        }, *this);
        return ret;
    }

    /* static */
    Message Message::fromString(const QString &s)
    {
        return fromJsonData(Util::Json::parseString(s, true).toMap()); // may throw
    }

    /* static */
    Message Message::fromJsonData(const QVariantMap & map, Id * id_out)
    {
        Message ret;
        ret.data = map;

        if (id_out)
            id_out->clear();

        // Grab the id first in case later processing fails.
        if (auto var = map.value("id"); !var.isNull()) {
            // note as per JSON-RPC 2.0 spec, we squash floats down to ints, discarding the fractional part
            // we will throw if the id is not a string, integer, or null
            ret.id = var;
            // inform caller of parsed id asap in case we later throw
            if (id_out)
                *id_out = ret.id;
        }


        if (ret.jsonRpcVersion() != RPC::jsonRpcVersion)
            throw InvalidError("Expected jsonrpc version 2.0");

        if (auto var = map.value("method");
                map.contains("method") && (QMetaType::Type(var.type()) != QMetaType::QString
                                           || (ret.method = var.toString()).isEmpty()
                                           || ret.method.startsWith("rpc.")))
            throw InvalidError("Invalid method");

        // todo: see if the below validation needs optimization


        // validate error as per JSON RPC 2.0
        if (ret.isError()) {
            auto errmap = ret.data.value("error").toMap();
            if (!errmap.contains("code") || !errmap.contains("message"))
                throw InvalidError("Expected error object to contain code and message");
            int n_req = errmap.contains("data") ? 3 : 2;
            if (errmap.count() != n_req)
                throw InvalidError("Unexpected keys in error object");
            bool ok;
            if (int code = errmap.value("code").toInt(&ok); !ok || errmap.value("code").toString() != QString::number(code))
                throw InvalidError("Expected error code to be an integer");
            static const KeySet required = { "id", "error", "jsonrpc" };
            if (KeySet::fromList(ret.data.keys()) != required)
                throw InvalidError("Error response not valid");
        }
        // validate request as per JSON RPC 2.0
        else if (ret.isRequest()) {
            const bool hasParams = ret.hasParams();
            const int n_ok = hasParams ? 4 : 3;
            if (ret.data.count() != n_ok)
                throw InvalidError("Invalid request");
            if (hasParams && !ret.isParamsMap() && !ret.isParamsList())
                throw InvalidError("Invalid params");
        }
        else if (ret.isNotif()) {
            const bool hasParams = ret.hasParams();
            const int n_ok = hasParams ? 3 : 2;
            if (ret.data.count() != n_ok)
                throw InvalidError("Invalid notification");
            if (hasParams && !ret.isParamsMap() && !ret.isParamsList())
                throw InvalidError("Invalid params");
        }
        else if (ret.isResponse()) {
            const int n_ok = 3;
            if (ret.data.count() != n_ok)
                throw InvalidError("Invalid response");
        }
        else {
            throw InvalidError("Invalid JSON RPC object");
        }

        // if we get to this point, the json meets minimal JSON RPC 2.0 specs.

        return ret;
    }

    /* static */
    Message Message::makeError(int code, const QString &message, const Id & id)
    {
        Message ret;
        auto & map = ret.data;
        map["jsonrpc"] = RPC::jsonRpcVersion;
        map["id"] = id; // may be "null"
        QVariantMap errMap;
        errMap["code"] = code;
        errMap["message"] = message;
        map["error"] = errMap;
        return ret;
    }

    /// uses provided schema -- will not throw exception
    /*static*/
    Message Message::makeResponse(const Id & reqId, const QVariant & result)
    {
        Message ret;
        auto & map = ret.data;
        map["jsonrpc"] = RPC::jsonRpcVersion;
        map["id"] = reqId;
        map["result"] = result;
        return ret;
    }

    /* static */
    Message Message::makeRequest(const Id & id, const QString &methodName, const QVariantList & params)
    {
        Message ret = makeNotification(methodName, params);
        auto & map = ret.data;
        map["id"] = id;
        return ret;
    }

    /* static */
    Message Message::makeRequest(const Id & id, const QString &methodName, const QVariantMap & params)
    {
        Message ret = makeNotification(methodName, params);
        auto & map = ret.data;
        map["id"] = id;
        return ret;
    }

    /* static */
    Message Message::makeNotification(const QString &methodName, const QVariantList & params)
    {
        Message ret;
        auto & map = ret.data;
        map["jsonrpc"] = RPC::jsonRpcVersion;
        map["method"] = methodName;
        map["params"] = params;
        return ret;
    }

    /* static */
    Message Message::makeNotification(const QString &methodName, const QVariantMap & params)
    {
        Message ret;
        auto & map = ret.data;
        map["jsonrpc"] = RPC::jsonRpcVersion;
        map["method"] = methodName;
        map["params"] = params;
        return ret;
    }

    ConnectionBase::ConnectionBase(const MethodMap & methods, qint64 id_in, QObject *parent, qint64 maxBuffer)
        : AbstractConnection(id_in, parent, maxBuffer), methods(methods)
    {
    }

    ConnectionBase::~ConnectionBase() {}

    void ConnectionBase::on_connected()
    {
        AbstractConnection::on_connected();
        // connection will be auto-disconnected on socket disconnect
        connectedConns.push_back(connect(this, &ConnectionBase::sendRequest, this, &ConnectionBase::_sendRequest));
        // connection will be auto-disconnected on socket disconnect
        connectedConns.push_back(connect(this, &ConnectionBase::sendNotification, this, &ConnectionBase::_sendNotification));
        // connection will be auto-disconnected on socket disconnect
        connectedConns.push_back(connect(this, &ConnectionBase::sendError, this, &ConnectionBase::_sendError));
        // connection will be auto-disconnected on socket disconnect
        connectedConns.push_back(connect(this, &ConnectionBase::sendResult, this, &ConnectionBase::_sendResult));
    }

    void ConnectionBase::on_disconnected()
    {
        AbstractConnection::on_disconnected(); // will auto-disconnect all QMetaObject::Connections appearing in connectedConns
        idMethodMap.clear();
    }

    void ConnectionBase::_sendRequest(const Message::Id & reqid, const QString &method, const QVariantList & params)
    {
        if (status != Connected || !socket) {
            Error() << __FUNCTION__ << " method: " << method << "; Not connected!";
            return;
        }
        QString json = Message::makeRequest(reqid, method, params).toJsonString();
        if (json.isEmpty()) {
            Error() << __FUNCTION__ << " method: " << method << "; Unable to generate request JSON! FIXME!";
            return;
        }
        if (idMethodMap.size() >= MAX_UNANSWERED_REQUESTS) {  // prevent memory leaks in case of misbehaving peer
            Warning() << "Closing connection because too many unanswered requests for: " << prettyName();
            do_disconnect();
            return;
        }

        idMethodMap[reqid] = method; // remember method sent out to associate it back.

        const auto data = json.toUtf8();
        Debug() << "Sending json: " << Util::Ellipsify(data);
        // below send() ends up calling do_write immediately (which is connected to send)
        emit send( wrapForSend(data) );
    }
    void ConnectionBase::_sendNotification(const QString &method, const QVariantList & params)
    {
        if (status != Connected || !socket) {
            Error() << __FUNCTION__ << " method: " << method << "; Not connected!";
            return;
        }
        QString json = Message::makeNotification(method, params).toJsonString();
        if (json.isEmpty()) {
            Error() << __FUNCTION__ << " method: " << method << "; Unable to generate request JSON! FIXME!";
            return;
        }
        const auto data = json.toUtf8();
        Debug() << "Sending json: " << Util::Ellipsify(data);
        // below send() ends up calling do_write immediately (which is connected to send)
        emit send( wrapForSend(data) );
    }
    void ConnectionBase::_sendError(bool disc, int code, const QString &msg, const Message::Id & reqId)
    {
        if (status != Connected || !socket) {
            Error() << __FUNCTION__ << "; Not connected!";
            return;
        }
        QString json = Message::makeError(code, msg, reqId).toJsonString();
        const auto data = json.toUtf8();
        Debug() << "Sending json: " << Util::Ellipsify(data);
        // below send() ends up calling do_write immediately (which is connected to send)
        emit send( wrapForSend(data) );
        if (disc) {
            do_disconnect(true); // graceful disconnect
        }
    }
    void ConnectionBase::_sendResult(const Message::Id & reqid, const QString &method, const QVariant & result)
    {
        if (status != Connected || !socket) {
            Error() << __FUNCTION__ << " method: " << method << "; Not connected!";
            return;
        }
        QString json = Message::makeResponse(reqid, result).toJsonString();
        if (json.isEmpty()) {
            Error() << __FUNCTION__ << " method: " << method << "; Unable to generate result JSON! FIXME!";
            return;
        }
        const auto data = json.toUtf8();
        Debug() << "Sending result json: " << Util::Ellipsify(data);
        // below send() ends up calling do_write immediately (which is connected to send)
        emit send( wrapForSend(data) );
    }

    void ConnectionBase::processJson(const QByteArray &json)
    {
        Message::Id msgId;
        try {
            Message message = Message::fromJsonData( Util::Json::parseString(json, true).toMap() , &msgId); // may throw

            static const auto ValidateParams = [](const Message &msg, const Method &m) {
                if (!msg.hasParams()) {
                    if ( (m.opt_kwParams.has_value() && !m.opt_kwParams->isEmpty())
                         || (m.opt_nPosParams.has_value() && *m.opt_nPosParams != 0
                             && *m.opt_nPosParams != Method::ANY_POS_PARAMS) )
                        throw InvalidParameters("Missing required params");
                } else if (msg.isParamsList()) {
                    // positional args specified
                    if (!m.opt_nPosParams.has_value())
                        throw InvalidParameters("Postional params are not supported for this method");
                    const int num = msg.paramsList().count();
                    const int nPosParams = *m.opt_nPosParams;
                    if (nPosParams == Method::ANY_POS_PARAMS)
                        return;
                    if (nPosParams >= 0 && num != nPosParams) {
                        throw InvalidParameters(QString("Expected %1 parameters for %2, got %3 instead").arg(nPosParams).arg(m.method).arg(num));
                    } else if (nPosParams < 0 && num < -nPosParams) {
                        throw InvalidParameters(QString("Expected at least %1 parameters for %2, got %3 instead").arg(-nPosParams).arg(m.method).arg(num));
                    }
                } else if (msg.isParamsMap()) {
                    // named args specified
                    if (!m.opt_kwParams.has_value())
                        throw InvalidParameters("Named params are not supported for this method");
                    const auto nameset = KeySet::fromList(msg.paramsMap().keys());
                    const auto & kwSet = *m.opt_kwParams;
                    if (m.allowUnknownNamedParams) {
                        if (!(kwSet - nameset).isEmpty())
                            throw InvalidParameters("Required parameters missing");
                    } else {
                        if (nameset != kwSet)
                            throw InvalidParameters("Unknown or missing parameters");
                    }
                }
            };

            if (message.isError()) {
                // error message
                emit gotErrorMessage(id, message);
            } else if (message.isNotif()) {
                try {
                    const auto it = methods.find(message.method);
                    if (it == methods.end())
                        throw UnknownMethod("Unknown method");
                    const Method & m = it.value();
                    if (m.allowsNotifications) {
                        ValidateParams(message, m);
                        emit gotMessage(id, message);
                    } else {
                        throw Exception(QString("Ignoring unexpected notification"));
                    }
                } catch (const Exception & e) {
                    // Note: we emit peerError here so that the tally of number of errors goes up and we eventually disconnect the offending peer.
                    // This should not cause an error message to be sent to the peer.
                    emit peerError(this->id, QString("Error processing notification '%1' from %2: %3").arg(message.method, prettyName(), e.what()));
                }
            } else if (message.isRequest()) {
                const auto it = methods.find(message.method);
                const Method *m = it != methods.end() ? &it.value() : nullptr;
                if (!m || !m->allowsRequests)
                    throw UnknownMethod(QString("Unsupported request: %1").arg(message.method));
                ValidateParams(message, *m);
                emit gotMessage(id, message);
            } else if (message.isResponse()) {
                QString meth = idMethodMap.take(message.id);
                if (meth.isEmpty()) {
                    throw BadPeer(QString("Unexpected response (id: %1)").arg(message.id.toString()));
                }
                message.method = meth;
                emit gotMessage(id, message);
            } else {
                // Not a Request and not a Response or Notification or Error. Not JSON-RPC 2.0.
                throw InvalidRequest("Invalid JSON");
            }
            lastGood = Util::getTime(); // update "lastGood" as this is used to determine if stale or not.
        } catch (const Exception &e) {
            // TODO: clean this up. It's rather inelegant. :/
            const bool wasJsonParse = dynamic_cast<const Util::Json::ParseError *>(&e);
            const bool wasUnk = dynamic_cast<const UnknownMethod *>(&e);
            const bool wasInv = dynamic_cast<const InvalidRequest *>(&e) || dynamic_cast<const InvalidError *>(&e);
            const bool wasInvParms = dynamic_cast<const InvalidParameters *>(&e);
            int code = Code_Custom;
            if (wasJsonParse) code = Code_ParseError;
            else if (wasInvParms) code = Code_InvalidParams;
            else if (wasUnk) code = Code_MethodNotFOund;
            else if (wasInv) code = Code_InvalidRequest;
            bool doDisconnect = errorPolicy & ErrorPolicyDisconnect;
            if (errorPolicy & ErrorPolicySendErrorMessage) {
                emit sendError(doDisconnect, code, QString(e.what()).left(80), msgId);
                if (!doDisconnect)
                    emit peerError(id, e.what());
                doDisconnect = false; // if was true, already enqueued graceful disconnect after error reply, if was false, no-op here
            }
            if (doDisconnect) {
                Error() << "Error reading/parsing data coming in: " << e.what();
                do_disconnect();
                status = Bad;
            }
        } // end try/catch
    }

    /* --- LinefeedConnection --- */
    LinefeedConnection::~LinefeedConnection() {} ///< for vtable

    void LinefeedConnection::on_readyRead()
    {
        Debug() << __FUNCTION__;
        // TODO: This may be slow for large loads.
        while (socket->canReadLine()) {
            auto data = socket->readLine();
            nReceived += data.length();
#ifdef QT_DEBUG
            {
                auto line = data.trimmed();  // todo -- remove -- may be too slow
                Debug() << "Got: " << line;  // todo -- remove -- may be too slow
            }
#endif
            processJson(data);
            //Debug() << "Re-parsed message: " << message.toJsonString();
        }
        if (socket->bytesAvailable() > MAX_BUFFER) {
            Error() << prettyName() << " fatal error: " << QString("Peer has sent us more than %1 bytes without a newline! Bad peer?").arg(MAX_BUFFER);
            do_disconnect();
            status = Bad;
        }
    }

    QByteArray LinefeedConnection::wrapForSend(const QByteArray &d)
    {
        return d + "\r\n";
    }

    /* --- HttpConnection --- */
    // Very much a work in progress -- TODO: Implement
    HttpConnection::~HttpConnection() {} ///< for vtable
    void HttpConnection::setAuth(const QString &username, const QString &password)
    {
        authCookie = QString("%1:%2").arg(username).arg(password).toUtf8().toBase64();
    }
    void HttpConnection::on_readyRead()
    {
        // TODO...
    }
    QByteArray HttpConnection::wrapForSend(const QByteArray &data)
    {
        static const QByteArray NL("\r\n");
        QByteArray responseHeader;
        const QByteArray suffix = !data.endsWith("\n") ? NL : "";
        {
            QTextStream ss(&responseHeader, QIODevice::WriteOnly);
            ss.setCodec(QTextCodec::codecForName("UTF-8"));
            ss << "POST / HTTP/1.1" << NL;
            ss << "Content-Type: application/json-rpc" << NL;
            if (!authCookie.isEmpty())
                ss << "Authorization: Basic " << authCookie << NL;
            ss << "Content-Length: " << (data.length()+suffix.length()) << NL;
            ss << NL;
        }
        return responseHeader + data + suffix;
    }


} // end namespace RPC