Line data Source code
1 : /*
2 : * Copyright (C) 2004-2026 Savoir-faire Linux Inc.
3 : *
4 : * This program is free software: you can redistribute it and/or modify
5 : * it under the terms of the GNU General Public License as published by
6 : * the Free Software Foundation, either version 3 of the License, or
7 : * (at your option) any later version.
8 : *
9 : * This program is distributed in the hope that it will be useful,
10 : * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 : * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 : * GNU General Public License for more details.
13 : *
14 : * You should have received a copy of the GNU General Public License
15 : * along with this program. If not, see <https://www.gnu.org/licenses/>.
16 : */
17 :
18 : #include "swarm_manager.h"
19 : #include "jamidht/timestamp.h"
20 : #include <dhtnet/multiplexed_socket.h>
21 : #include <dhtnet/channel_utils.h>
22 : #include <opendht/thread_pool.h>
23 :
24 : namespace jami {
25 :
26 : using namespace swarm_protocol;
27 :
28 : static bool
29 11 : isNewerLease(const MobileLease& candidate, const MobileLease& reference)
30 : {
31 11 : return candidate.expires_at > reference.expires_at
32 11 : || (candidate.expires_at == reference.expires_at && candidate.issued_at > reference.issued_at);
33 : }
34 :
35 759 : SwarmManager::SwarmManager(const NodeId& id,
36 : bool isMobile,
37 : const std::mt19937_64& rand,
38 : ToConnectCb&& toConnectCb,
39 : std::string conversationId,
40 : MobileLeaseProvider mobileLeaseProvider,
41 : MobileLeaseIssuerValidator mobileLeaseIssuerValidator,
42 : CertificateProvider certificateProvider,
43 759 : CertificateFetcher certificateFetcher)
44 759 : : id_(id)
45 759 : , isMobile_(isMobile)
46 759 : , conversationId_(std::move(conversationId))
47 759 : , rd(rand)
48 759 : , mobileLeaseProvider_(std::move(mobileLeaseProvider))
49 759 : , mobileLeaseIssuerValidator_(std::move(mobileLeaseIssuerValidator))
50 759 : , certificateProvider_(std::move(certificateProvider))
51 759 : , certificateFetcher_(std::move(certificateFetcher))
52 2277 : , toConnectCb_(toConnectCb)
53 : {
54 759 : routing_table.setId(id);
55 759 : }
56 :
57 1518 : SwarmManager::~SwarmManager()
58 : {
59 759 : if (!isShutdown_)
60 244 : shutdown();
61 759 : }
62 :
63 : bool
64 2417 : SwarmManager::setKnownNodes(const std::vector<NodeId>& known_nodes)
65 : {
66 2417 : isShutdown_ = false;
67 2421 : std::vector<NodeId> newNodes;
68 : {
69 2421 : std::lock_guard lock(mutex);
70 5593 : for (const auto& nodeId : known_nodes) {
71 3175 : if (addKnownNode(nodeId)) {
72 884 : newNodes.emplace_back(nodeId);
73 : }
74 : }
75 2418 : }
76 :
77 2423 : if (newNodes.empty())
78 1706 : return false;
79 :
80 716 : dht::ThreadPool::io().run([w = weak(), newNodes = std::move(newNodes)] {
81 717 : auto shared = w.lock();
82 715 : if (!shared)
83 0 : return;
84 : // If we detect a new node which already got a TCP link
85 : // we can use it to speed-up the bootstrap (because opening
86 : // a new channel will be easy)
87 715 : std::set<NodeId> toConnect;
88 1597 : for (const auto& nodeId : newNodes) {
89 888 : if (shared->toConnectCb_ && shared->toConnectCb_(nodeId))
90 201 : toConnect.emplace(nodeId);
91 : }
92 712 : shared->maintainBuckets(toConnect);
93 715 : });
94 717 : return true;
95 2423 : }
96 :
97 : void
98 646 : SwarmManager::setMobileNodes(const std::vector<NodeId>& mobile_nodes)
99 : {
100 646 : bool changed = false;
101 : {
102 646 : std::lock_guard lock(mutex);
103 649 : const auto now = toSecondsSinceEpoch(std::chrono::system_clock::now());
104 649 : if (!conversationId_.empty() && now >= LEGACY_MOBILE_NODE_SUNSET)
105 0 : return;
106 769 : for (const auto& nodeId : mobile_nodes) {
107 120 : changed |= addMobileNodes(nodeId);
108 120 : if (!conversationId_.empty() && !mobileNodeLeases_.contains(nodeId))
109 1 : changed |= legacyMobileNodeExpiries_.try_emplace(nodeId, LEGACY_MOBILE_NODE_SUNSET).second;
110 : }
111 649 : scheduleMobileLeaseExpiryInternal();
112 648 : }
113 649 : if (changed)
114 22 : emitMobileNodesChanged();
115 : }
116 :
117 : void
118 16 : SwarmManager::setMobileNodes(const std::vector<MobileNodeInfo>& mobile_nodes, bool requireLease)
119 : {
120 16 : bool changed = false;
121 16 : size_t records = 0;
122 34 : for (const auto& mobile : mobile_nodes) {
123 18 : if (records++ == MAX_MOBILE_NODE_INFOS)
124 0 : break;
125 18 : changed |= setMobileNodeInfo(mobile, requireLease);
126 : }
127 16 : if (changed)
128 6 : emitMobileNodesChanged();
129 16 : }
130 :
131 : bool
132 26 : SwarmManager::setMobileNodeInfo(const MobileNodeInfo& mobile,
133 : bool requireLease,
134 : const std::shared_ptr<dhtnet::ChannelSocketInterface>& source)
135 : {
136 26 : if (mobile.id == id_)
137 0 : return false;
138 :
139 26 : if (!mobile.lease) {
140 6 : if (requireLease)
141 1 : return false;
142 : const auto now = static_cast<uint64_t>(
143 5 : std::chrono::duration_cast<std::chrono::seconds>(std::chrono::system_clock::now().time_since_epoch())
144 5 : .count());
145 5 : if (!conversationId_.empty() && now >= LEGACY_MOBILE_NODE_SUNSET)
146 0 : return false;
147 5 : std::lock_guard lock(mutex);
148 5 : bool changed = addMobileNodes(mobile.id);
149 5 : if (!conversationId_.empty() && !mobileNodeLeases_.contains(mobile.id)) {
150 0 : changed |= legacyMobileNodeExpiries_.try_emplace(mobile.id, LEGACY_MOBILE_NODE_SUNSET).second;
151 0 : scheduleMobileLeaseExpiryInternal();
152 : }
153 5 : return changed;
154 5 : }
155 :
156 20 : const auto& lease = *mobile.lease;
157 20 : if (lease.device_id != mobile.id || !precheckLease(lease))
158 2 : return false;
159 :
160 : {
161 : // Gossip re-announces the same lease every round: skip the whole
162 : // resolution when what we already verified is at least as good.
163 18 : std::lock_guard lock(mutex);
164 18 : auto known = mobileNodeLeases_.find(mobile.id);
165 18 : if (known != mobileNodeLeases_.end() && !isNewerLease(lease, known->second))
166 1 : return false;
167 18 : }
168 :
169 : // The certificate is never gossiped: resolve it from the account certificate
170 : // store, which already holds every device we ever connected to (the swarm
171 : // channel pins its TLS peer certificate) and everything resolved before.
172 17 : if (certificateProvider_) {
173 12 : if (auto certificate = certificateProvider_(mobile.id)) {
174 6 : if (!verifyLease(*certificate, lease))
175 1 : return false;
176 5 : std::lock_guard lock(mutex);
177 5 : return commitLeaseInternal(lease);
178 17 : }
179 : }
180 :
181 11 : std::lock_guard lock(mutex);
182 11 : enqueuePendingLeaseInternal(lease, source);
183 11 : return false;
184 11 : }
185 :
186 : void
187 2403 : SwarmManager::addChannel(const std::shared_ptr<dhtnet::ChannelSocketInterface>& channel)
188 : {
189 : // JAMI_WARNING("[SwarmManager {}] addChannel! with {}", fmt::ptr(this), channel->deviceId().to_view());
190 2403 : if (channel) {
191 2403 : auto emit = false;
192 2403 : auto added = false;
193 : {
194 2403 : std::lock_guard lock(mutex);
195 2403 : emit = routing_table.isEmpty();
196 2402 : auto bucket = routing_table.findBucket(channel->deviceId());
197 2402 : added = routing_table.addNode(channel, bucket);
198 2403 : }
199 2403 : if (added) {
200 1549 : std::error_code ec;
201 1549 : resetNodeExpiry(ec, channel, id_);
202 : }
203 2403 : receiveMessage(channel);
204 2400 : if (emit && onConnectionChanged_) {
205 : // If it's the first channel we add, we're now connected!
206 288 : JAMI_DEBUG("[SwarmManager {}] Bootstrap: Connected!", fmt::ptr(this));
207 288 : onConnectionChanged_(true);
208 : }
209 : }
210 2401 : }
211 :
212 : void
213 712 : SwarmManager::removeNode(const NodeId& nodeId)
214 : {
215 712 : std::unique_lock lk(mutex);
216 713 : if (isConnectedWith(nodeId)) {
217 636 : removeNodeInternal(nodeId);
218 636 : lk.unlock();
219 636 : maintainBuckets();
220 : }
221 713 : }
222 :
223 : void
224 236 : SwarmManager::changeMobility(const NodeId& nodeId, bool isMobile)
225 : {
226 : {
227 236 : std::lock_guard lock(mutex);
228 237 : auto bucket = routing_table.findBucket(nodeId);
229 239 : bucket->changeMobility(nodeId, isMobile);
230 238 : }
231 239 : emitMobileNodesChanged();
232 237 : }
233 :
234 : bool
235 1228 : SwarmManager::isConnectedWith(const NodeId& deviceId)
236 : {
237 1228 : return routing_table.hasNode(deviceId);
238 : }
239 :
240 : void
241 808 : SwarmManager::shutdown()
242 : {
243 808 : if (isShutdown_) {
244 19 : return;
245 : }
246 789 : isShutdown_ = true;
247 789 : std::lock_guard lock(mutex);
248 789 : mobileLeaseExpiryTimer_.cancel();
249 789 : for (auto& [peer, state] : outstandingCertRequests_)
250 0 : if (state.timer)
251 0 : state.timer->cancel();
252 789 : outstandingCertRequests_.clear();
253 789 : routing_table.shutdownAllNodes();
254 789 : }
255 :
256 : void
257 21 : SwarmManager::restart()
258 : {
259 21 : isShutdown_ = false;
260 21 : std::lock_guard lock(mutex);
261 21 : scheduleMobileLeaseExpiryInternal();
262 21 : }
263 :
264 : bool
265 3690 : SwarmManager::addKnownNode(const NodeId& nodeId)
266 : {
267 3690 : return routing_table.addKnownNode(nodeId);
268 : }
269 :
270 : bool
271 136 : SwarmManager::addMobileNodes(const NodeId& nodeId)
272 : {
273 136 : if (id_ != nodeId) {
274 134 : return routing_table.addMobileNode(nodeId);
275 : }
276 2 : return false;
277 : }
278 :
279 : bool
280 526 : SwarmManager::isMobileNodeCurrentInternal(const NodeId& nodeId, int64_t now) const
281 : {
282 526 : if (auto lease = mobileNodeLeases_.find(nodeId); lease != mobileNodeLeases_.end())
283 28 : return lease->second.expires_at > now;
284 500 : if (auto legacy = legacyMobileNodeExpiries_.find(nodeId); legacy != legacyMobileNodeExpiries_.end())
285 3 : return legacy->second > now;
286 496 : return conversationId_.empty();
287 : }
288 :
289 : bool
290 34 : SwarmManager::precheckLease(const MobileLease& lease) const
291 : {
292 34 : if (lease.format_version != 1 || lease.conversation_id != conversationId_ || lease.conversation_id.empty()
293 33 : || lease.conversation_id.size() > MAX_MOBILE_LEASE_IDENTIFIER_SIZE || lease.signature.empty()
294 33 : || lease.signature.size() > MAX_MOBILE_LEASE_SIGNATURE_SIZE || !lease.issuer_id
295 68 : || !mobileLeaseIssuerValidator_ || !mobileLeaseIssuerValidator_(lease.issuer_id))
296 3 : return false;
297 :
298 31 : constexpr auto MAX_CLOCK_SKEW = std::chrono::seconds(5 * 60);
299 31 : const auto now = std::chrono::system_clock::now();
300 31 : const auto issued = std::chrono::system_clock::time_point(std::chrono::seconds(lease.issued_at));
301 31 : const auto expires = std::chrono::system_clock::time_point(std::chrono::seconds(lease.expires_at));
302 62 : if (issued > now + MAX_CLOCK_SKEW || expires <= now || expires <= issued
303 62 : || expires - issued > MAX_MOBILE_LEASE_DURATION)
304 0 : return false;
305 31 : return true;
306 : }
307 :
308 : bool
309 13 : SwarmManager::verifyLease(const dht::crypto::Certificate& certificate, const MobileLease& lease) const
310 : {
311 : try {
312 26 : if (certificate.getLongId() != lease.device_id || !certificate.issuer
313 26 : || certificate.issuer->getId() != lease.issuer_id)
314 0 : return false;
315 13 : dht::crypto::TrustList trust;
316 13 : trust.add(*certificate.issuer);
317 13 : if (!trust.verify(certificate))
318 0 : return false;
319 13 : auto certificateExpiry = toSecondsSinceEpoch(certificate.getExpiration());
320 13 : if (lease.expires_at > certificateExpiry)
321 0 : return false;
322 13 : const auto payload = mobileLeasePayload(lease);
323 13 : return certificate.getPublicKey().checkSignature(payload, lease.signature);
324 13 : } catch (const std::exception& e) {
325 0 : JAMI_WARNING("Ignoring invalid mobile lease for {}: {}", lease.device_id, e.what());
326 0 : return false;
327 0 : }
328 : }
329 :
330 : bool
331 11 : SwarmManager::commitLeaseInternal(const MobileLease& lease)
332 : {
333 11 : auto pending = pendingMobileLeases_.find(lease.device_id);
334 11 : if (pending != pendingMobileLeases_.end()) {
335 6 : if (!isNewerLease(pending->second.lease, lease))
336 6 : pendingMobileLeases_.erase(pending);
337 : }
338 :
339 11 : bool changed = addMobileNodes(lease.device_id);
340 11 : auto current = mobileNodeLeases_.find(lease.device_id);
341 11 : if (current == mobileNodeLeases_.end() || isNewerLease(lease, current->second)) {
342 11 : mobileNodeLeases_.insert_or_assign(lease.device_id, lease);
343 11 : legacyMobileNodeExpiries_.erase(lease.device_id);
344 11 : changed = true;
345 : }
346 11 : scheduleMobileLeaseExpiryInternal();
347 11 : return changed;
348 : }
349 :
350 : void
351 11 : SwarmManager::enqueuePendingLeaseInternal(const MobileLease& lease,
352 : const std::shared_ptr<dhtnet::ChannelSocketInterface>& source)
353 : {
354 11 : const auto& nodeId = lease.device_id;
355 11 : auto pending = pendingMobileLeases_.find(nodeId);
356 11 : if (pending != pendingMobileLeases_.end()) {
357 1 : if (isNewerLease(lease, pending->second.lease))
358 0 : pending->second.lease = lease;
359 1 : if (source)
360 1 : pending->second.source = source;
361 : } else {
362 10 : if (pendingMobileLeases_.size() >= MAX_PENDING_MOBILE_LEASES) {
363 : // Evict the entry that would expire first: it is the least useful to keep resolving.
364 0 : auto oldest = std::min_element(pendingMobileLeases_.begin(),
365 : pendingMobileLeases_.end(),
366 0 : [](const auto& a, const auto& b) {
367 0 : return a.second.lease.expires_at < b.second.lease.expires_at;
368 : });
369 0 : if (oldest != pendingMobileLeases_.end() && oldest->second.lease.expires_at >= lease.expires_at)
370 0 : return;
371 0 : pendingMobileLeases_.erase(oldest);
372 : }
373 10 : pendingMobileLeases_.emplace(nodeId, PendingLease {lease, source});
374 : }
375 :
376 11 : if (certFetchInFlight_.count(nodeId))
377 0 : return;
378 :
379 11 : if (source) {
380 6 : certFetchInFlight_.emplace(nodeId);
381 6 : dht::ThreadPool::io().run([w = weak(), source, nodeId] {
382 6 : if (auto shared = w.lock())
383 18 : shared->requestCertificates(source, {nodeId});
384 6 : });
385 : } else {
386 5 : certFetchInFlight_.emplace(nodeId);
387 5 : dht::ThreadPool::io().run([w = weak(), nodeId] {
388 5 : if (auto shared = w.lock())
389 5 : shared->fetchCertificateFromDht(nodeId);
390 5 : });
391 : }
392 : }
393 :
394 : void
395 4 : SwarmManager::abandonLeaseInternal(const NodeId& nodeId)
396 : {
397 4 : certFetchInFlight_.erase(nodeId);
398 4 : pendingMobileLeases_.erase(nodeId);
399 4 : }
400 :
401 : void
402 6 : SwarmManager::requestCertificates(const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket,
403 : const std::vector<NodeId>& ids)
404 : {
405 6 : if (!socket || ids.empty() || isShutdown_) {
406 0 : std::lock_guard lock(mutex);
407 0 : for (const auto& id : ids)
408 0 : certFetchInFlight_.erase(id);
409 0 : return;
410 0 : }
411 6 : const auto peer = NodeId(socket->deviceId());
412 6 : CertRequest request;
413 : {
414 6 : std::lock_guard lock(mutex);
415 6 : auto& state = outstandingCertRequests_[peer];
416 6 : if (state.timer) {
417 : // One request in flight per peer: drop the resolution so that the
418 : // next gossip round asks again.
419 2 : for (const auto& id : ids)
420 1 : certFetchInFlight_.erase(id);
421 1 : return;
422 : }
423 10 : for (const auto& id : ids) {
424 5 : if (request.ids.size() >= MAX_CERT_REQUEST_IDS) {
425 0 : certFetchInFlight_.erase(id);
426 0 : continue;
427 : }
428 5 : request.ids.emplace_back(id);
429 : }
430 5 : if (request.ids.empty()) {
431 0 : outstandingCertRequests_.erase(peer);
432 0 : return;
433 : }
434 5 : state.ids.insert(request.ids.begin(), request.ids.end());
435 5 : state.timer = std::make_shared<asio::steady_timer>(*Manager::instance().ioContext());
436 5 : state.timer->expires_after(CERT_REQUEST_TIMEOUT);
437 5 : state.timer->async_wait([w = weak(), peer](const asio::error_code& ec) {
438 5 : if (ec == asio::error::operation_aborted)
439 4 : return;
440 1 : auto shared = w.lock();
441 1 : if (!shared)
442 0 : return;
443 1 : std::vector<NodeId> unanswered;
444 : {
445 1 : std::lock_guard lock(shared->mutex);
446 1 : auto it = shared->outstandingCertRequests_.find(peer);
447 1 : if (it == shared->outstandingCertRequests_.end())
448 0 : return;
449 1 : unanswered.assign(it->second.ids.begin(), it->second.ids.end());
450 1 : shared->outstandingCertRequests_.erase(it);
451 1 : }
452 : // The peer did not answer: fall back to the DHT.
453 2 : for (const auto& id : unanswered)
454 1 : shared->fetchCertificateFromDht(id);
455 1 : });
456 6 : }
457 :
458 5 : Message msg;
459 5 : msg.is_mobile = isMobile_;
460 5 : msg.cert_request = std::move(request);
461 :
462 5 : msgpack::sbuffer buffer;
463 5 : msgpack::packer<msgpack::sbuffer> pk(&buffer);
464 5 : pk.pack(msg);
465 :
466 5 : std::error_code ec;
467 5 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
468 5 : if (ec)
469 0 : JAMI_ERROR("{}", ec.message());
470 6 : }
471 :
472 : void
473 2 : SwarmManager::onCertRequest(const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket, const CertRequest& request)
474 : {
475 2 : if (!socket || request.ids.empty() || !certificateProvider_)
476 0 : return;
477 :
478 2 : std::vector<NodeId> toAnswer;
479 : {
480 2 : std::lock_guard lock(mutex);
481 5 : for (const auto& id : request.ids) {
482 3 : if (toAnswer.size() >= MAX_CERT_REQUEST_IDS)
483 0 : break;
484 : // A peer must not be able to use the swarm as a generic certificate
485 : // oracle: only serve certificates for devices we ourselves announced
486 : // as mobile in this conversation, plus our own when we are mobile.
487 3 : if (id != id_ && !mobileNodeLeases_.count(id))
488 2 : continue;
489 1 : toAnswer.emplace_back(id);
490 : }
491 2 : }
492 2 : if (toAnswer.empty())
493 1 : return;
494 :
495 1 : CertResponse response;
496 1 : size_t totalSize = 0;
497 2 : for (const auto& id : toAnswer) {
498 1 : auto certificate = certificateProvider_(id);
499 1 : if (!certificate)
500 0 : continue;
501 1 : auto packed = certificate->getPacked();
502 2 : if (packed.empty() || packed.size() > MAX_MOBILE_CERTIFICATE_SIZE
503 2 : || totalSize + packed.size() > MAX_MOBILE_CERTIFICATES_SIZE)
504 0 : continue;
505 1 : totalSize += packed.size();
506 1 : response.certificates.emplace_back(std::move(packed));
507 1 : }
508 1 : if (response.certificates.empty())
509 0 : return;
510 :
511 1 : Message msg;
512 1 : msg.is_mobile = isMobile_;
513 1 : msg.cert_response = std::move(response);
514 :
515 1 : msgpack::sbuffer buffer;
516 1 : msgpack::packer<msgpack::sbuffer> pk(&buffer);
517 1 : pk.pack(msg);
518 :
519 1 : std::error_code ec;
520 1 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
521 1 : if (ec)
522 0 : JAMI_ERROR("{}", ec.message());
523 2 : }
524 :
525 : void
526 5 : SwarmManager::onCertResponse(const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket, const CertResponse& response)
527 : {
528 5 : if (!socket)
529 0 : return;
530 5 : const auto peer = NodeId(socket->deviceId());
531 :
532 5 : std::set<NodeId> requested;
533 : {
534 5 : std::lock_guard lock(mutex);
535 5 : auto it = outstandingCertRequests_.find(peer);
536 5 : if (it == outstandingCertRequests_.end())
537 1 : return; // Unsolicited.
538 4 : requested = std::move(it->second.ids);
539 4 : if (it->second.timer)
540 4 : it->second.timer->cancel();
541 4 : outstandingCertRequests_.erase(it);
542 5 : }
543 :
544 4 : size_t totalSize = 0;
545 8 : for (const auto& packed : response.certificates) {
546 8 : if (packed.empty() || packed.size() > MAX_MOBILE_CERTIFICATE_SIZE
547 8 : || totalSize + packed.size() > MAX_MOBILE_CERTIFICATES_SIZE)
548 0 : break;
549 4 : totalSize += packed.size();
550 : try {
551 4 : auto certificate = std::make_shared<dht::crypto::Certificate>(packed);
552 4 : const auto nodeId = certificate->getLongId();
553 4 : if (!requested.erase(nodeId))
554 1 : continue; // Not something we asked for.
555 3 : onCertificateResolved(nodeId, certificate);
556 4 : } catch (const std::exception& e) {
557 0 : JAMI_WARNING("Ignoring invalid certificate from {}: {}", peer, e.what());
558 0 : }
559 : }
560 :
561 : // Whatever the peer could not provide is worth one DHT lookup.
562 5 : for (const auto& nodeId : requested)
563 1 : fetchCertificateFromDht(nodeId);
564 5 : }
565 :
566 : void
567 7 : SwarmManager::fetchCertificateFromDht(const NodeId& nodeId)
568 : {
569 7 : if (isShutdown_)
570 0 : return;
571 7 : if (!certificateFetcher_) {
572 2 : std::lock_guard lock(mutex);
573 2 : abandonLeaseInternal(nodeId);
574 2 : return;
575 2 : }
576 5 : certificateFetcher_(nodeId, [w = weak(), nodeId](const std::shared_ptr<dht::crypto::Certificate>& certificate) {
577 5 : auto shared = w.lock();
578 5 : if (!shared)
579 0 : return;
580 5 : if (certificate && certificate->getLongId() == nodeId)
581 4 : shared->onCertificateResolved(nodeId, certificate);
582 : else {
583 1 : std::lock_guard lock(shared->mutex);
584 1 : shared->abandonLeaseInternal(nodeId);
585 1 : }
586 5 : });
587 : }
588 :
589 : void
590 7 : SwarmManager::onCertificateResolved(const NodeId& nodeId, const std::shared_ptr<dht::crypto::Certificate>& certificate)
591 : {
592 7 : if (!certificate)
593 0 : return;
594 :
595 7 : std::optional<MobileLease> lease;
596 : {
597 7 : std::lock_guard lock(mutex);
598 7 : certFetchInFlight_.erase(nodeId);
599 7 : auto pending = pendingMobileLeases_.find(nodeId);
600 7 : if (pending == pendingMobileLeases_.end())
601 0 : return;
602 7 : lease = pending->second.lease;
603 7 : }
604 :
605 : // Re-run the cheap checks: the lease may have expired while we were resolving.
606 7 : if (!precheckLease(*lease) || !verifyLease(*certificate, *lease)) {
607 1 : std::lock_guard lock(mutex);
608 1 : auto pending = pendingMobileLeases_.find(nodeId);
609 1 : if (pending != pendingMobileLeases_.end() && !isNewerLease(pending->second.lease, *lease))
610 1 : abandonLeaseInternal(nodeId);
611 1 : return;
612 1 : }
613 :
614 6 : bool changed = false;
615 : {
616 6 : std::lock_guard lock(mutex);
617 6 : changed = commitLeaseInternal(*lease);
618 6 : }
619 6 : if (changed)
620 6 : emitMobileNodesChanged();
621 7 : }
622 :
623 : std::optional<MobileNodeInfo>
624 3039 : SwarmManager::localMobileNodeInfo()
625 : {
626 3039 : if (!isMobile_ || !mobileLeaseProvider_)
627 3037 : return std::nullopt;
628 :
629 2 : std::lock_guard renewalLock(mobileLeaseRenewalMtx_);
630 :
631 3 : auto renewalThresholdTime = toSecondsSinceEpoch(std::chrono::system_clock::now() + MOBILE_LEASE_RENEWAL_THRESHOLD);
632 : {
633 3 : std::lock_guard lock(mutex);
634 5 : if (localMobileNodeInfo_ && localMobileNodeInfo_->lease
635 5 : && localMobileNodeInfo_->lease->expires_at > renewalThresholdTime)
636 1 : return localMobileNodeInfo_;
637 3 : }
638 :
639 2 : auto renewed = mobileLeaseProvider_();
640 4 : if (!renewed || renewed->id != id_ || !renewed->lease || renewed->lease->device_id != id_
641 4 : || !precheckLease(*renewed->lease))
642 0 : return std::nullopt;
643 2 : std::lock_guard lock(mutex);
644 2 : localMobileNodeInfo_ = std::move(renewed);
645 2 : return localMobileNodeInfo_;
646 3 : }
647 :
648 : void
649 698 : SwarmManager::scheduleMobileLeaseExpiryInternal()
650 : {
651 698 : mobileLeaseExpiryTimer_.cancel();
652 696 : if ((mobileNodeLeases_.empty() && legacyMobileNodeExpiries_.empty()) || isShutdown_)
653 685 : return;
654 :
655 12 : auto nearestExpiry = std::numeric_limits<int64_t>::max();
656 24 : for (const auto& [node, lease] : mobileNodeLeases_)
657 12 : nearestExpiry = std::min(nearestExpiry, lease.expires_at);
658 13 : for (const auto& [node, expiry] : legacyMobileNodeExpiries_)
659 1 : nearestExpiry = std::min(nearestExpiry, expiry);
660 12 : auto expiryTime = timePointFromSeconds(nearestExpiry);
661 12 : const auto now = std::chrono::system_clock::now();
662 12 : constexpr auto MAX_TIMER_DELAY_SECONDS = std::chrono::minutes(1);
663 12 : const auto delay = std::min<std::chrono::system_clock::duration>(expiryTime > now ? expiryTime - now : std::chrono::seconds(0), MAX_TIMER_DELAY_SECONDS);
664 12 : mobileLeaseExpiryTimer_.expires_after(delay);
665 12 : mobileLeaseExpiryTimer_.async_wait([w = weak()](const asio::error_code& ec) {
666 12 : if (auto shared = w.lock())
667 12 : shared->expireMobileLeases(ec);
668 12 : });
669 : }
670 :
671 : void
672 4 : SwarmManager::expireMobileLeases(const asio::error_code& ec)
673 : {
674 4 : if (ec == asio::error::operation_aborted)
675 3 : return;
676 :
677 1 : bool changed = false;
678 : {
679 1 : std::lock_guard lock(mutex);
680 1 : auto now = toSecondsSinceEpoch(std::chrono::system_clock::now());
681 2 : for (auto it = mobileNodeLeases_.begin(); it != mobileNodeLeases_.end();) {
682 1 : if (it->second.expires_at > now) {
683 0 : ++it;
684 0 : continue;
685 : }
686 1 : const auto nodeId = it->first;
687 1 : it = mobileNodeLeases_.erase(it);
688 1 : auto legacy = legacyMobileNodeExpiries_.find(nodeId);
689 1 : if (legacy == legacyMobileNodeExpiries_.end() || legacy->second <= now) {
690 1 : routing_table.removeMobileNode(nodeId);
691 1 : routing_table.findBucket(nodeId)->changeMobility(nodeId, false);
692 : }
693 1 : changed = true;
694 : }
695 1 : for (auto it = legacyMobileNodeExpiries_.begin(); it != legacyMobileNodeExpiries_.end();) {
696 0 : if (it->second > now) {
697 0 : ++it;
698 0 : continue;
699 : }
700 0 : const auto nodeId = it->first;
701 0 : it = legacyMobileNodeExpiries_.erase(it);
702 0 : auto lease = mobileNodeLeases_.find(nodeId);
703 0 : if (lease == mobileNodeLeases_.end() || lease->second.expires_at <= now) {
704 0 : routing_table.removeMobileNode(nodeId);
705 0 : routing_table.findBucket(nodeId)->changeMobility(nodeId, false);
706 : }
707 0 : changed = true;
708 : }
709 1 : for (auto it = pendingMobileLeases_.begin(); it != pendingMobileLeases_.end();) {
710 0 : if (it->second.lease.expires_at > now)
711 0 : ++it;
712 : else
713 0 : it = pendingMobileLeases_.erase(it);
714 : }
715 1 : scheduleMobileLeaseExpiryInternal();
716 1 : }
717 1 : if (changed)
718 1 : emitMobileNodesChanged();
719 : }
720 :
721 : void
722 277 : SwarmManager::emitMobileNodesChanged()
723 : {
724 277 : std::lock_guard emissionLock(mobileNodesEmissionMtx_);
725 276 : auto mobileNodes = getKnownMobileNodes();
726 276 : auto mobileNodeInfos = getKnownMobileNodeInfos();
727 276 : OnMobileNodesChanged callback;
728 276 : OnMobileNodeInfosChanged infosCallback;
729 : {
730 276 : std::lock_guard callbackLock(onMobileNodesChangedMtx_);
731 276 : callback = onMobileNodesChanged_;
732 276 : infosCallback = onMobileNodeInfosChanged_;
733 275 : }
734 277 : if (callback)
735 19 : callback(mobileNodes);
736 277 : if (infosCallback)
737 1 : infosCallback(mobileNodeInfos);
738 276 : }
739 :
740 : void
741 1992 : SwarmManager::maintainBuckets(const std::set<NodeId>& toConnect)
742 : {
743 1992 : std::set<NodeId> nodes = toConnect;
744 1990 : std::unique_lock lock(mutex);
745 1987 : auto& buckets = routing_table.getBuckets();
746 6006 : for (auto it = buckets.begin(); it != buckets.end(); ++it) {
747 4008 : auto& bucket = *it;
748 4012 : bool myBucket = routing_table.contains(it, id_);
749 6046 : auto connecting_nodes = myBucket ? bucket.getConnectingNodesSize()
750 2028 : : bucket.getConnectingNodesSize() + bucket.getNodesSize();
751 4016 : if (connecting_nodes < Bucket::BUCKET_MAX_SIZE) {
752 2355 : auto nodesToTry = bucket.getKnownNodesRandom(Bucket::BUCKET_MAX_SIZE - connecting_nodes, rd);
753 3426 : for (auto& node : nodesToTry)
754 1069 : routing_table.addConnectingNode(node);
755 :
756 2350 : nodes.insert(nodesToTry.begin(), nodesToTry.end());
757 2357 : }
758 : }
759 1993 : lock.unlock();
760 3082 : for (const auto& node : nodes)
761 1091 : tryConnect(node);
762 1990 : }
763 :
764 : void
765 1547 : SwarmManager::sendRequest(const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket,
766 : const NodeId& nodeId,
767 : Query q,
768 : int numberNodes)
769 : {
770 1547 : auto selfMobileInfo = localMobileNodeInfo();
771 3096 : dht::ThreadPool::io().run(
772 3097 : [socket, isMobile = isMobile_, selfMobileInfo = std::move(selfMobileInfo), nodeId, q, numberNodes] {
773 1549 : msgpack::sbuffer buffer;
774 1549 : msgpack::packer<msgpack::sbuffer> pk(&buffer);
775 1548 : Message msg;
776 1548 : msg.is_mobile = isMobile;
777 1548 : msg.self_mobile_info = selfMobileInfo;
778 1548 : msg.request = Request {q, numberNodes, nodeId};
779 1547 : pk.pack(msg);
780 :
781 1547 : std::error_code ec;
782 1546 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
783 1549 : if (ec) {
784 2 : JAMI_ERROR("{}", ec.message());
785 : }
786 1548 : });
787 1549 : }
788 :
789 : void
790 1490 : SwarmManager::sendAnswer(const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket, const Message& msg_)
791 : {
792 1490 : if (msg_.request->q != Query::FIND)
793 0 : return;
794 :
795 1489 : auto selfMobileInfo = localMobileNodeInfo();
796 1490 : Message msg;
797 : {
798 1490 : std::lock_guard lock(mutex);
799 1491 : auto nodes = routing_table.closestNodes(msg_.request->nodeId, msg_.request->num);
800 1493 : auto bucket = routing_table.findBucket(msg_.request->nodeId);
801 1493 : const auto& m_nodes = bucket->getMobileNodes();
802 1495 : std::vector<NodeId> responseMobileNodes;
803 1495 : responseMobileNodes.reserve(m_nodes.size());
804 1493 : std::vector<MobileNodeInfo> mobileNodeInfos;
805 1493 : mobileNodeInfos.reserve(m_nodes.size());
806 1492 : const auto now = toSecondsSinceEpoch(std::chrono::system_clock::now());
807 1494 : for (const auto& node : m_nodes) {
808 1 : if (!isMobileNodeCurrentInternal(node, now))
809 0 : continue;
810 1 : responseMobileNodes.emplace_back(node);
811 1 : if (mobileNodeInfos.size() >= MAX_MOBILE_NODE_INFOS)
812 0 : continue;
813 1 : auto lease = mobileNodeLeases_.find(node);
814 1 : if (lease == mobileNodeLeases_.end()) {
815 1 : if (msg_.v >= 3)
816 1 : continue;
817 0 : mobileNodeInfos.emplace_back(MobileNodeInfo {node, std::nullopt});
818 : } else {
819 0 : mobileNodeInfos.emplace_back(MobileNodeInfo {node, lease->second});
820 : }
821 : }
822 1493 : Response toResponse {Query::FOUND, nodes, std::move(responseMobileNodes), std::move(mobileNodeInfos)};
823 :
824 1494 : msg.is_mobile = isMobile_;
825 1494 : msg.self_mobile_info = std::move(selfMobileInfo);
826 1492 : msg.response = std::move(toResponse);
827 1492 : }
828 :
829 1495 : msgpack::sbuffer buffer;
830 1496 : msgpack::packer<msgpack::sbuffer> pk(&buffer);
831 1496 : pk.pack(msg);
832 :
833 1493 : std::error_code ec;
834 1493 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
835 1496 : if (ec) {
836 3 : JAMI_ERROR("{}", ec.message());
837 2 : return;
838 : }
839 1499 : }
840 :
841 : void
842 2403 : SwarmManager::receiveMessage(const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket)
843 : {
844 4806 : socket->setOnRecv(dhtnet::buildMsgpackReader<Message>(
845 4804 : [w = weak(), wsocket = std::weak_ptr<dhtnet::ChannelSocketInterface>(socket)](Message&& msg) {
846 2985 : auto shared = w.lock();
847 2982 : auto socket = wsocket.lock();
848 2977 : if (!shared || !socket)
849 0 : return std::make_error_code(std::errc::operation_canceled);
850 :
851 2973 : auto validMobileAnnouncement = msg.v < 3 || shared->conversationId_.empty();
852 2969 : if (msg.self_mobile_info && msg.self_mobile_info->id == socket->deviceId()) {
853 : // The peer's own certificate is authenticated by the channel's TLS
854 : // handshake and pinned when the swarm channel was added, so this
855 : // resolves locally without any lookup.
856 5 : validMobileAnnouncement = msg.self_mobile_info->lease
857 10 : && msg.self_mobile_info->lease->device_id == msg.self_mobile_info->id
858 10 : && shared->precheckLease(*msg.self_mobile_info->lease);
859 5 : if (validMobileAnnouncement && shared->setMobileNodeInfo(*msg.self_mobile_info, true, socket))
860 1 : shared->emitMobileNodesChanged();
861 : }
862 2980 : if (msg.is_mobile && validMobileAnnouncement) {
863 233 : if (msg.v < 3 && !shared->conversationId_.empty())
864 3 : shared->setMobileNodes(std::vector<NodeId> {socket->deviceId()});
865 233 : shared->changeMobility(socket->deviceId(), true);
866 : }
867 :
868 2983 : if (msg.cert_request) {
869 2 : shared->onCertRequest(socket, *msg.cert_request);
870 2978 : } else if (msg.cert_response) {
871 5 : shared->onCertResponse(socket, *msg.cert_response);
872 2976 : } else if (msg.request) {
873 1490 : shared->sendAnswer(socket, msg);
874 :
875 1483 : } else if (msg.response) {
876 1487 : shared->setKnownNodes(msg.response->nodes);
877 1489 : const auto requireLease = msg.v >= 3 && !shared->conversationId_.empty();
878 1488 : bool changed = false;
879 1488 : size_t records = 0;
880 1492 : for (const auto& mobile : msg.response->mobile_node_infos) {
881 4 : if (records++ == MAX_MOBILE_NODE_INFOS)
882 0 : break;
883 4 : changed |= shared->setMobileNodeInfo(mobile, requireLease, socket);
884 : }
885 1486 : if (changed)
886 0 : shared->emitMobileNodesChanged();
887 1487 : const auto acceptLegacy = msg.v < 3 || shared->conversationId_.empty();
888 1487 : if (acceptLegacy)
889 614 : shared->setMobileNodes(msg.response->mobile_nodes);
890 : }
891 2997 : return std::error_code();
892 2996 : }));
893 :
894 2402 : socket->onShutdown([w = weak(), deviceId = socket->deviceId()](const std::error_code&) {
895 1330 : dht::ThreadPool::io().run([w, deviceId] {
896 1329 : auto shared = w.lock();
897 1321 : if (shared && !shared->isShutdown_) {
898 712 : shared->removeNode(deviceId);
899 : }
900 1329 : });
901 1335 : });
902 2400 : }
903 :
904 : void
905 1548 : SwarmManager::resetNodeExpiry(const asio::error_code& ec,
906 : const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket,
907 : NodeId node)
908 : {
909 1548 : NodeId idToFind;
910 1548 : std::list<Bucket>::iterator bucket;
911 :
912 1548 : if (ec == asio::error::operation_aborted)
913 0 : return;
914 :
915 1549 : if (!node) {
916 0 : bucket = routing_table.findBucket(socket->deviceId());
917 0 : idToFind = bucket->randomId(rd);
918 : } else {
919 1548 : bucket = routing_table.findBucket(node);
920 1548 : idToFind = node;
921 : }
922 :
923 1548 : sendRequest(socket, idToFind, Query::FIND, Bucket::BUCKET_MAX_SIZE);
924 :
925 1549 : if (!node) {
926 0 : auto& nodeTimer = bucket->getNodeTimer(socket);
927 0 : nodeTimer.expires_after(FIND_PERIOD);
928 0 : nodeTimer.async_wait(std::bind(&jami::SwarmManager::resetNodeExpiry,
929 0 : shared_from_this(),
930 : std::placeholders::_1,
931 : socket,
932 0 : NodeId {}));
933 : }
934 : }
935 :
936 : void
937 1608 : SwarmManager::tryConnect(const NodeId& nodeId, bool noNewSocket)
938 : {
939 1608 : if (needSocketCb_)
940 1604 : needSocketCb_(
941 3210 : nodeId.toString(),
942 3207 : [w = weak(), nodeId](const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket) {
943 1439 : auto shared = w.lock();
944 1438 : if (!shared || shared->isShutdown_)
945 272 : return true;
946 1168 : if (socket) {
947 1063 : shared->addChannel(socket);
948 1061 : return true;
949 : }
950 105 : std::unique_lock lk(shared->mutex);
951 105 : auto bucket = shared->routing_table.findBucket(nodeId);
952 105 : bucket->removeConnectingNode(nodeId);
953 105 : if (!bucket->hasMobileNode(nodeId))
954 104 : bucket->addKnownNode(nodeId);
955 105 : if (shared->routing_table.getActiveNodesCount() == 0 && shared->onConnectionChanged_) {
956 58 : lk.unlock();
957 58 : JAMI_LOG("[SwarmManager {:p}] Bootstrap: all connections failed", fmt::ptr(shared.get()));
958 58 : shared->onConnectionChanged_(false);
959 : }
960 105 : return true;
961 1438 : },
962 : noNewSocket);
963 1608 : }
964 :
965 : void
966 636 : SwarmManager::removeNodeInternal(const NodeId& nodeId)
967 : {
968 636 : routing_table.removeNode(nodeId);
969 636 : }
970 :
971 : void
972 520 : SwarmManager::connectNode(const NodeId& nodeId)
973 : {
974 : {
975 520 : std::lock_guard lock(mutex);
976 520 : if (isShutdown_)
977 3 : return;
978 517 : if (isConnectedWith(nodeId))
979 0 : return;
980 517 : addKnownNode(nodeId);
981 517 : if (!routing_table.addConnectingNode(nodeId))
982 0 : return;
983 520 : }
984 517 : tryConnect(nodeId, true);
985 : }
986 :
987 : std::vector<NodeId>
988 20 : SwarmManager::getAllNodes() const
989 : {
990 20 : std::lock_guard lock(mutex);
991 40 : return routing_table.getAllNodes();
992 20 : }
993 :
994 : std::vector<NodeId>
995 1816 : SwarmManager::getConnectedNodes() const
996 : {
997 1816 : std::lock_guard lock(mutex);
998 3632 : return routing_table.getConnectedNodes();
999 1816 : }
1000 :
1001 : std::vector<NodeId>
1002 57 : SwarmManager::getMobileNodesToNotify()
1003 : {
1004 57 : std::lock_guard lock(mutex);
1005 114 : return routing_table.getMobileNodesToNotify();
1006 57 : }
1007 :
1008 : std::vector<NodeId>
1009 314 : SwarmManager::getKnownMobileNodes() const
1010 : {
1011 314 : std::lock_guard lock(mutex);
1012 629 : return routing_table.getKnownMobileNodes();
1013 315 : }
1014 :
1015 : std::vector<MobileNodeInfo>
1016 330 : SwarmManager::getKnownMobileNodeInfos() const
1017 : {
1018 330 : std::lock_guard lock(mutex);
1019 331 : std::vector<MobileNodeInfo> infos;
1020 331 : const auto now = toSecondsSinceEpoch(std::chrono::system_clock::now());
1021 854 : for (const auto& node : routing_table.getKnownMobileNodes()) {
1022 524 : if (!isMobileNodeCurrentInternal(node, now))
1023 31 : continue;
1024 493 : auto lease = mobileNodeLeases_.find(node);
1025 494 : infos.emplace_back(MobileNodeInfo {node,
1026 988 : lease == mobileNodeLeases_.end()
1027 494 : ? std::nullopt
1028 28 : : std::optional<MobileLease>(lease->second)});
1029 332 : }
1030 659 : return infos;
1031 330 : }
1032 :
1033 : std::vector<MobileNodeInfo>
1034 1776 : SwarmManager::getMobileNodeInfosToNotify()
1035 : {
1036 1776 : std::lock_guard lock(mutex);
1037 1775 : std::vector<MobileNodeInfo> infos;
1038 1775 : const auto now = toSecondsSinceEpoch(std::chrono::system_clock::now());
1039 1777 : for (const auto& node : routing_table.getMobileNodesToNotify()) {
1040 1 : if (!isMobileNodeCurrentInternal(node, now))
1041 0 : continue;
1042 1 : auto lease = mobileNodeLeases_.find(node);
1043 1 : infos.emplace_back(MobileNodeInfo {node,
1044 2 : lease == mobileNodeLeases_.end()
1045 1 : ? std::nullopt
1046 0 : : std::optional<MobileLease>(lease->second)});
1047 1775 : }
1048 3551 : return infos;
1049 1775 : }
1050 :
1051 : std::vector<std::map<std::string, std::string>>
1052 2 : SwarmManager::getRoutingTableInfo() const
1053 : {
1054 2 : std::lock_guard lock(mutex);
1055 2 : auto stats = routing_table.getRoutingTableStats();
1056 2 : const auto toNotify = routing_table.getMobileNodesToNotify();
1057 2 : std::set<std::string> responsible;
1058 4 : for (const auto& node : toNotify)
1059 2 : responsible.emplace(node.toString());
1060 2 : std::vector<std::map<std::string, std::string>> result;
1061 2 : result.reserve(stats.size());
1062 11 : for (const auto& stat : stats) {
1063 72 : result.push_back({{"id", stat.id},
1064 9 : {"device", stat.id},
1065 9 : {"status", stat.status},
1066 9 : {"remoteAddress", stat.remoteAddress},
1067 9 : {"mobile", stat.isMobile ? "true" : "false"},
1068 9 : {"responsible", responsible.count(stat.id) ? "true" : "false"}});
1069 9 : if (stat.connectionTime != std::chrono::system_clock::time_point::min()) {
1070 0 : auto tt = std::chrono::system_clock::to_time_t(stat.connectionTime);
1071 0 : result.back().emplace("connectionTime", std::to_string(tt));
1072 : }
1073 : }
1074 4 : return result;
1075 29 : }
1076 :
1077 : bool
1078 3369 : SwarmManager::isConnected() const
1079 : {
1080 3369 : std::lock_guard lock(mutex);
1081 6737 : return !routing_table.isEmpty();
1082 3369 : }
1083 :
1084 : void
1085 16 : SwarmManager::deleteNode(const std::vector<NodeId>& nodes)
1086 : {
1087 16 : bool mobileNodesChanged = false;
1088 : {
1089 16 : std::lock_guard lock(mutex);
1090 16 : auto mobileNodes = routing_table.getKnownMobileNodes();
1091 32 : for (const auto& node : nodes) {
1092 16 : routing_table.deleteNode(node);
1093 16 : mobileNodesChanged |= mobileNodeLeases_.erase(node) != 0;
1094 16 : mobileNodesChanged |= legacyMobileNodeExpiries_.erase(node) != 0;
1095 16 : pendingMobileLeases_.erase(node);
1096 : }
1097 16 : scheduleMobileLeaseExpiryInternal();
1098 16 : mobileNodesChanged |= mobileNodes != routing_table.getKnownMobileNodes();
1099 16 : }
1100 16 : if (mobileNodesChanged)
1101 2 : emitMobileNodesChanged();
1102 16 : maintainBuckets();
1103 16 : }
1104 :
1105 : } // namespace jami
|
