1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
|
#include "../include/sibs/BootstrapConnection.hpp"
#include "../include/Log.hpp"
#include <sibs/SafeDeserializer.hpp>
namespace sibs
{
BootstrapConnection::BootstrapConnection(const Ipv4 &bootstrapAddress)
{
PubSubResult connectResult = PubSubResult::OK;
std::string connectResultStr;
bool connected = false;
connections.connectServer(bootstrapAddress, [this, &connectResult, &connectResultStr, &connected](std::shared_ptr<DirectConnectionPeer> peer, PubSubResult result, const std::string &resultStr)
{
serverPeer = peer;
connectResult = result;
connectResultStr = resultStr;
connected = true;
}, std::bind(&BootstrapConnection::receiveDataFromServer, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
while(!connected)
{
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
if(connectResult != PubSubResult::OK)
{
std::string errMsg = "Failed to connect to bootstrap node, error: ";
errMsg += connectResultStr;
throw BootstrapConnectionException(errMsg);
}
}
// TODO: This is vulnerable against MitM attack, replace with asymmetric cryptography, get data signed with server private key and verify against known server public key
void BootstrapConnection::receiveDataFromServer(std::shared_ptr<DirectConnectionPeer> peer, const void *data, const usize size)
{
Log::debug("BootstrapConnection: Received subscriber(s) from bootstrap node");
sibs::SafeDeserializer deserializer((const u8*)data, size);
PubsubKey pubsubKey;
deserializer.extract(pubsubKey.data.data(), PUBSUB_KEY_LENGTH);
listenerCallbackFuncMutex.lock();
auto listenerFuncIt = listenCallbackFuncs.find(pubsubKey);
if(listenerFuncIt == listenCallbackFuncs.end())
{
Log::debug("BoostrapConnection: No listener found for key XXX, ignoring...");
listenerCallbackFuncMutex.unlock();
return;
}
auto listenerCallbackFunc = listenerFuncIt->second;
listenerCallbackFuncMutex.unlock();
while(!deserializer.empty())
{
sa_family_t addressFamily = deserializer.extract<u32>();
if(addressFamily == AF_INET)
{
in_addr_t ipv4Address = deserializer.extract<u32>();
u16 port = deserializer.extract<u16>();
Ipv4 newPeerAddress;
newPeerAddress.address.sin_family = addressFamily;
newPeerAddress.address.sin_addr.s_addr = ipv4Address;
newPeerAddress.address.sin_port = port;
memset(newPeerAddress.address.sin_zero, 0, sizeof(newPeerAddress.address.sin_zero));
connections.connect(newPeerAddress, [this, pubsubKey](std::shared_ptr<DirectConnectionPeer> newPeer, PubSubResult result, const std::string &resultStr)
{
if(result == PubSubResult::OK)
{
subscribedPeersMutex.lock();
subscribedPeers[pubsubKey].push_back(newPeer);
subscribedPeersMutex.unlock();
Log::debug("BootstrapConnection: Connected to peer (ip: %s, port: %d) given by bootstrap node", newPeer->address.getAddress().c_str(), newPeer->address.getPort());
}
else
Log::error("BootstrapConnection: Failed to connect to peer given by bootstrap node, error: %s", resultStr.c_str());
}, std::bind(&BootstrapConnection::receiveDataFromPeer, this, listenerCallbackFunc, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
}
else
{
Log::error("BootstrapConnection: Unknown address family: %d", addressFamily);
return;
}
}
}
void BootstrapConnection::receiveDataFromPeer(BoostrapConnectionListenCallbackFunc listenCallbackFunc, std::shared_ptr<DirectConnectionPeer> peer, const void *data, const usize size)
{
if(listenCallbackFunc)
listenCallbackFunc(data, size);
}
void BootstrapConnection::listen(const PubsubKey &pubsubKey, BoostrapConnectionListenCallbackFunc callbackFunc)
{
{
std::lock_guard<std::mutex> lock(listenerCallbackFuncMutex);
if(listenCallbackFuncs.find(pubsubKey) != listenCallbackFuncs.end())
throw PubsubKeyAlreadyListeningException("");
listenCallbackFuncs[pubsubKey] = callbackFunc;
}
connections.send(serverPeer, std::make_shared<std::vector<u8>>(pubsubKey.data.begin(), pubsubKey.data.end()),
[](PubSubResult result, const std::string &resultStr)
{
Log::debug("BootstrapConnection::listen: PubSubResult: %d, result string: %s", result, resultStr.c_str());
});
}
void BootstrapConnection::put(const PubsubKey &pubsubKey, std::shared_ptr<std::vector<u8>> data)
{
{
std::lock_guard<std::mutex> lock(listenerCallbackFuncMutex);
auto listenCallbackFuncIt = listenCallbackFuncs.find(pubsubKey);
if(listenCallbackFuncIt != listenCallbackFuncs.end() && listenCallbackFuncIt->second)
listenCallbackFuncIt->second(data->data(), data->size());
}
std::lock_guard<std::mutex> lock(subscribedPeersMutex);
auto peersIt = subscribedPeers.find(pubsubKey);
if(peersIt == subscribedPeers.end())
{
return;
}
for(auto &peer : peersIt->second)
{
connections.send(peer, data);
}
}
}
|
