ETM-Schurig/powersync-remoteproxy
Archived
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix package fragmentation in TCP server and client

Browse Source
This commit is contained in:
Simon Stürz 2019-06-12 18:57:14 +02:00 committed by Simon Stürz
parent 52ce0b5084
commit b49a9767c7
8 changed files with 235 additions and 151 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

159
libnymea-remoteproxy/jsonrpcserver.cpp
View File

@ -173,6 +173,85 @@ void JsonRpcServer::unregisterHandler(JsonHandler *handler)
m_handlers.remove(handler->name());
}
void JsonRpcServer::processDataPackage(ProxyClient *proxyClient, const QByteArray &data)
{
QJsonParseError error;
QJsonDocument jsonDoc = QJsonDocument::fromJson(data, &error);
if(error.error != QJsonParseError::NoError) {
qCWarning(dcJsonRpc) << "Failed to parse JSON data" << data << ":" << error.errorString();
sendErrorResponse(proxyClient, -1, QString("Failed to parse JSON data: %1").arg(error.errorString()));
proxyClient->killConnection("Invalid JSON data received.");
return;
}
QVariantMap message = jsonDoc.toVariant().toMap();
bool success = false;
int commandId = message.value("id").toInt(&success);
if (!success) {
qCWarning(dcJsonRpc()) << "Error parsing command. Missing \"id\":" << message;
sendErrorResponse(proxyClient, -1, "Error parsing command. Missing 'id'");
proxyClient->killConnection("The id property is missing in the request.");
return;
}
QStringList commandList = message.value("method").toString().split('.');
if (commandList.count() != 2) {
qCWarning(dcJsonRpc) << "Error parsing method.\nGot:" << message.value("method").toString() << "\nExpected: \"Namespace.method\"";
sendErrorResponse(proxyClient, commandId, QString("Error parsing method. Got: '%1'', Expected: 'Namespace.method'").arg(message.value("method").toString()));
proxyClient->killConnection("Invalid method passed.");
return;
}
QString targetNamespace = commandList.first();
QString method = commandList.last();
JsonHandler *handler = m_handlers.value(targetNamespace);
if (!handler) {
sendErrorResponse(proxyClient, commandId, "No such namespace");
proxyClient->killConnection("No such namespace.");
return;
}
if (!handler->hasMethod(method)) {
sendErrorResponse(proxyClient, commandId, "No such method");
proxyClient->killConnection("No such method.");
return;
}
QVariantMap params = message.value("params").toMap();
QPair<bool, QString> validationResult = handler->validateParams(method, params);
if (!validationResult.first) {
sendErrorResponse(proxyClient, commandId, "Invalid params: " + validationResult.second);
proxyClient->killConnection("Invalid params passed.");
return;
}
JsonReply *reply;
QMetaObject::invokeMethod(handler, method.toLatin1().data(), Q_RETURN_ARG(JsonReply*, reply), Q_ARG(QVariantMap, params), Q_ARG(ProxyClient *, proxyClient));
if (reply->type() == JsonReply::TypeAsync) {
m_asyncReplies.insert(reply, proxyClient);
reply->setClientId(proxyClient->clientId());
reply->setCommandId(commandId);
connect(reply, &JsonReply::finished, this, &JsonRpcServer::asyncReplyFinished);
reply->startWait();
} else {
Q_ASSERT_X((targetNamespace == "RemoteProxy" && method == "Introspect") || handler->validateReturns(method, reply->data()).first
,"validating return value", formatAssertion(targetNamespace, method, handler, reply->data()).toLatin1().data());
// QPair<bool, QString> returnValidation = reply->handler()->validateReturns(reply->method(), reply->data());
// if (!returnValidation.first) {
// qCWarning(dcJsonRpc()) << "Return value validation failed of sync reply. This should never happen. Please check the source code.";
// }
reply->setClientId(proxyClient->clientId());
reply->setCommandId(commandId);
sendResponse(proxyClient, commandId, reply->data());
reply->deleteLater();
}
}
void JsonRpcServer::setup()
{
registerHandler(this);
@ -251,80 +330,18 @@ void JsonRpcServer::processData(ProxyClient *proxyClient, const QByteArray &data
qCDebug(dcJsonRpcTraffic()) << "Incoming data from" << proxyClient << ": " << data;
QJsonParseError error;
QJsonDocument jsonDoc = QJsonDocument::fromJson(data, &error);
if(error.error != QJsonParseError::NoError) {
qCWarning(dcJsonRpc) << "Failed to parse JSON data" << data << ":" << error.errorString();
sendErrorResponse(proxyClient, -1, QString("Failed to parse JSON data: %1").arg(error.errorString()));
proxyClient->killConnection("Invalid JSON data received.");
// Handle package fragmentation
QList<QByteArray> packages = proxyClient->processData(data);
// Make sure the buffer size is in range
if (proxyClient->bufferSizeViolation()) {
qCWarning(dcJsonRpc()) << "Data buffer size violation from" << proxyClient;
proxyClient->killConnection("Data buffer size violation. Data >= 10 kB");
return;
}
QVariantMap message = jsonDoc.toVariant().toMap();
bool success = false;
int commandId = message.value("id").toInt(&success);
if (!success) {
qCWarning(dcJsonRpc()) << "Error parsing command. Missing \"id\":" << message;
sendErrorResponse(proxyClient, -1, "Error parsing command. Missing 'id'");
proxyClient->killConnection("The id property is missing in the request.");
return;
}
QStringList commandList = message.value("method").toString().split('.');
if (commandList.count() != 2) {
qCWarning(dcJsonRpc) << "Error parsing method.\nGot:" << message.value("method").toString() << "\nExpected: \"Namespace.method\"";
sendErrorResponse(proxyClient, commandId, QString("Error parsing method. Got: '%1'', Expected: 'Namespace.method'").arg(message.value("method").toString()));
proxyClient->killConnection("Invalid method passed.");
return;
}
QString targetNamespace = commandList.first();
QString method = commandList.last();
JsonHandler *handler = m_handlers.value(targetNamespace);
if (!handler) {
sendErrorResponse(proxyClient, commandId, "No such namespace");
proxyClient->killConnection("No such namespace.");
return;
}
if (!handler->hasMethod(method)) {
sendErrorResponse(proxyClient, commandId, "No such method");
proxyClient->killConnection("No such method.");
return;
}
QVariantMap params = message.value("params").toMap();
QPair<bool, QString> validationResult = handler->validateParams(method, params);
if (!validationResult.first) {
sendErrorResponse(proxyClient, commandId, "Invalid params: " + validationResult.second);
proxyClient->killConnection("Invalid params passed.");
return;
}
JsonReply *reply;
QMetaObject::invokeMethod(handler, method.toLatin1().data(), Q_RETURN_ARG(JsonReply*, reply), Q_ARG(QVariantMap, params), Q_ARG(ProxyClient *, proxyClient));
if (reply->type() == JsonReply::TypeAsync) {
m_asyncReplies.insert(reply, proxyClient);
reply->setClientId(proxyClient->clientId());
reply->setCommandId(commandId);
connect(reply, &JsonReply::finished, this, &JsonRpcServer::asyncReplyFinished);
reply->startWait();
} else {
Q_ASSERT_X((targetNamespace == "RemoteProxy" && method == "Introspect") || handler->validateReturns(method, reply->data()).first
,"validating return value", formatAssertion(targetNamespace, method, handler, reply->data()).toLatin1().data());
// QPair<bool, QString> returnValidation = reply->handler()->validateReturns(reply->method(), reply->data());
// if (!returnValidation.first) {
// qCWarning(dcJsonRpc()) << "Return value validation failed of sync reply. This should never happen. Please check the source code.";
// }
reply->setClientId(proxyClient->clientId());
reply->setCommandId(commandId);
sendResponse(proxyClient, commandId, reply->data());
reply->deleteLater();
foreach (const QByteArray &package, packages) {
processDataPackage(proxyClient, package);
}
}

2
libnymea-remoteproxy/jsonrpcserver.h
View File

@ -58,6 +58,7 @@ private:
QHash<QString, JsonHandler *> m_handlers;
QHash<JsonReply *, ProxyClient *> m_asyncReplies;
QList<ProxyClient *> m_clients;
int m_notificationId = 0;
void sendResponse(ProxyClient *client, int commandId, const QVariantMap &params = QVariantMap());
@ -67,6 +68,7 @@ private:
void registerHandler(JsonHandler *handler);
void unregisterHandler(JsonHandler *handler);
void processDataPackage(ProxyClient *proxyClient, const QByteArray &data);
private slots:
void setup();

37
libnymea-remoteproxy/proxyclient.cpp
View File

@ -27,6 +27,7 @@
#include "engine.h"
#include "proxyclient.h"
#include "loggingcategories.h"
#include <QDateTime>
@ -210,6 +211,42 @@ void ProxyClient::killConnection(const QString &reason)
m_interface->killClientConnection(m_clientId, reason);
}
int ProxyClient::generateMessageId()
{
m_messageId++;
return m_messageId;
}
QList<QByteArray> ProxyClient::processData(const QByteArray &data)
{
QList<QByteArray> packages;
// Handle packet fragmentation
m_dataBuffers.append(data);
int splitIndex = m_dataBuffers.indexOf("}\n{");
while (splitIndex > -1) {
packages.append(m_dataBuffers.left(splitIndex + 1));
m_dataBuffers = m_dataBuffers.right(m_dataBuffers.length() - splitIndex - 2);
splitIndex = m_dataBuffers.indexOf("}\n{");
}
if (m_dataBuffers.trimmed().endsWith("}")) {
packages.append(m_dataBuffers);
m_dataBuffers.clear();
}
if (m_dataBuffers.size() > 1024 * 10) {
qCWarning(dcJsonRpc()) << "Client buffer larger than 10KB and no valid data. This is a buffer size violation.";
m_bufferSizeViolation = true;
}
return packages;
}
bool ProxyClient::bufferSizeViolation() const
{
return m_bufferSizeViolation;
}
QDebug operator<<(QDebug debug, ProxyClient *proxyClient)
{
debug.nospace() << "ProxyClient(";

10
libnymea-remoteproxy/proxyclient.h
View File

@ -95,6 +95,11 @@ public:
void sendData(const QByteArray &data);
void killConnection(const QString &reason);
// Json server methods
int generateMessageId();
QList<QByteArray> processData(const QByteArray &data);
bool bufferSizeViolation() const;
private:
TransportInterface *m_interface = nullptr;
QTimer *m_timer = nullptr;
@ -114,6 +119,11 @@ private:
QString m_userName;
// Json data information
int m_messageId = 0;
QByteArray m_dataBuffers;
bool m_bufferSizeViolation = false;
quint64 m_rxDataCount = 0;
quint64 m_txDataCount = 0;

22
libnymea-remoteproxyclient/proxyjsonrpcclient.cpp
View File

@ -73,10 +73,8 @@ void JsonRpcClient::sendRequest(const QVariantMap &request)
m_connection->sendData(data);
}
void JsonRpcClient::processData(const QByteArray &data)
void JsonRpcClient::processDataPackage(const QByteArray &data)
{
qCDebug(dcRemoteProxyClientJsonRpcTraffic()) << "Received data:" << data;
QJsonParseError error;
QJsonDocument jsonDoc = QJsonDocument::fromJson(data, &error);
if (error.error != QJsonParseError::NoError) {
@ -119,4 +117,22 @@ void JsonRpcClient::processData(const QByteArray &data)
}
}
void JsonRpcClient::processData(const QByteArray &data)
{
qCDebug(dcRemoteProxyClientJsonRpcTraffic()) << "Received data:" << data;
// Handle packet fragmentation
m_dataBuffer.append(data);
int splitIndex = m_dataBuffer.indexOf("}\n{");
while (splitIndex > -1) {
processDataPackage(m_dataBuffer.left(splitIndex + 1));
m_dataBuffer = m_dataBuffer.right(m_dataBuffer.length() - splitIndex - 2);
splitIndex = m_dataBuffer.indexOf("}\n{");
}
if (m_dataBuffer.trimmed().endsWith("}")) {
processDataPackage(m_dataBuffer);
m_dataBuffer.clear();
}
}
}

2
libnymea-remoteproxyclient/proxyjsonrpcclient.h
View File

@ -54,10 +54,12 @@ private:
ProxyConnection *m_connection = nullptr;
int m_commandId = 0;
QByteArray m_dataBuffer;
QHash<int, JsonReply *> m_replies;
void sendRequest(const QVariantMap &request);
void processDataPackage(const QByteArray &data);
signals:
void tunnelEstablished(const QString clientName, const QString &clientUuid);

152
tests/test-offline/nymea-remoteproxy-tests-offline.cpp
View File

@ -446,7 +446,7 @@ void RemoteProxyOfflineTests::apiBasicCalls_data()
QTest::newRow("valid call") << QByteArray("{\"id\":42, \"method\":\"RemoteProxy.Hello\"}") << 42 << "success";
QTest::newRow("missing id") << QByteArray("{\"method\":\"RemoteProxy.Hello\"}") << -1 << "error";
QTest::newRow("missing method") << QByteArray("{\"id\":42}") << 42 << "error";
QTest::newRow("invalid json") << QByteArray("{\"id\":42, \"method\":\"RemoteProx") << -1 << "error";
QTest::newRow("invalid json") << QByteArray("{\"id\":42, \"method\":\"RemoteProx}") << -1 << "error";
QTest::newRow("invalid function") << QByteArray("{\"id\":42, \"method\":\"RemoteProxy.Explode\"}") << 42 << "error";
QTest::newRow("invalid namespace") << QByteArray("{\"id\":42, \"method\":\"ProxyRemote.Hello\"}") << 42 << "error";
QTest::newRow("missing dot") << QByteArray("{\"id\":42, \"method\":\"RemoteProxyHello\"}") << 42 << "error";
@ -1212,98 +1212,98 @@ void RemoteProxyOfflineTests::authenticationReplyConnection()
stopServer();
}
//void RemoteProxyOfflineTests::tcpRemoteConnection()
//{
// // Start the server
// startServer();
void RemoteProxyOfflineTests::tcpRemoteConnection()
{
// Start the server
startServer();
// // Configure mock authenticator
// m_mockAuthenticator->setTimeoutDuration(100);
// m_mockAuthenticator->setExpectedAuthenticationError();
// Configure mock authenticator
m_mockAuthenticator->setTimeoutDuration(100);
m_mockAuthenticator->setExpectedAuthenticationError();
// QString nameConnectionOne = "Test client one";
// QUuid uuidConnectionOne = QUuid::createUuid();
QString nameConnectionOne = "Test client one";
QUuid uuidConnectionOne = QUuid::createUuid();
// QString nameConnectionTwo = "Test client two";
// QUuid uuidConnectionTwo = QUuid::createUuid();
QString nameConnectionTwo = "Test client two";
QUuid uuidConnectionTwo = QUuid::createUuid();
// QByteArray dataOne = "Hello from client one :-)";
// QByteArray dataTwo = "Hello from client two :-)";
QByteArray dataOne = "Hello from client one :-)";
QByteArray dataTwo = "Hello from client two :-)";
// // Create two connection
// RemoteProxyConnection *connectionOne = new RemoteProxyConnection(uuidConnectionOne, nameConnectionOne, RemoteProxyConnection::ConnectionTypeTcpSocket, this);
// connect(connectionOne, &RemoteProxyConnection::sslErrors, this, &BaseTest::ignoreConnectionSslError);
// Create two connection
RemoteProxyConnection *connectionOne = new RemoteProxyConnection(uuidConnectionOne, nameConnectionOne, RemoteProxyConnection::ConnectionTypeTcpSocket, this);
connect(connectionOne, &RemoteProxyConnection::sslErrors, this, &BaseTest::ignoreConnectionSslError);
// RemoteProxyConnection *connectionTwo = new RemoteProxyConnection(uuidConnectionTwo, nameConnectionTwo, RemoteProxyConnection::ConnectionTypeTcpSocket, this);
// connect(connectionTwo, &RemoteProxyConnection::sslErrors, this, &BaseTest::ignoreConnectionSslError);
RemoteProxyConnection *connectionTwo = new RemoteProxyConnection(uuidConnectionTwo, nameConnectionTwo, RemoteProxyConnection::ConnectionTypeTcpSocket, this);
connect(connectionTwo, &RemoteProxyConnection::sslErrors, this, &BaseTest::ignoreConnectionSslError);
// // Connect one
// QSignalSpy connectionOneReadySpy(connectionOne, &RemoteProxyConnection::ready);
// QVERIFY(connectionOne->connectServer(m_serverUrlTcp));
// connectionOneReadySpy.wait();
// QVERIFY(connectionOneReadySpy.count() == 1);
// QVERIFY(connectionOne->isConnected());
// Connect one
QSignalSpy connectionOneReadySpy(connectionOne, &RemoteProxyConnection::ready);
QVERIFY(connectionOne->connectServer(m_serverUrlTcp));
connectionOneReadySpy.wait();
QVERIFY(connectionOneReadySpy.count() == 1);
QVERIFY(connectionOne->isConnected());
// // Connect two
// QSignalSpy connectionTwoReadySpy(connectionTwo, &RemoteProxyConnection::ready);
// QVERIFY(connectionTwo->connectServer(m_serverUrlTcp));
// connectionTwoReadySpy.wait();
// QVERIFY(connectionTwoReadySpy.count() == 1);
// QVERIFY(connectionTwo->isConnected());
// Connect two
QSignalSpy connectionTwoReadySpy(connectionTwo, &RemoteProxyConnection::ready);
QVERIFY(connectionTwo->connectServer(m_serverUrlTcp));
connectionTwoReadySpy.wait();
QVERIFY(connectionTwoReadySpy.count() == 1);
QVERIFY(connectionTwo->isConnected());
// // Authenticate one
// QSignalSpy remoteConnectionEstablishedOne(connectionOne, &RemoteProxyConnection::remoteConnectionEstablished);
// QSignalSpy connectionOneAuthenticatedSpy(connectionOne, &RemoteProxyConnection::authenticated);
// QVERIFY(connectionOne->authenticate(m_testToken));
// connectionOneAuthenticatedSpy.wait();
// QVERIFY(connectionOneAuthenticatedSpy.count() == 1);
// QVERIFY(connectionOne->isConnected());
// QVERIFY(connectionOne->isAuthenticated());
// QVERIFY(connectionOne->state() == RemoteProxyConnection::StateAuthenticated);
// Authenticate one
QSignalSpy remoteConnectionEstablishedOne(connectionOne, &RemoteProxyConnection::remoteConnectionEstablished);
QSignalSpy connectionOneAuthenticatedSpy(connectionOne, &RemoteProxyConnection::authenticated);
QVERIFY(connectionOne->authenticate(m_testToken));
connectionOneAuthenticatedSpy.wait();
QVERIFY(connectionOneAuthenticatedSpy.count() == 1);
QVERIFY(connectionOne->isConnected());
QVERIFY(connectionOne->isAuthenticated());
QVERIFY(connectionOne->state() == RemoteProxyConnection::StateAuthenticated);
// // Authenticate two
// QSignalSpy remoteConnectionEstablishedTwo(connectionTwo, &RemoteProxyConnection::remoteConnectionEstablished);
// QSignalSpy connectionTwoAuthenticatedSpy(connectionTwo, &RemoteProxyConnection::authenticated);
// QVERIFY(connectionTwo->authenticate(m_testToken));
// connectionTwoAuthenticatedSpy.wait();
// qDebug() << connectionTwoAuthenticatedSpy.count();
// QVERIFY(connectionTwoAuthenticatedSpy.count() == 1);
// QVERIFY(connectionTwo->isConnected());
// QVERIFY(connectionTwo->isAuthenticated());
// Authenticate two
QSignalSpy remoteConnectionEstablishedTwo(connectionTwo, &RemoteProxyConnection::remoteConnectionEstablished);
QSignalSpy connectionTwoAuthenticatedSpy(connectionTwo, &RemoteProxyConnection::authenticated);
QVERIFY(connectionTwo->authenticate(m_testToken));
connectionTwoAuthenticatedSpy.wait();
qDebug() << connectionTwoAuthenticatedSpy.count();
QVERIFY(connectionTwoAuthenticatedSpy.count() == 1);
QVERIFY(connectionTwo->isConnected());
QVERIFY(connectionTwo->isAuthenticated());
// // Wait for both to be connected
// remoteConnectionEstablishedOne.wait(500);
// remoteConnectionEstablishedTwo.wait(500);
// Wait for both to be connected
remoteConnectionEstablishedOne.wait(500);
remoteConnectionEstablishedTwo.wait(500);
// QVERIFY(remoteConnectionEstablishedOne.count() == 1);
// QVERIFY(remoteConnectionEstablishedTwo.count() == 1);
// QVERIFY(connectionOne->state() == RemoteProxyConnection::StateRemoteConnected);
// QVERIFY(connectionTwo->state() == RemoteProxyConnection::StateRemoteConnected);
QVERIFY(remoteConnectionEstablishedOne.count() == 1);
QVERIFY(remoteConnectionEstablishedTwo.count() == 1);
QVERIFY(connectionOne->state() == RemoteProxyConnection::StateRemoteConnected);
QVERIFY(connectionTwo->state() == RemoteProxyConnection::StateRemoteConnected);
// QCOMPARE(connectionOne->tunnelPartnerName(), nameConnectionTwo);
// QCOMPARE(connectionOne->tunnelPartnerUuid(), uuidConnectionTwo.toString());
// QCOMPARE(connectionTwo->tunnelPartnerName(), nameConnectionOne);
// QCOMPARE(connectionTwo->tunnelPartnerUuid(), uuidConnectionOne.toString());
QCOMPARE(connectionOne->tunnelPartnerName(), nameConnectionTwo);
QCOMPARE(connectionOne->tunnelPartnerUuid(), uuidConnectionTwo.toString());
QCOMPARE(connectionTwo->tunnelPartnerName(), nameConnectionOne);
QCOMPARE(connectionTwo->tunnelPartnerUuid(), uuidConnectionOne.toString());
// // Pipe data trought the tunnel
// QSignalSpy remoteConnectionDataOne(connectionOne, &RemoteProxyConnection::dataReady);
// QSignalSpy remoteConnectionDataTwo(connectionTwo, &RemoteProxyConnection::dataReady);
// Pipe data trought the tunnel
QSignalSpy remoteConnectionDataOne(connectionOne, &RemoteProxyConnection::dataReady);
QSignalSpy remoteConnectionDataTwo(connectionTwo, &RemoteProxyConnection::dataReady);
// connectionOne->sendData(dataOne);
// remoteConnectionDataTwo.wait(500);
// QVERIFY(remoteConnectionDataTwo.count() == 1);
// QCOMPARE(remoteConnectionDataTwo.at(0).at(0).toByteArray().trimmed(), dataOne);
connectionOne->sendData(dataOne);
remoteConnectionDataTwo.wait(500);
QVERIFY(remoteConnectionDataTwo.count() == 1);
QCOMPARE(remoteConnectionDataTwo.at(0).at(0).toByteArray().trimmed(), dataOne);
// connectionTwo->sendData(dataTwo);
// remoteConnectionDataOne.wait(500);
// QVERIFY(remoteConnectionDataOne.count() == 1);
// QCOMPARE(remoteConnectionDataOne.at(0).at(0).toByteArray().trimmed(), dataTwo);
connectionTwo->sendData(dataTwo);
remoteConnectionDataOne.wait(500);
QVERIFY(remoteConnectionDataOne.count() == 1);
QCOMPARE(remoteConnectionDataOne.at(0).at(0).toByteArray().trimmed(), dataTwo);
// connectionOne->deleteLater();
// connectionTwo->deleteLater();
connectionOne->deleteLater();
connectionTwo->deleteLater();
// // Clean up
// stopServer();
//}
// Clean up
stopServer();
}
QTEST_MAIN(RemoteProxyOfflineTests)

2
tests/test-offline/nymea-remoteproxy-tests-offline.h
View File

@ -82,7 +82,7 @@ private slots:
void authenticationReplyConnection();
// TCP Websocket combinations
//void tcpRemoteConnection();
void tcpRemoteConnection();
};