Mirrored_Repos/obs-websocket
1
0
Fork 0
mirror of https://github.com/Palakis/obs-websocket.git synced 2024-08-30 18:12:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

base: Use Unix EOL

Browse Source
This commit is contained in:
tt2468 2021-04-29 21:13:34 -07:00
parent 904e866a07
commit e151a9a8db
10 changed files with 614 additions and 614 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/WebSocketProtocol.h
View File

@ -15,4 +15,4 @@ namespace WebSocketProtocol {
};
ProcessResult ProcessMessage(websocketpp::connection_hdl hdl, WebSocketSession *session, json incomingMessage);
}
}

808
src/WebSocketServer.cpp
View File

@ -1,404 +1,404 @@
#include <chrono>
#include <thread>
#include <QtConcurrent>
#include <QDateTime>
#include <QTime>
#include "WebSocketServer.h"
#include "WebSocketProtocol.h"
#include "obs-websocket.h"
#include "Config.h"
#include "utils/Utils.h"
#include "plugin-macros.generated.h"
WebSocketServer::WebSocketServer() :
QObject(nullptr),
_sessions()
{
// Randomize the random number generator
qsrand(QTime::currentTime().msec());
_server.get_alog().clear_channels(websocketpp::log::alevel::all);
_server.get_elog().clear_channels(websocketpp::log::elevel::all);
_server.init_asio();
#ifndef _WIN32
_server.set_reuse_addr(true);
#endif
_server.set_open_handler(
websocketpp::lib::bind(
&WebSocketServer::onOpen, this, websocketpp::lib::placeholders::_1
)
);
_server.set_close_handler(
websocketpp::lib::bind(
&WebSocketServer::onClose, this, websocketpp::lib::placeholders::_1
)
);
_server.set_message_handler(
websocketpp::lib::bind(
&WebSocketServer::onMessage, this, websocketpp::lib::placeholders::_1, websocketpp::lib::placeholders::_2
)
);
}
WebSocketServer::~WebSocketServer()
{
if (_server.is_listening())
Stop();
}
void WebSocketServer::ServerRunner()
{
blog(LOG_INFO, "[WebSocketServer::ServerRunner] IO thread started.");
try {
_server.run();
} catch (websocketpp::exception const & e) {
blog(LOG_ERROR, "[WebSocketServer::ServerRunner] websocketpp instance returned an error: %s", e.what());
} catch (const std::exception & e) {
blog(LOG_ERROR, "[WebSocketServer::ServerRunner] websocketpp instance returned an error: %s", e.what());
} catch (...) {
blog(LOG_ERROR, "[WebSocketServer::ServerRunner] websocketpp instance returned an error");
}
blog(LOG_INFO, "[WebSocketServer::ServerRunner] IO thread exited.");
}
void WebSocketServer::Start()
{
if (_server.is_listening()) {
blog(LOG_WARNING, "[WebSocketServer::Start] Call to Start() but the server is already listening.");
return;
}
auto conf = GetConfig();
if (!conf) {
blog(LOG_ERROR, "[WebSocketServer::Start] Unable to retreive config!");
return;
}
_serverPort = conf->ServerPort;
_serverPassword = conf->ServerPassword;
_debugEnabled = conf->DebugEnabled;
AuthenticationRequired = conf->AuthRequired;
AuthenticationSalt = Utils::Crypto::GenerateSalt();
AuthenticationSecret = Utils::Crypto::GenerateSecret(conf->ServerPassword.toStdString(), AuthenticationSalt);
// Set log levels if debug is enabled
if (_debugEnabled) {
_server.get_alog().set_channels(websocketpp::log::alevel::all);
_server.get_alog().clear_channels(websocketpp::log::alevel::frame_header | websocketpp::log::alevel::frame_payload | websocketpp::log::alevel::control);
_server.get_elog().set_channels(websocketpp::log::elevel::all);
_server.get_alog().clear_channels(websocketpp::log::elevel::devel | websocketpp::log::elevel::library);
} else {
_server.get_alog().clear_channels(websocketpp::log::alevel::all);
_server.get_elog().clear_channels(websocketpp::log::elevel::all);
}
_server.reset();
websocketpp::lib::error_code errorCode;
_server.listen(websocketpp::lib::asio::ip::tcp::v4(), _serverPort, errorCode);
if (errorCode) {
std::string errorCodeMessage = errorCode.message();
blog(LOG_INFO, "[WebSocketServer::Start] Listen failed: %s", errorCodeMessage.c_str());
return;
}
_server.start_accept();
_serverThread = std::thread(&WebSocketServer::ServerRunner, this);
blog(LOG_INFO, "[WebSocketServer::Start] Server started successfully on port %d. Possible connect address: %s", _serverPort, Utils::Platform::GetLocalAddress().c_str());
}
void WebSocketServer::Stop()
{
if (!_server.is_listening()) {
blog(LOG_WARNING, "[WebSocketServer::Stop] Call to Stop() but the server is not listening.");
return;
}
_server.stop_listening();
std::unique_lock<std::mutex> lock(_sessionMutex);
for (auto const& [hdl, session] : _sessions) {
websocketpp::lib::error_code errorCode;
_server.pause_reading(hdl, errorCode);
if (errorCode) {
blog(LOG_INFO, "[WebSocketServer::Stop] Error: %s", errorCode.message().c_str());
continue;
}
_server.close(hdl, websocketpp::close::status::going_away, "Server stopping.", errorCode);
if (errorCode) {
blog(LOG_INFO, "[WebSocketServer::Stop] Error: %s", errorCode.message().c_str());
continue;
}
}
lock.unlock();
_threadPool.waitForDone();
// This can delay the thread that it is running on. Bad but kinda required.
while (_sessions.size() > 0) {
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
_serverThread.join();
blog(LOG_INFO, "[WebSocketServer::Stop] Server stopped successfully");
}
void WebSocketServer::InvalidateSession(websocketpp::connection_hdl hdl)
{
blog(LOG_INFO, "[WebSocketServer::InvalidateSession] Invalidating a session.");
websocketpp::lib::error_code errorCode;
_server.pause_reading(hdl, errorCode);
if (errorCode) {
blog(LOG_INFO, "[WebSocketServer::InvalidateSession] Error: %s", errorCode.message().c_str());
return;
}
_server.close(hdl, WebSocketCloseCode::SessionInvalidated, "Your session has been invalidated.", errorCode);
if (errorCode) {
blog(LOG_INFO, "[WebSocketServer::InvalidateSession] Error: %s", errorCode.message().c_str());
return;
}
}
std::vector<WebSocketServer::WebSocketSessionState> WebSocketServer::GetWebSocketSessions()
{
std::vector<WebSocketServer::WebSocketSessionState> webSocketSessions;
std::unique_lock<std::mutex> lock(_sessionMutex);
for (auto & [hdl, session] : _sessions) {
uint64_t connectedAt = session.ConnectedAt();
uint64_t incomingMessages = session.IncomingMessages();
uint64_t outgoingMessages = session.OutgoingMessages();
std::string remoteAddress = session.RemoteAddress();
webSocketSessions.emplace_back(WebSocketSessionState{hdl, remoteAddress, connectedAt, incomingMessages, outgoingMessages});
}
lock.unlock();
return webSocketSessions;
}
QString WebSocketServer::GetConnectString()
{
QString ret;
if (AuthenticationRequired)
ret = QString("obswebsocket|%1:%2|%3").arg(QString::fromStdString(Utils::Platform::GetLocalAddress())).arg(_serverPort).arg(_serverPassword);
else
ret = QString("obswebsocket|%1:%2").arg(QString::fromStdString(Utils::Platform::GetLocalAddress())).arg(_serverPort);
return ret;
}
// It isn't consistent to directly call the WebSocketServer from the events system, but it would also be dumb to make it unnecessarily complicated.
void WebSocketServer::BroadcastEvent(uint64_t requiredIntent, std::string eventType, json eventData)
{
QtConcurrent::run(&_threadPool, [=]() {
// Populate message object
json eventMessage;
eventMessage["messageType"] = "Event";
eventMessage["eventType"] = eventType;
if (eventData.is_object())
eventMessage["eventData"] = eventData;
// Initialize objects. The broadcast process only dumps the data when its needed.
std::string messageJson;
std::string messageMsgPack;
// Recurse connected sessions and send the event to suitable sessions.
std::unique_lock<std::mutex> lock(_sessionMutex);
for (auto & it : _sessions) {
if (!it.second.IsIdentified())
continue;
if ((it.second.EventSubscriptions() & requiredIntent) != 0) {
websocketpp::lib::error_code errorCode;
switch (it.second.Encoding()) {
case WebSocketEncoding::Json:
if (messageJson.empty()) {
messageJson = eventMessage.dump();
}
_server.send((websocketpp::connection_hdl)it.first, messageJson, websocketpp::frame::opcode::text, errorCode);
it.second.IncrementOutgoingMessages();
break;
case WebSocketEncoding::MsgPack:
if (messageMsgPack.empty()) {
auto msgPackData = json::to_msgpack(eventMessage);
messageMsgPack = std::string(msgPackData.begin(), msgPackData.end());
}
_server.send((websocketpp::connection_hdl)it.first, messageMsgPack, websocketpp::frame::opcode::binary, errorCode);
it.second.IncrementOutgoingMessages();
break;
}
}
}
lock.unlock();
if (_debugEnabled)
blog(LOG_INFO, "[WebSocketServer::BroadcastEvent] Outgoing event:\n%s", eventMessage.dump(2).c_str());
});
}
void WebSocketServer::onOpen(websocketpp::connection_hdl hdl)
{
auto conn = _server.get_con_from_hdl(hdl);
// Build new session
std::unique_lock<std::mutex> lock(_sessionMutex);
auto &session = _sessions[hdl];
lock.unlock();
// Configure session details
session.SetRemoteAddress(conn->get_remote_endpoint());
session.SetConnectedAt(QDateTime::currentSecsSinceEpoch());
std::string contentType = conn->get_request_header("Content-Type");
if (contentType == "") {
;
} else if (contentType == "application/json") {
session.SetEncoding(WebSocketEncoding::Json);
} else if (contentType == "application/msgpack") {
session.SetEncoding(WebSocketEncoding::MsgPack);
} else {
conn->close(WebSocketCloseCode::InvalidContentType, "Your HTTP `Content-Type` header specifies an invalid encoding type.");
return;
}
// Build `Hello`
json helloMessage;
helloMessage["messageType"] = "Hello";
helloMessage["obsWebSocketVersion"] = OBS_WEBSOCKET_VERSION;
helloMessage["rpcVersion"] = OBS_WEBSOCKET_RPC_VERSION;
// todo: Add request and event lists
if (AuthenticationRequired) {
std::string sessionChallenge = Utils::Crypto::GenerateSalt();
session.SetChallenge(sessionChallenge);
helloMessage["authentication"] = {};
helloMessage["authentication"]["challenge"] = sessionChallenge;
helloMessage["authentication"]["salt"] = AuthenticationSalt;
}
// Send object to client
websocketpp::lib::error_code errorCode;
auto sessionEncoding = session.Encoding();
if (sessionEncoding == WebSocketEncoding::Json) {
std::string helloMessageJson = helloMessage.dump();
_server.send(hdl, helloMessageJson, websocketpp::frame::opcode::text, errorCode);
} else if (sessionEncoding == WebSocketEncoding::MsgPack) {
auto msgPackData = json::to_msgpack(helloMessage);
std::string messageMsgPack(msgPackData.begin(), msgPackData.end());
_server.send(hdl, messageMsgPack, websocketpp::frame::opcode::binary, errorCode);
}
session.IncrementOutgoingMessages();
}
void WebSocketServer::onClose(websocketpp::connection_hdl hdl)
{
auto conn = _server.get_con_from_hdl(hdl);
// Get info from the session and then delete it
std::unique_lock<std::mutex> lock(_sessionMutex);
auto &session = _sessions[hdl];
bool isIdentified = session.IsIdentified();
uint64_t connectedAt = session.ConnectedAt();
uint64_t incomingMessages = session.IncomingMessages();
uint64_t outgoingMessages = session.OutgoingMessages();
std::string remoteAddress = session.RemoteAddress();
_sessions.erase(hdl);
lock.unlock();
// Build SessionState object for signal
WebSocketSessionState state;
state.remoteAddress = remoteAddress;
state.connectedAt = connectedAt;
state.incomingMessages = incomingMessages;
state.outgoingMessages = outgoingMessages;
// Emit signals
emit ClientDisconnected(state, conn->get_local_close_code());
if (isIdentified)
emit IdentifiedClientDisconnected(state, conn->get_local_close_code());
}
void WebSocketServer::onMessage(websocketpp::connection_hdl hdl, websocketpp::server<websocketpp::config::asio>::message_ptr message)
{
auto opcode = message->get_opcode();
std::string payload = message->get_payload();
QtConcurrent::run(&_threadPool, [=]() {
std::unique_lock<std::mutex> lock(_sessionMutex);
auto &session = _sessions[hdl];
lock.unlock();
session.IncrementIncomingMessages();
json incomingMessage;
// Check for invalid opcode and decode
websocketpp::lib::error_code errorCode;
uint8_t sessionEncoding = session.Encoding();
if (sessionEncoding == WebSocketEncoding::Json) {
if (opcode != websocketpp::frame::opcode::text) {
if (!session.IgnoreInvalidMessages())
_server.close(hdl, WebSocketCloseCode::MessageDecodeError, "Your session encoding is set to Json, but a binary message was received.", errorCode);
return;
}
try {
incomingMessage = json::parse(payload);
} catch (json::parse_error& e) {
if (!session.IgnoreInvalidMessages())
_server.close(hdl, WebSocketCloseCode::MessageDecodeError, std::string("Unable to decode Json: ") + e.what(), errorCode);
return;
}
} else if (sessionEncoding == WebSocketEncoding::MsgPack) {
if (opcode != websocketpp::frame::opcode::binary) {
if (!session.IgnoreInvalidMessages())
_server.close(hdl, WebSocketCloseCode::MessageDecodeError, "Your session encoding is set to MsgPack, but a text message was received.", errorCode);
return;
}
try {
incomingMessage = json::from_msgpack(payload);
} catch (json::parse_error& e) {
if (!session.IgnoreInvalidMessages())
_server.close(hdl, WebSocketCloseCode::MessageDecodeError, std::string("Unable to decode MsgPack: ") + e.what(), errorCode);
return;
}
}
if (_debugEnabled)
blog(LOG_INFO, "[WebSocketServer::onMessage] Incoming message (decoded):\n%s", incomingMessage.dump(2).c_str());
WebSocketProtocol::ProcessResult ret = WebSocketProtocol::ProcessMessage(hdl, &session, incomingMessage);
if (ret.closeCode != WebSocketCloseCode::DontClose) {
websocketpp::lib::error_code errorCode;
_server.close(hdl, ret.closeCode, ret.closeReason, errorCode);
return;
}
if (!ret.result.is_null()) {
websocketpp::lib::error_code errorCode;
if (sessionEncoding == WebSocketEncoding::Json) {
std::string helloMessageJson = ret.result.dump();
_server.send(hdl, helloMessageJson, websocketpp::frame::opcode::text, errorCode);
} else if (sessionEncoding == WebSocketEncoding::MsgPack) {
auto msgPackData = json::to_msgpack(ret.result);
std::string messageMsgPack(msgPackData.begin(), msgPackData.end());
_server.send(hdl, messageMsgPack, websocketpp::frame::opcode::binary, errorCode);
}
session.IncrementOutgoingMessages();
if (_debugEnabled)
blog(LOG_INFO, "[WebSocketServer::onMessage] Outgoing message:\n%s", ret.result.dump(2).c_str());
if (errorCode)
blog(LOG_WARNING, "[WebSocketServer::onMessage] Sending message to client failed: %s", errorCode.message().c_str());
}
});
}
#include <chrono>
#include <thread>
#include <QtConcurrent>
#include <QDateTime>
#include <QTime>
#include "WebSocketServer.h"
#include "WebSocketProtocol.h"
#include "obs-websocket.h"
#include "Config.h"
#include "utils/Utils.h"
#include "plugin-macros.generated.h"
WebSocketServer::WebSocketServer() :
QObject(nullptr),
_sessions()
{
// Randomize the random number generator
qsrand(QTime::currentTime().msec());
_server.get_alog().clear_channels(websocketpp::log::alevel::all);
_server.get_elog().clear_channels(websocketpp::log::elevel::all);
_server.init_asio();
#ifndef _WIN32
_server.set_reuse_addr(true);
#endif
_server.set_open_handler(
websocketpp::lib::bind(
&WebSocketServer::onOpen, this, websocketpp::lib::placeholders::_1
)
);
_server.set_close_handler(
websocketpp::lib::bind(
&WebSocketServer::onClose, this, websocketpp::lib::placeholders::_1
)
);
_server.set_message_handler(
websocketpp::lib::bind(
&WebSocketServer::onMessage, this, websocketpp::lib::placeholders::_1, websocketpp::lib::placeholders::_2
)
);
}
WebSocketServer::~WebSocketServer()
{
if (_server.is_listening())
Stop();
}
void WebSocketServer::ServerRunner()
{
blog(LOG_INFO, "[WebSocketServer::ServerRunner] IO thread started.");
try {
_server.run();
} catch (websocketpp::exception const & e) {
blog(LOG_ERROR, "[WebSocketServer::ServerRunner] websocketpp instance returned an error: %s", e.what());
} catch (const std::exception & e) {
blog(LOG_ERROR, "[WebSocketServer::ServerRunner] websocketpp instance returned an error: %s", e.what());
} catch (...) {
blog(LOG_ERROR, "[WebSocketServer::ServerRunner] websocketpp instance returned an error");
}
blog(LOG_INFO, "[WebSocketServer::ServerRunner] IO thread exited.");
}
void WebSocketServer::Start()
{
if (_server.is_listening()) {
blog(LOG_WARNING, "[WebSocketServer::Start] Call to Start() but the server is already listening.");
return;
}
auto conf = GetConfig();
if (!conf) {
blog(LOG_ERROR, "[WebSocketServer::Start] Unable to retreive config!");
return;
}
_serverPort = conf->ServerPort;
_serverPassword = conf->ServerPassword;
_debugEnabled = conf->DebugEnabled;
AuthenticationRequired = conf->AuthRequired;
AuthenticationSalt = Utils::Crypto::GenerateSalt();
AuthenticationSecret = Utils::Crypto::GenerateSecret(conf->ServerPassword.toStdString(), AuthenticationSalt);
// Set log levels if debug is enabled
if (_debugEnabled) {
_server.get_alog().set_channels(websocketpp::log::alevel::all);
_server.get_alog().clear_channels(websocketpp::log::alevel::frame_header | websocketpp::log::alevel::frame_payload | websocketpp::log::alevel::control);
_server.get_elog().set_channels(websocketpp::log::elevel::all);
_server.get_alog().clear_channels(websocketpp::log::elevel::devel | websocketpp::log::elevel::library);
} else {
_server.get_alog().clear_channels(websocketpp::log::alevel::all);
_server.get_elog().clear_channels(websocketpp::log::elevel::all);
}
_server.reset();
websocketpp::lib::error_code errorCode;
_server.listen(websocketpp::lib::asio::ip::tcp::v4(), _serverPort, errorCode);
if (errorCode) {
std::string errorCodeMessage = errorCode.message();
blog(LOG_INFO, "[WebSocketServer::Start] Listen failed: %s", errorCodeMessage.c_str());
return;
}
_server.start_accept();
_serverThread = std::thread(&WebSocketServer::ServerRunner, this);
blog(LOG_INFO, "[WebSocketServer::Start] Server started successfully on port %d. Possible connect address: %s", _serverPort, Utils::Platform::GetLocalAddress().c_str());
}
void WebSocketServer::Stop()
{
if (!_server.is_listening()) {
blog(LOG_WARNING, "[WebSocketServer::Stop] Call to Stop() but the server is not listening.");
return;
}
_server.stop_listening();
std::unique_lock<std::mutex> lock(_sessionMutex);
for (auto const& [hdl, session] : _sessions) {
websocketpp::lib::error_code errorCode;
_server.pause_reading(hdl, errorCode);
if (errorCode) {
blog(LOG_INFO, "[WebSocketServer::Stop] Error: %s", errorCode.message().c_str());
continue;
}
_server.close(hdl, websocketpp::close::status::going_away, "Server stopping.", errorCode);
if (errorCode) {
blog(LOG_INFO, "[WebSocketServer::Stop] Error: %s", errorCode.message().c_str());
continue;
}
}
lock.unlock();
_threadPool.waitForDone();
// This can delay the thread that it is running on. Bad but kinda required.
while (_sessions.size() > 0) {
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
_serverThread.join();
blog(LOG_INFO, "[WebSocketServer::Stop] Server stopped successfully");
}
void WebSocketServer::InvalidateSession(websocketpp::connection_hdl hdl)
{
blog(LOG_INFO, "[WebSocketServer::InvalidateSession] Invalidating a session.");
websocketpp::lib::error_code errorCode;
_server.pause_reading(hdl, errorCode);
if (errorCode) {
blog(LOG_INFO, "[WebSocketServer::InvalidateSession] Error: %s", errorCode.message().c_str());
return;
}
_server.close(hdl, WebSocketCloseCode::SessionInvalidated, "Your session has been invalidated.", errorCode);
if (errorCode) {
blog(LOG_INFO, "[WebSocketServer::InvalidateSession] Error: %s", errorCode.message().c_str());
return;
}
}
std::vector<WebSocketServer::WebSocketSessionState> WebSocketServer::GetWebSocketSessions()
{
std::vector<WebSocketServer::WebSocketSessionState> webSocketSessions;
std::unique_lock<std::mutex> lock(_sessionMutex);
for (auto & [hdl, session] : _sessions) {
uint64_t connectedAt = session.ConnectedAt();
uint64_t incomingMessages = session.IncomingMessages();
uint64_t outgoingMessages = session.OutgoingMessages();
std::string remoteAddress = session.RemoteAddress();
webSocketSessions.emplace_back(WebSocketSessionState{hdl, remoteAddress, connectedAt, incomingMessages, outgoingMessages});
}
lock.unlock();
return webSocketSessions;
}
QString WebSocketServer::GetConnectString()
{
QString ret;
if (AuthenticationRequired)
ret = QString("obswebsocket|%1:%2|%3").arg(QString::fromStdString(Utils::Platform::GetLocalAddress())).arg(_serverPort).arg(_serverPassword);
else
ret = QString("obswebsocket|%1:%2").arg(QString::fromStdString(Utils::Platform::GetLocalAddress())).arg(_serverPort);
return ret;
}
// It isn't consistent to directly call the WebSocketServer from the events system, but it would also be dumb to make it unnecessarily complicated.
void WebSocketServer::BroadcastEvent(uint64_t requiredIntent, std::string eventType, json eventData)
{
QtConcurrent::run(&_threadPool, [=]() {
// Populate message object
json eventMessage;
eventMessage["messageType"] = "Event";
eventMessage["eventType"] = eventType;
if (eventData.is_object())
eventMessage["eventData"] = eventData;
// Initialize objects. The broadcast process only dumps the data when its needed.
std::string messageJson;
std::string messageMsgPack;
// Recurse connected sessions and send the event to suitable sessions.
std::unique_lock<std::mutex> lock(_sessionMutex);
for (auto & it : _sessions) {
if (!it.second.IsIdentified())
continue;
if ((it.second.EventSubscriptions() & requiredIntent) != 0) {
websocketpp::lib::error_code errorCode;
switch (it.second.Encoding()) {
case WebSocketEncoding::Json:
if (messageJson.empty()) {
messageJson = eventMessage.dump();
}
_server.send((websocketpp::connection_hdl)it.first, messageJson, websocketpp::frame::opcode::text, errorCode);
it.second.IncrementOutgoingMessages();
break;
case WebSocketEncoding::MsgPack:
if (messageMsgPack.empty()) {
auto msgPackData = json::to_msgpack(eventMessage);
messageMsgPack = std::string(msgPackData.begin(), msgPackData.end());
}
_server.send((websocketpp::connection_hdl)it.first, messageMsgPack, websocketpp::frame::opcode::binary, errorCode);
it.second.IncrementOutgoingMessages();
break;
}
}
}
lock.unlock();
if (_debugEnabled)
blog(LOG_INFO, "[WebSocketServer::BroadcastEvent] Outgoing event:\n%s", eventMessage.dump(2).c_str());
});
}
void WebSocketServer::onOpen(websocketpp::connection_hdl hdl)
{
auto conn = _server.get_con_from_hdl(hdl);
// Build new session
std::unique_lock<std::mutex> lock(_sessionMutex);
auto &session = _sessions[hdl];
lock.unlock();
// Configure session details
session.SetRemoteAddress(conn->get_remote_endpoint());
session.SetConnectedAt(QDateTime::currentSecsSinceEpoch());
std::string contentType = conn->get_request_header("Content-Type");
if (contentType == "") {
;
} else if (contentType == "application/json") {
session.SetEncoding(WebSocketEncoding::Json);
} else if (contentType == "application/msgpack") {
session.SetEncoding(WebSocketEncoding::MsgPack);
} else {
conn->close(WebSocketCloseCode::InvalidContentType, "Your HTTP `Content-Type` header specifies an invalid encoding type.");
return;
}
// Build `Hello`
json helloMessage;
helloMessage["messageType"] = "Hello";
helloMessage["obsWebSocketVersion"] = OBS_WEBSOCKET_VERSION;
helloMessage["rpcVersion"] = OBS_WEBSOCKET_RPC_VERSION;
// todo: Add request and event lists
if (AuthenticationRequired) {
std::string sessionChallenge = Utils::Crypto::GenerateSalt();
session.SetChallenge(sessionChallenge);
helloMessage["authentication"] = {};
helloMessage["authentication"]["challenge"] = sessionChallenge;
helloMessage["authentication"]["salt"] = AuthenticationSalt;
}
// Send object to client
websocketpp::lib::error_code errorCode;
auto sessionEncoding = session.Encoding();
if (sessionEncoding == WebSocketEncoding::Json) {
std::string helloMessageJson = helloMessage.dump();
_server.send(hdl, helloMessageJson, websocketpp::frame::opcode::text, errorCode);
} else if (sessionEncoding == WebSocketEncoding::MsgPack) {
auto msgPackData = json::to_msgpack(helloMessage);
std::string messageMsgPack(msgPackData.begin(), msgPackData.end());
_server.send(hdl, messageMsgPack, websocketpp::frame::opcode::binary, errorCode);
}
session.IncrementOutgoingMessages();
}
void WebSocketServer::onClose(websocketpp::connection_hdl hdl)
{
auto conn = _server.get_con_from_hdl(hdl);
// Get info from the session and then delete it
std::unique_lock<std::mutex> lock(_sessionMutex);
auto &session = _sessions[hdl];
bool isIdentified = session.IsIdentified();
uint64_t connectedAt = session.ConnectedAt();
uint64_t incomingMessages = session.IncomingMessages();
uint64_t outgoingMessages = session.OutgoingMessages();
std::string remoteAddress = session.RemoteAddress();
_sessions.erase(hdl);
lock.unlock();
// Build SessionState object for signal
WebSocketSessionState state;
state.remoteAddress = remoteAddress;
state.connectedAt = connectedAt;
state.incomingMessages = incomingMessages;
state.outgoingMessages = outgoingMessages;
// Emit signals
emit ClientDisconnected(state, conn->get_local_close_code());
if (isIdentified)
emit IdentifiedClientDisconnected(state, conn->get_local_close_code());
}
void WebSocketServer::onMessage(websocketpp::connection_hdl hdl, websocketpp::server<websocketpp::config::asio>::message_ptr message)
{
auto opcode = message->get_opcode();
std::string payload = message->get_payload();
QtConcurrent::run(&_threadPool, [=]() {
std::unique_lock<std::mutex> lock(_sessionMutex);
auto &session = _sessions[hdl];
lock.unlock();
session.IncrementIncomingMessages();
json incomingMessage;
// Check for invalid opcode and decode
websocketpp::lib::error_code errorCode;
uint8_t sessionEncoding = session.Encoding();
if (sessionEncoding == WebSocketEncoding::Json) {
if (opcode != websocketpp::frame::opcode::text) {
if (!session.IgnoreInvalidMessages())
_server.close(hdl, WebSocketCloseCode::MessageDecodeError, "Your session encoding is set to Json, but a binary message was received.", errorCode);
return;
}
try {
incomingMessage = json::parse(payload);
} catch (json::parse_error& e) {
if (!session.IgnoreInvalidMessages())
_server.close(hdl, WebSocketCloseCode::MessageDecodeError, std::string("Unable to decode Json: ") + e.what(), errorCode);
return;
}
} else if (sessionEncoding == WebSocketEncoding::MsgPack) {
if (opcode != websocketpp::frame::opcode::binary) {
if (!session.IgnoreInvalidMessages())
_server.close(hdl, WebSocketCloseCode::MessageDecodeError, "Your session encoding is set to MsgPack, but a text message was received.", errorCode);
return;
}
try {
incomingMessage = json::from_msgpack(payload);
} catch (json::parse_error& e) {
if (!session.IgnoreInvalidMessages())
_server.close(hdl, WebSocketCloseCode::MessageDecodeError, std::string("Unable to decode MsgPack: ") + e.what(), errorCode);
return;
}
}
if (_debugEnabled)
blog(LOG_INFO, "[WebSocketServer::onMessage] Incoming message (decoded):\n%s", incomingMessage.dump(2).c_str());
WebSocketProtocol::ProcessResult ret = WebSocketProtocol::ProcessMessage(hdl, &session, incomingMessage);
if (ret.closeCode != WebSocketCloseCode::DontClose) {
websocketpp::lib::error_code errorCode;
_server.close(hdl, ret.closeCode, ret.closeReason, errorCode);
return;
}
if (!ret.result.is_null()) {
websocketpp::lib::error_code errorCode;
if (sessionEncoding == WebSocketEncoding::Json) {
std::string helloMessageJson = ret.result.dump();
_server.send(hdl, helloMessageJson, websocketpp::frame::opcode::text, errorCode);
} else if (sessionEncoding == WebSocketEncoding::MsgPack) {
auto msgPackData = json::to_msgpack(ret.result);
std::string messageMsgPack(msgPackData.begin(), msgPackData.end());
_server.send(hdl, messageMsgPack, websocketpp::frame::opcode::binary, errorCode);
}
session.IncrementOutgoingMessages();
if (_debugEnabled)
blog(LOG_INFO, "[WebSocketServer::onMessage] Outgoing message:\n%s", ret.result.dump(2).c_str());
if (errorCode)
blog(LOG_WARNING, "[WebSocketServer::onMessage] Sending message to client failed: %s", errorCode.message().c_str());
}
});
}

268
src/WebSocketSession.cpp
View File

@ -1,134 +1,134 @@
#include "WebSocketSession.h"
#include "plugin-macros.generated.h"
WebSocketSession::WebSocketSession() :
_remoteAddress(""),
_connectedAt(0),
_incomingMessages(0),
_outgoingMessages(0),
_encoding(0),
_challenge(""),
_rpcVersion(OBS_WEBSOCKET_RPC_VERSION),
_isIdentified(false),
_ignoreInvalidMessages(false),
_ignoreNonFatalRequestChecks(false),
_eventSubscriptions(0)
{
}
std::string WebSocketSession::RemoteAddress()
{
std::lock_guard<std::mutex> lock(_remoteAddressMutex);
std::string ret(_remoteAddress);
return ret;
}
void WebSocketSession::SetRemoteAddress(std::string address)
{
std::lock_guard<std::mutex> lock(_remoteAddressMutex);
_remoteAddress = address;
}
uint64_t WebSocketSession::ConnectedAt()
{
return _connectedAt.load();
}
void WebSocketSession::SetConnectedAt(uint64_t at)
{
_connectedAt.store(at);
}
uint64_t WebSocketSession::IncomingMessages()
{
return _incomingMessages.load();
}
void WebSocketSession::IncrementIncomingMessages()
{
_incomingMessages++;
}
uint64_t WebSocketSession::OutgoingMessages()
{
return _outgoingMessages.load();
}
void WebSocketSession::IncrementOutgoingMessages()
{
_outgoingMessages++;
}
uint8_t WebSocketSession::Encoding()
{
return _encoding.load();
}
void WebSocketSession::SetEncoding(uint8_t encoding)
{
_encoding.store(encoding);
}
std::string WebSocketSession::Challenge()
{
std::lock_guard<std::mutex> lock(_challengeMutex);
std::string ret(_challenge);
return ret;
}
void WebSocketSession::SetChallenge(std::string challengeString)
{
std::lock_guard<std::mutex> lock(_challengeMutex);
_challenge = challengeString;
}
uint8_t WebSocketSession::RpcVersion()
{
return _rpcVersion.load();
}
void WebSocketSession::SetRpcVersion(uint8_t version)
{
_rpcVersion.store(version);
}
bool WebSocketSession::IsIdentified()
{
return _isIdentified.load();
}
void WebSocketSession::SetIsIdentified(bool identified)
{
_isIdentified.store(identified);
}
bool WebSocketSession::IgnoreInvalidMessages()
{
return _ignoreInvalidMessages.load();
}
void WebSocketSession::SetIgnoreInvalidMessages(bool ignore)
{
_ignoreInvalidMessages.store(ignore);
}
bool WebSocketSession::IgnoreNonFatalRequestChecks()
{
return _ignoreNonFatalRequestChecks.load();
}
void WebSocketSession::SetIgnoreNonFatalRequestChecks(bool ignore)
{
_ignoreNonFatalRequestChecks.store(ignore);
}
uint64_t WebSocketSession::EventSubscriptions()
{
return _eventSubscriptions.load();
}
void WebSocketSession::SetEventSubscriptions(uint64_t subscriptions)
{
_eventSubscriptions.store(subscriptions);
}
#include "WebSocketSession.h"
#include "plugin-macros.generated.h"
WebSocketSession::WebSocketSession() :
_remoteAddress(""),
_connectedAt(0),
_incomingMessages(0),
_outgoingMessages(0),
_encoding(0),
_challenge(""),
_rpcVersion(OBS_WEBSOCKET_RPC_VERSION),
_isIdentified(false),
_ignoreInvalidMessages(false),
_ignoreNonFatalRequestChecks(false),
_eventSubscriptions(0)
{
}
std::string WebSocketSession::RemoteAddress()
{
std::lock_guard<std::mutex> lock(_remoteAddressMutex);
std::string ret(_remoteAddress);
return ret;
}
void WebSocketSession::SetRemoteAddress(std::string address)
{
std::lock_guard<std::mutex> lock(_remoteAddressMutex);
_remoteAddress = address;
}
uint64_t WebSocketSession::ConnectedAt()
{
return _connectedAt.load();
}
void WebSocketSession::SetConnectedAt(uint64_t at)
{
_connectedAt.store(at);
}
uint64_t WebSocketSession::IncomingMessages()
{
return _incomingMessages.load();
}
void WebSocketSession::IncrementIncomingMessages()
{
_incomingMessages++;
}
uint64_t WebSocketSession::OutgoingMessages()
{
return _outgoingMessages.load();
}
void WebSocketSession::IncrementOutgoingMessages()
{
_outgoingMessages++;
}
uint8_t WebSocketSession::Encoding()
{
return _encoding.load();
}
void WebSocketSession::SetEncoding(uint8_t encoding)
{
_encoding.store(encoding);
}
std::string WebSocketSession::Challenge()
{
std::lock_guard<std::mutex> lock(_challengeMutex);
std::string ret(_challenge);
return ret;
}
void WebSocketSession::SetChallenge(std::string challengeString)
{
std::lock_guard<std::mutex> lock(_challengeMutex);
_challenge = challengeString;
}
uint8_t WebSocketSession::RpcVersion()
{
return _rpcVersion.load();
}
void WebSocketSession::SetRpcVersion(uint8_t version)
{
_rpcVersion.store(version);
}
bool WebSocketSession::IsIdentified()
{
return _isIdentified.load();
}
void WebSocketSession::SetIsIdentified(bool identified)
{
_isIdentified.store(identified);
}
bool WebSocketSession::IgnoreInvalidMessages()
{
return _ignoreInvalidMessages.load();
}
void WebSocketSession::SetIgnoreInvalidMessages(bool ignore)
{
_ignoreInvalidMessages.store(ignore);
}
bool WebSocketSession::IgnoreNonFatalRequestChecks()
{
return _ignoreNonFatalRequestChecks.load();
}
void WebSocketSession::SetIgnoreNonFatalRequestChecks(bool ignore)
{
_ignoreNonFatalRequestChecks.store(ignore);
}
uint64_t WebSocketSession::EventSubscriptions()
{
return _eventSubscriptions.load();
}
void WebSocketSession::SetEventSubscriptions(uint64_t subscriptions)
{
_eventSubscriptions.store(subscriptions);
}

118
src/WebSocketSession.h
View File

@ -1,59 +1,59 @@
#pragma once
#include <mutex>
#include <string>
#include <atomic>
class WebSocketSession
{
public:
WebSocketSession();
std::string RemoteAddress();
void SetRemoteAddress(std::string address);
uint64_t ConnectedAt();
void SetConnectedAt(uint64_t at);
uint64_t IncomingMessages();
void IncrementIncomingMessages();
uint64_t OutgoingMessages();
void IncrementOutgoingMessages();
uint8_t Encoding();
void SetEncoding(uint8_t encoding);
std::string Challenge();
void SetChallenge(std::string challenge);
uint8_t RpcVersion();
void SetRpcVersion(uint8_t version);
bool IsIdentified();
void SetIsIdentified(bool identified);
bool IgnoreInvalidMessages();
void SetIgnoreInvalidMessages(bool ignore);
bool IgnoreNonFatalRequestChecks();
void SetIgnoreNonFatalRequestChecks(bool ignore);
uint64_t EventSubscriptions();
void SetEventSubscriptions(uint64_t subscriptions);
private:
std::mutex _remoteAddressMutex;
std::string _remoteAddress;
std::atomic<uint64_t> _connectedAt;
std::atomic<uint64_t> _incomingMessages;
std::atomic<uint64_t> _outgoingMessages;
std::atomic<uint8_t> _encoding;
std::mutex _challengeMutex;
std::string _challenge;
std::atomic<uint8_t> _rpcVersion;
std::atomic<bool> _isIdentified;
std::atomic<bool> _ignoreInvalidMessages;
std::atomic<bool> _ignoreNonFatalRequestChecks;
std::atomic<uint64_t> _eventSubscriptions;
};
#pragma once
#include <mutex>
#include <string>
#include <atomic>
class WebSocketSession
{
public:
WebSocketSession();
std::string RemoteAddress();
void SetRemoteAddress(std::string address);
uint64_t ConnectedAt();
void SetConnectedAt(uint64_t at);
uint64_t IncomingMessages();
void IncrementIncomingMessages();
uint64_t OutgoingMessages();
void IncrementOutgoingMessages();
uint8_t Encoding();
void SetEncoding(uint8_t encoding);
std::string Challenge();
void SetChallenge(std::string challenge);
uint8_t RpcVersion();
void SetRpcVersion(uint8_t version);
bool IsIdentified();
void SetIsIdentified(bool identified);
bool IgnoreInvalidMessages();
void SetIgnoreInvalidMessages(bool ignore);
bool IgnoreNonFatalRequestChecks();
void SetIgnoreNonFatalRequestChecks(bool ignore);
uint64_t EventSubscriptions();
void SetEventSubscriptions(uint64_t subscriptions);
private:
std::mutex _remoteAddressMutex;
std::string _remoteAddress;
std::atomic<uint64_t> _connectedAt;
std::atomic<uint64_t> _incomingMessages;
std::atomic<uint64_t> _outgoingMessages;
std::atomic<uint8_t> _encoding;
std::mutex _challengeMutex;
std::string _challenge;
std::atomic<uint8_t> _rpcVersion;
std::atomic<bool> _isIdentified;
std::atomic<bool> _ignoreInvalidMessages;
std::atomic<bool> _ignoreNonFatalRequestChecks;
std::atomic<uint64_t> _eventSubscriptions;
};

6
src/requesthandler/RequestHandler.cpp
View File

@ -1,3 +1,3 @@
#include "RequestHandler.h"
#include "../plugin-macros.generated.h"
#include "RequestHandler.h"
#include "../plugin-macros.generated.h"

6
src/requesthandler/rpc/Request.cpp
View File

@ -1,3 +1,3 @@
#include "Request.h"
#include "../../plugin-macros.generated.h"
#include "Request.h"
#include "../../plugin-macros.generated.h"

6
src/requesthandler/rpc/Request.h
View File

@ -1,3 +1,3 @@
#pragma once
#include <nlohmann/json.hpp>
#pragma once
#include <nlohmann/json.hpp>

6
src/requesthandler/rpc/RequestResult.cpp
View File

@ -1,3 +1,3 @@
#include "RequestResult.h"
#include "../../plugin-macros.generated.h"
#include "RequestResult.h"
#include "../../plugin-macros.generated.h"

6
src/requesthandler/rpc/RequestResult.h
View File

@ -1,3 +1,3 @@
#pragma once
#include <nlohmann/json.hpp>
#pragma once
#include <nlohmann/json.hpp>

2
src/utils/Platform.cpp
View File

@ -54,4 +54,4 @@ QString Utils::Platform::GetCommandLineArgument(QString arg)
return "";
return parser.value(cmdlineOption);
}
}