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