Skip to content

Add IP address logging in relay components #187

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
codemaster1104 wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Add IP address logging in relay components #187

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion TODO
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,6 @@ rate limits (maybe not needed now that we have plugins?)
event writes per second per ip
max connections per ip (nginx?)
max bandwidth up/down (nginx?)
log IP address in sendNoticeError and elsewhere where it makes sense
? events that contain IP/pubkey/etc block-lists in their contents
? limit on total number of events from a DBScan, not just per filter
? time limit on DBScan
Expand Down
26 changes: 13 additions & 13 deletions src/apps/relay/RelayIngester.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ void RelayServer::runIngester(ThreadPool<MsgIngester>::Thread &thr) {
if (msg->payload.starts_with('[')) {
auto payload = tao::json::from_string(msg->payload);

if (cfg().relay__logging__dumpInAll) LI << "[" << msg->connId << "] dumpInAll: " << msg->payload;
if (cfg().relay__logging__dumpInAll) LI << "[" << msg->connId << " " << renderIP(msg->ipAddr) << "] dumpInAll: " << msg->payload;

auto &arr = jsonGetArray(payload, "message is not an array");
if (arr.size() < 2) throw herr("too few array elements");
Expand All @@ -27,18 +27,18 @@ void RelayServer::runIngester(ThreadPool<MsgIngester>::Thread &thr) {

if (cmd == "EVENT") {
PROM_INC_CLIENT_MSG(cmd);
if (cfg().relay__logging__dumpInEvents) LI << "[" << msg->connId << "] dumpInEvent: " << msg->payload;
if (cfg().relay__logging__dumpInEvents) LI << "[" << msg->connId << " " << renderIP(msg->ipAddr) << "] dumpInEvent: " << msg->payload;

try {
ingesterProcessEvent(txn, rsctx, msg->connId, msg->ipAddr, arr[1], writerMsgs);
} catch (std::exception &e) {
sendOKResponse(msg->connId, arr[1].is_object() && arr[1].at("id").is_string() ? arr[1].at("id").get_string() : "?",
false, std::string("invalid: ") + e.what());
if (cfg().relay__logging__invalidEvents) LI << "Rejected invalid event: " << e.what();
if (cfg().relay__logging__invalidEvents) LI << "[" << msg->connId << " " << renderIP(msg->ipAddr) << "] Rejected invalid event: " << e.what();
}
} else if (cmd == "AUTH") {
PROM_INC_CLIENT_MSG(cmd);
if (cfg().relay__logging__dumpInAll) LI << "[" << msg->connId << "] dumpInAuth: " << msg->payload;
if (cfg().relay__logging__dumpInAll) LI << "[" << msg->connId << " " << renderIP(msg->ipAddr) << "] dumpInAuth: " << msg->payload;

try {
ingesterProcessAuth(rsctx, msg->connId, arr[1]);
Expand All @@ -47,7 +47,7 @@ void RelayServer::runIngester(ThreadPool<MsgIngester>::Thread &thr) {
}
} else if (cmd == "REQ" || cmd == "COUNT") {
PROM_INC_CLIENT_MSG(cmd);
if (cfg().relay__logging__dumpInReqs) LI << "[" << msg->connId << "] dumpInReq: " << msg->payload;
if (cfg().relay__logging__dumpInReqs) LI << "[" << msg->connId << " " << renderIP(msg->ipAddr) << "] dumpInReq: " << msg->payload;

std::string subIdStr;

Expand All @@ -59,7 +59,7 @@ void RelayServer::runIngester(ThreadPool<MsgIngester>::Thread &thr) {
}
} else if (cmd == "CLOSE") {
PROM_INC_CLIENT_MSG(cmd);
if (cfg().relay__logging__dumpInReqs) LI << "[" << msg->connId << "] dumpInReq: " << msg->payload;
if (cfg().relay__logging__dumpInReqs) LI << "[" << msg->connId << " " << renderIP(msg->ipAddr) << "] dumpInReq: " << msg->payload;

try {
ingesterProcessClose(txn, msg->connId, arr);
Expand Down Expand Up @@ -120,7 +120,7 @@ void RelayServer::ingesterProcessEvent(lmdb::txn &txn, RelayServerCtx &rsctx, ui
if (packed.kind() == 6 || packed.kind() == 16) {
if (origJson.at("content").get_string().find("[\"-\"]") != std::string::npos) {
auto idHex = to_hex(packed.id());
LI << "Repost embedded a protected event, blocking: " << idHex;
LI << "[" << connId << " " << renderIP(ipAddr) << "] Repost embedded a protected event, blocking: " << idHex;
sendOKResponse(connId, idHex, false, "blocked: reposts can't embed protected events");
return;
}
Expand All @@ -143,14 +143,14 @@ void RelayServer::ingesterProcessEvent(lmdb::txn &txn, RelayServerCtx &rsctx, ui
auto idHex = to_hex(packed.id());

if (!cfg().relay__auth__enabled) {
LI << "[" << connId << "] Protected event and auth disabled, rejecting: " << idHex;
LI << "[" << connId << " " << renderIP(ipAddr) << "] Protected event and auth disabled, rejecting: " << idHex;
sendOKResponse(connId, idHex, false, "blocked: event marked as protected");
return;
}

if (cfg().relay__auth__serviceUrl.empty()) {
// If we don't have a serviceUrl, just fail
LI << "[" << connId << "] Protected event and no serviceUrl configured, rejecting: " << idHex;
LI << "[" << connId << " " << renderIP(ipAddr) << "] Protected event and no serviceUrl configured, rejecting: " << idHex;
sendOKResponse(connId, idHex, false, "blocked: event marked as protected");
return;
}
Expand All @@ -161,7 +161,7 @@ void RelayServer::ingesterProcessEvent(lmdb::txn &txn, RelayServerCtx &rsctx, ui
auto challenge = rsctx.challengeGenerator.get();
rsctx.connIdToAuthStatus.emplace(connId, challenge);

LI << "[" << connId << "] Protected event, requesting AUTH: " << idHex;
LI << "[" << connId << " " << renderIP(ipAddr) << "] Protected event, requesting AUTH: " << idHex;
sendAuthChallenge(connId, challenge);
sendOKResponse(connId, idHex, false, "auth-required: event marked as protected");
return;
Expand All @@ -171,7 +171,7 @@ void RelayServer::ingesterProcessEvent(lmdb::txn &txn, RelayServerCtx &rsctx, ui

if (!as.isAuthed()) {
// not authenticated
LI << "[" << connId << "] Protected event, AUTH already requested: " << idHex;
LI << "[" << connId << " " << renderIP(ipAddr) << "] Protected event, AUTH already requested: " << idHex;
sendOKResponse(connId, idHex, false, "auth-required: event marked as protected");
return;
} else if (as.authed != packed.pubkey()) {
Expand All @@ -189,7 +189,7 @@ void RelayServer::ingesterProcessEvent(lmdb::txn &txn, RelayServerCtx &rsctx, ui
auto existing = lookupEventById(txn, packed.id());
if (existing) {
auto hexId = to_hex(packed.id());
LI << "[" << connId << "] Duplicate event, skipping: " << hexId;
LI << "[" << connId << " " << renderIP(ipAddr) << "] Duplicate event, skipping: " << hexId;
sendOKResponse(connId, hexId, true, "duplicate: have this event");
return;
}
Expand Down Expand Up @@ -270,7 +270,7 @@ void RelayServer::ingesterProcessAuth(RelayServerCtx &rsctx, uint64_t connId, co
// set the connection as authenticated with this pubkey
as.authed = packed.pubkey();

LI << "[" << connId << "] AUTHed as " << to_hex(packed.pubkey());
LI << "[" << connId << " " << getIpForConn(connId) << "] AUTHed as " << to_hex(packed.pubkey());
sendOKResponse(connId, to_hex(packed.id()), true, "successfully authenticated");
}

Expand Down
14 changes: 7 additions & 7 deletions src/apps/relay/RelayNegentropy.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,7 @@ void RelayServer::runNegentropy(ThreadPool<MsgNegentropy>::Thread &thr) {
Negentropy ne(storage, 500'000);
resp = ne.reconcile(msg);
} catch (std::exception &e) {
LI << "[" << connId << "] Error parsing negentropy message: " << e.what();
LI << "[" << connId << " " << getIpForConn(connId) << "] Error parsing negentropy message: " << e.what();

PROM_INC_RELAY_MSG("NEG-ERR");
sendToConn(connId, tao::json::to_string(tao::json::value::array({
Expand All @@ -112,7 +112,7 @@ void RelayServer::runNegentropy(ThreadPool<MsgNegentropy>::Thread &thr) {
return;
}

LI << "[" << connId << "] negentropy SEND session=" << subId.sv() << " bytesOut=" << resp.size();
LI << "[" << connId << " " << getIpForConn(connId) << "] negentropy SEND session=" << subId.sv() << " bytesOut=" << resp.size();

PROM_INC_RELAY_MSG("NEG-MSG");
sendToConn(connId, tao::json::to_string(tao::json::value::array({
Expand Down Expand Up @@ -144,11 +144,11 @@ void RelayServer::runNegentropy(ThreadPool<MsgNegentropy>::Thread &thr) {
auto *view = std::get_if<NegentropyViews::MemoryView>(userView);
if (!view) throw herr("bad variant, expected memory view");

LI << "[" << sub.connId << "] negentropy QUERY matched " << view->levIds.size() << " events in "
LI << "[" << sub.connId << " " << getIpForConn(sub.connId) << "] negentropy QUERY matched " << view->levIds.size() << " events in "
<< (hoytech::curr_time_us() - view->startTime) << "us";

if (view->levIds.size() > cfg().relay__negentropy__maxSyncEvents) {
LI << "[" << sub.connId << "] negentropy QUERY size exceeded " << cfg().relay__negentropy__maxSyncEvents;
LI << "[" << sub.connId << " " << getIpForConn(sub.connId) << "] negentropy QUERY size exceeded " << cfg().relay__negentropy__maxSyncEvents;

PROM_INC_RELAY_MSG("NEG-ERR");
sendToConn(sub.connId, tao::json::to_string(tao::json::value::array({
Expand Down Expand Up @@ -204,7 +204,7 @@ void RelayServer::runNegentropy(ThreadPool<MsgNegentropy>::Thread &thr) {
return true;
});

LI << "[" << connId << "] negentropy NEW session=" << subId.sv() << " bytesIn=" << msg->negPayload.size() << " tree=" << (treeId ? std::to_string(*treeId) : "NONE");
LI << "[" << connId << " " << getIpForConn(connId) << "] negentropy NEW session=" << subId.sv() << " bytesIn=" << msg->negPayload.size() << " tree=" << (treeId ? std::to_string(*treeId) : "NONE");

if (treeId) {
negentropy::storage::BTreeLMDB storage(txn, negentropyDbi, *treeId);
Expand Down Expand Up @@ -242,7 +242,7 @@ void RelayServer::runNegentropy(ThreadPool<MsgNegentropy>::Thread &thr) {
continue;
}

LI << "[" << msg->connId << "] negentropy RECV session=" << msg->subId.sv() << " bytesIn=" << msg->negPayload.size();
LI << "[" << msg->connId << " " << getIpForConn(msg->connId) << "] negentropy RECV session=" << msg->subId.sv() << " bytesIn=" << msg->negPayload.size();

if (auto *view = std::get_if<NegentropyViews::MemoryView>(userView)) {
if (!view->storageVector.sealed) {
Expand All @@ -259,7 +259,7 @@ void RelayServer::runNegentropy(ThreadPool<MsgNegentropy>::Thread &thr) {
handleReconcile(msg->connId, msg->subId, subStorage, msg->negPayload);
}
} else if (auto msg = std::get_if<MsgNegentropy::NegClose>(&newMsg.msg)) {
LI << "[" << msg->connId << "] negentropy CLOSE session=" << msg->subId.sv();
LI << "[" << msg->connId << " " << getIpForConn(msg->connId) << "] negentropy CLOSE session=" << msg->subId.sv();

queries.removeSub(msg->connId, msg->subId);
views.removeView(msg->connId, msg->subId);
Expand Down
17 changes: 15 additions & 2 deletions src/apps/relay/RelayServer.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
#include <iostream>
#include <memory>
#include <algorithm>
#include <shared_mutex>

#include <hoytech/time.h>
#include <hoytech/hex.h>
Expand Down Expand Up @@ -213,6 +214,18 @@ struct RelayServer {

void runSignalHandler();

// Connection tracking

mutable std::shared_mutex connIdToIpMutex;
flat_hash_map<uint64_t, std::string> connIdToIp;

std::string getIpForConn(uint64_t connId) const {
std::shared_lock lock(connIdToIpMutex);
auto it = connIdToIp.find(connId);
if (it != connIdToIp.end()) return renderIP(it->second);
return "unknown";
}

// Utils (can be called by any thread)

void sendToConn(uint64_t connId, std::string &&payload) {
Expand Down Expand Up @@ -248,15 +261,15 @@ struct RelayServer {

void sendNoticeError(uint64_t connId, std::string &&payload) {
PROM_INC_RELAY_MSG("NOTICE");
LI << "sending error to [" << connId << "]: " << payload;
LI << "sending error to [" << connId << " " << getIpForConn(connId) << "]: " << payload;
auto reply = tao::json::value::array({ "NOTICE", std::string("ERROR: ") + payload });
tpWebsocket.dispatch(0, MsgWebsocket{MsgWebsocket::Send{connId, std::move(tao::json::to_string(reply))}});
hubTrigger->send();
}

void sendClosedError(uint64_t connId, const std::string &subId, std::string &&payload) {
PROM_INC_RELAY_MSG("CLOSED");
LI << "sending closed to [" << connId << "]: " << payload;
LI << "sending closed to [" << connId << " " << getIpForConn(connId) << "]: " << payload;
auto reply = tao::json::value::array({ "CLOSED", subId, std::string("ERROR: ") + payload });
tpWebsocket.dispatch(0, MsgWebsocket{MsgWebsocket::Send{connId, std::move(tao::json::to_string(reply))}});
hubTrigger->send();
Expand Down
10 changes: 10 additions & 0 deletions src/apps/relay/RelayWebsocket.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -267,6 +267,11 @@ void RelayServer::runWebsocket(ThreadPool<MsgWebsocket>::Thread &thr) {

if (c->ipAddr.size() == 0) c->ipAddr = ws->getAddressBytes();

{
std::unique_lock lock(connIdToIpMutex);
connIdToIp[connId] = c->ipAddr;
}

ws->setUserData((void*)c);
connIdToConnection.emplace(connId, c);

Expand Down Expand Up @@ -300,6 +305,11 @@ void RelayServer::runWebsocket(ThreadPool<MsgWebsocket>::Thread &thr) {

tpIngester.dispatch(connId, MsgIngester{MsgIngester::CloseConn{connId}});

{
std::unique_lock lock(connIdToIpMutex);
connIdToIp.erase(connId);
}

connIdToConnection.erase(connId);
delete c;

Expand Down
10 changes: 5 additions & 5 deletions src/apps/relay/RelayWriter.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ void RelayServer::runWriter(ThreadPool<MsgWriter>::Thread &thr) {
PackedEventView packed(msg->packedStr);
auto eventIdHex = to_hex(packed.id());

if (okMsg.size()) LI << "[" << msg->connId << "] write policy blocked event " << eventIdHex << ": " << okMsg;
if (okMsg.size()) LI << "[" << msg->connId << " " << renderIP(msg->ipAddr) << "] write policy blocked event " << eventIdHex << ": " << okMsg;

sendOKResponse(msg->connId, eventIdHex, res == PluginEventSifterResult::ShadowReject, okMsg);
}
Expand Down Expand Up @@ -89,8 +89,10 @@ void RelayServer::runWriter(ThreadPool<MsgWriter>::Thread &thr) {
std::string message;
bool written = false;

MsgWriter::AddEvent *addEventMsg = static_cast<MsgWriter::AddEvent*>(newEvent.userData);

if (newEvent.status == EventWriteStatus::Written) {
LI << "Inserted event. id=" << eventIdHex << " levId=" << newEvent.levId;
LI << "[" << addEventMsg->connId << " " << renderIP(addEventMsg->ipAddr) << "] Inserted event. id=" << eventIdHex << " levId=" << newEvent.levId;
written = true;
} else if (newEvent.status == EventWriteStatus::Duplicate) {
message = "duplicate: have this event";
Expand All @@ -102,11 +104,9 @@ void RelayServer::runWriter(ThreadPool<MsgWriter>::Thread &thr) {
}

if (newEvent.status != EventWriteStatus::Written) {
LI << "Rejected event. " << message << ", id=" << eventIdHex;
LI << "[" << addEventMsg->connId << " " << renderIP(addEventMsg->ipAddr) << "] Rejected event. " << message << ", id=" << eventIdHex;
}

MsgWriter::AddEvent *addEventMsg = static_cast<MsgWriter::AddEvent*>(newEvent.userData);

sendOKResponse(addEventMsg->connId, eventIdHex, written, message);
}
}
Expand Down