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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
|
#include "../include/sibs/BootstrapConnection.hpp"
#include "../include/Log.hpp"
#include <sibs/SafeDeserializer.hpp>
namespace sibs
{
BootstrapConnection::BootstrapConnection(const Ipv4 &bootstrapAddress)
{
connections.onRemoveDisconnectedPeer([this](std::shared_ptr<DirectConnectionPeer> peer)
{
std::lock_guard<std::mutex> lock(subscribedPeersMutex);
for(auto &topicUsers : subscribedPeers)
{
for(auto it = topicUsers.second.begin(); it != topicUsers.second.end(); )
{
if(peer->address == (*it)->address)
it = topicUsers.second.erase(it);
else
++it;
}
}
});
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(peer.get(), 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(nullptr, 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);
}
}
}
|
