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 : #include "jamidht/svc_discovery_channel_handler.h"
18 :
19 : #include "jamidht/account_manager.h"
20 : #include "jamidht/contact_list.h"
21 : #include "jamidht/service_manager.h"
22 : #include "logger.h"
23 :
24 : #include <algorithm>
25 : #include <cstring>
26 : #include <fstream>
27 :
28 : namespace jami {
29 :
30 : namespace {
31 :
32 : template<typename T>
33 : bool
34 2516 : sendMsg(const std::shared_ptr<dhtnet::ChannelSocket>& s, const T& msg)
35 : {
36 2516 : if (!s)
37 0 : return false;
38 2516 : msgpack::sbuffer buf;
39 2516 : msgpack::pack(buf, msg);
40 2516 : std::error_code ec;
41 2516 : auto sent = s->write(reinterpret_cast<const uint8_t*>(buf.data()), buf.size(), ec);
42 2516 : if (ec) {
43 0 : JAMI_WARNING("[SvcDiscovery] write error: {}", ec.message());
44 0 : return false;
45 : }
46 2516 : return sent == buf.size();
47 2516 : }
48 :
49 : constexpr const char* SVC_CACHE_FILENAME = "svc_discovery_cache.msgpack";
50 :
51 : } // namespace
52 :
53 721 : SvcDiscoveryChannelHandler::SvcDiscoveryChannelHandler(const std::shared_ptr<JamiAccount>& acc,
54 : dhtnet::ConnectionManager& cm,
55 721 : std::filesystem::path cachePath)
56 : : ChannelHandlerInterface()
57 721 : , account_(acc)
58 721 : , connectionManager_(cm)
59 721 : , state_(std::make_shared<State>())
60 1442 : , cachePath_(std::move(cachePath))
61 : {
62 721 : loadCache();
63 :
64 : // Install the default response callback that populates the cache.
65 721 : auto cachePathCopy = cachePath_;
66 1442 : state_->responseCb = [state = state_, cachePathCopy](const std::string& peerAccountUri,
67 : const std::string& peerDeviceId,
68 : const std::vector<svc_protocol::SvcInfo>& services) {
69 1258 : if (peerDeviceId.empty())
70 0 : return;
71 1257 : DeviceId dev;
72 : try {
73 1257 : dev = DeviceId(peerDeviceId);
74 0 : } catch (...) {
75 0 : return;
76 0 : }
77 1258 : CacheUpdateCb updateCb;
78 1258 : msgpack::sbuffer buf;
79 : {
80 1258 : std::lock_guard lk(state->mtx);
81 1258 : state->cache[dev] = State::CachedDeviceServices {peerAccountUri, services};
82 1258 : updateCb = state->cacheUpdateCb;
83 1258 : msgpack::pack(buf, state->cache);
84 1258 : }
85 : // Write to disk outside the lock
86 1258 : if (!cachePathCopy.empty()) {
87 1258 : std::error_code ec;
88 1258 : std::filesystem::create_directories(cachePathCopy, ec);
89 1258 : std::ofstream out(cachePathCopy / SVC_CACHE_FILENAME, std::ios::binary | std::ios::trunc);
90 1258 : if (out)
91 1258 : out.write(buf.data(), buf.size());
92 1258 : }
93 1258 : if (updateCb)
94 1258 : updateCb(peerAccountUri, dev, services);
95 1979 : };
96 721 : }
97 :
98 1442 : SvcDiscoveryChannelHandler::~SvcDiscoveryChannelHandler() = default;
99 :
100 : void
101 721 : SvcDiscoveryChannelHandler::onCacheUpdated(CacheUpdateCb cb)
102 : {
103 721 : std::lock_guard lk(state_->mtx);
104 721 : state_->cacheUpdateCb = std::move(cb);
105 721 : }
106 :
107 : void
108 0 : SvcDiscoveryChannelHandler::setOnResponse(ResponseCb cb)
109 : {
110 0 : std::lock_guard lk(state_->mtx);
111 0 : state_->responseCb = std::move(cb);
112 0 : }
113 :
114 : svc_protocol::SvcDiscResponse
115 1256 : SvcDiscoveryChannelHandler::buildResponse(JamiAccount& account, const std::string& peerAccountUri)
116 : {
117 1256 : svc_protocol::SvcDiscResponse out;
118 1256 : if (const auto& id = account.identity().first)
119 1256 : out.device = id->getPublicKey().getLongId().toString();
120 0 : auto checker = [&account](const std::string& uri) {
121 0 : return account.isContact(uri);
122 1256 : };
123 1256 : auto visible = account.serviceManager().getVisibleServices(peerAccountUri, checker);
124 1256 : out.services.reserve(visible.size());
125 1261 : for (auto& r : visible) {
126 5 : svc_protocol::SvcInfo info;
127 5 : info.id = std::move(r.id);
128 5 : info.name = std::move(r.name);
129 5 : info.description = std::move(r.description);
130 5 : info.proto = "tcp";
131 5 : info.scheme = std::move(r.scheme);
132 5 : info.preferred_port = r.preferredPort;
133 5 : out.services.push_back(std::move(info));
134 5 : }
135 2512 : return out;
136 1256 : }
137 :
138 : void
139 1259 : SvcDiscoveryChannelHandler::connect(const DeviceId& deviceId,
140 : const std::string& /*name*/,
141 : ConnectCb&& cb,
142 : const std::string& /*connectionType*/,
143 : bool /*forceNewConnection*/)
144 : {
145 1259 : auto userCb = std::make_shared<ConnectCb>(std::move(cb));
146 1259 : auto state = state_;
147 1259 : auto wacc = account_;
148 1259 : connectionManager_.connectDevice(deviceId,
149 2518 : std::string(svc_protocol::DiscoveryChannelName),
150 2518 : [userCb, state, wacc, this](std::shared_ptr<dhtnet::ChannelSocket> socket,
151 : const DeviceId& dev) {
152 1259 : if (socket) {
153 : // Retain the channel for its full lifetime so the response
154 : // can come back even if no one else holds it.
155 1253 : auto cert = socket->peerCertificate();
156 1253 : std::string peerAccountUri;
157 1253 : if (cert && cert->issuer)
158 1253 : peerAccountUri = cert->issuer->getId().toString();
159 : {
160 1253 : std::lock_guard lk(state->mtx);
161 1253 : state->channels[peerAccountUri].push_back(socket);
162 1253 : }
163 1253 : socket->onShutdown([state, ws = std::weak_ptr(socket), peerAccountUri](
164 : const std::error_code&) {
165 1253 : auto s = ws.lock();
166 1253 : if (!s)
167 0 : return;
168 1253 : std::lock_guard lk(state->mtx);
169 1252 : auto it = state->channels.find(peerAccountUri);
170 1253 : if (it != state->channels.end()) {
171 1253 : auto& vec = it->second;
172 1253 : vec.erase(std::remove(vec.begin(), vec.end(), s), vec.end());
173 1253 : if (vec.empty())
174 622 : state->channels.erase(it);
175 : }
176 1253 : });
177 : // The initiating side immediately writes a Query so the server
178 : // can respond. We need to install a reader to handle the
179 : // response too.
180 1253 : installReader(socket, peerAccountUri);
181 3759 : if (!sendMsg(socket, svc_protocol::SvcDiscQuery {}))
182 0 : JAMI_WARNING("[SvcDiscovery] failed to send SvcDiscQuery to {}",
183 : peerAccountUri);
184 1253 : }
185 1259 : if (*userCb)
186 1259 : (*userCb)(socket, dev);
187 1259 : });
188 2518 : }
189 :
190 : bool
191 1254 : SvcDiscoveryChannelHandler::onRequest(const std::shared_ptr<dht::crypto::Certificate>& peer, const std::string& /*name*/)
192 : {
193 1254 : return peer && peer->issuer;
194 : }
195 :
196 : void
197 2505 : SvcDiscoveryChannelHandler::onReady(const std::shared_ptr<dht::crypto::Certificate>& peer,
198 : const std::string& /*name*/,
199 : std::shared_ptr<dhtnet::ChannelSocket> channel)
200 : {
201 2505 : if (!channel)
202 0 : return;
203 : // The initiator already retains the socket and installs its reader from
204 : // connect(); no need to do the work twice.
205 2505 : if (channel->isInitiator())
206 1253 : return;
207 1252 : if (!peer || !peer->issuer) {
208 0 : channel->shutdown();
209 0 : return;
210 : }
211 1252 : auto peerUri = peer->issuer->getId().toString();
212 : {
213 1252 : std::lock_guard lk(state_->mtx);
214 1252 : state_->channels[peerUri].push_back(channel);
215 1252 : }
216 1252 : auto state = state_;
217 1252 : channel->onShutdown([state, ws = std::weak_ptr(channel), peerUri](const std::error_code&) {
218 1252 : auto s = ws.lock();
219 1252 : if (!s)
220 0 : return;
221 1252 : std::lock_guard lk(state->mtx);
222 1252 : auto it = state->channels.find(peerUri);
223 1252 : if (it != state->channels.end()) {
224 1251 : auto& vec = it->second;
225 1252 : vec.erase(std::remove(vec.begin(), vec.end(), s), vec.end());
226 1252 : if (vec.empty())
227 620 : state->channels.erase(it);
228 : }
229 1251 : });
230 1252 : installReader(channel, peer->issuer->getId().toString());
231 1252 : }
232 :
233 : void
234 2504 : SvcDiscoveryChannelHandler::installReader(const std::shared_ptr<dhtnet::ChannelSocket>& channel,
235 : std::string peerAccountUri)
236 : {
237 2504 : auto reader = std::make_shared<msgpack::unpacker>();
238 2504 : reader->reserve_buffer(4096);
239 2505 : auto wacc = account_;
240 2505 : auto state = state_;
241 2503 : std::weak_ptr<dhtnet::ChannelSocket> wsock = channel;
242 :
243 2502 : channel->setOnRecv([reader, wacc, state, wsock, peerAccountUri = std::move(peerAccountUri)](const uint8_t* data,
244 : size_t size) -> ssize_t {
245 2506 : if (size == 0)
246 0 : return 0;
247 2506 : if (reader->buffer_capacity() < size)
248 0 : reader->reserve_buffer(size);
249 2505 : std::memcpy(reader->buffer(), data, size);
250 2505 : reader->buffer_consumed(size);
251 :
252 2505 : msgpack::object_handle oh;
253 5012 : while (reader->next(oh)) {
254 2506 : const auto& obj = oh.get();
255 2507 : const auto type = svc_protocol::peekType(obj);
256 2507 : const auto v = svc_protocol::peekVersion(obj);
257 2507 : auto sock = wsock.lock();
258 2507 : if (!sock)
259 0 : return static_cast<ssize_t>(size);
260 :
261 2507 : if (type == svc_protocol::MsgType::Query) {
262 1249 : auto acc = wacc.lock();
263 1249 : if (!acc) {
264 0 : sock->shutdown();
265 0 : continue;
266 : }
267 1249 : if (v > svc_protocol::MaxVersion) {
268 0 : svc_protocol::SvcDiscVersionMismatch vm;
269 0 : vm.max_supported = svc_protocol::MaxVersion;
270 0 : sendMsg(sock, vm);
271 0 : continue;
272 0 : }
273 1249 : auto resp = SvcDiscoveryChannelHandler::buildResponse(*acc, peerAccountUri);
274 1249 : JAMI_LOG("[SvcDiscovery] returning {} service(s) to peer={}", resp.services.size(), peerAccountUri);
275 1249 : if (!sendMsg(sock, resp))
276 0 : JAMI_WARNING("[SvcDiscovery] failed to send SvcDiscResponse to {}", peerAccountUri);
277 2507 : } else if (type == svc_protocol::MsgType::ServiceList) {
278 1244 : svc_protocol::SvcDiscResponse resp;
279 : try {
280 1244 : obj.convert(resp);
281 0 : } catch (const std::exception& e) {
282 0 : JAMI_WARNING("[SvcDiscovery] bad service_list: {}", e.what());
283 0 : continue;
284 0 : }
285 1244 : ResponseCb cb;
286 : {
287 1243 : std::lock_guard lk(state->mtx);
288 1244 : cb = state->responseCb;
289 1244 : }
290 1244 : if (cb)
291 1244 : cb(peerAccountUri, resp.device, resp.services);
292 : else
293 0 : JAMI_WARNING("[SvcDiscovery] no responseCb set; dropping {} services from {}",
294 : resp.services.size(),
295 : peerAccountUri);
296 1258 : } else if (type == svc_protocol::MsgType::ServiceUpdate) {
297 14 : svc_protocol::SvcDiscServiceUpdate update;
298 : try {
299 14 : obj.convert(update);
300 0 : } catch (const std::exception& e) {
301 0 : JAMI_WARNING("[SvcDiscovery] bad service_update: {}", e.what());
302 0 : continue;
303 0 : }
304 14 : ResponseCb cb;
305 : {
306 14 : std::lock_guard lk(state->mtx);
307 14 : cb = state->responseCb;
308 14 : }
309 14 : if (cb)
310 14 : cb(peerAccountUri, update.device, update.services);
311 14 : } else if (type == svc_protocol::MsgType::VersionMismatch || type == svc_protocol::MsgType::Error) {
312 0 : ResponseCb cb;
313 : {
314 0 : std::lock_guard lk(state->mtx);
315 0 : cb = state->responseCb;
316 0 : }
317 0 : if (cb)
318 0 : cb(peerAccountUri, std::string {}, {});
319 0 : } else {
320 0 : JAMI_WARNING("[SvcDiscovery] unknown message type '{}'", type);
321 : }
322 2507 : }
323 2507 : return static_cast<ssize_t>(size);
324 2507 : });
325 2504 : }
326 :
327 : void
328 8 : SvcDiscoveryChannelHandler::broadcastServiceUpdate()
329 : {
330 8 : auto acc = account_.lock();
331 8 : if (!acc)
332 0 : return;
333 :
334 : // Snapshot connected peers and their channels under the lock.
335 8 : std::map<std::string, std::vector<std::shared_ptr<dhtnet::ChannelSocket>>> snapshot;
336 : {
337 8 : std::lock_guard lk(state_->mtx);
338 8 : snapshot = state_->channels;
339 8 : }
340 :
341 8 : if (snapshot.empty())
342 1 : return;
343 :
344 7 : std::size_t peerCount = 0;
345 14 : for (const auto& [peerUri, sockets] : snapshot) {
346 : // Build a per-peer filtered service list.
347 7 : auto resp = buildResponse(*acc, peerUri);
348 7 : svc_protocol::SvcDiscServiceUpdate update;
349 7 : update.device = resp.device;
350 7 : update.services = std::move(resp.services);
351 :
352 21 : for (const auto& sock : sockets)
353 14 : sendMsg(sock, update);
354 7 : ++peerCount;
355 7 : }
356 7 : JAMI_LOG("[SvcDiscovery] broadcast service_update to {} peer(s)", peerCount);
357 9 : }
358 :
359 : void
360 1259 : SvcDiscoveryChannelHandler::refreshDevice(const std::string& /*peerUri*/, const DeviceId& deviceId)
361 : {
362 3777 : connect(deviceId,
363 2518 : std::string(svc_protocol::DiscoveryChannelName),
364 1258 : [](std::shared_ptr<dhtnet::ChannelSocket>, const DeviceId&) {
365 : // Connection callback — nothing needed here; the cache is
366 : // updated when the response is read via installReader/responseCb.
367 1259 : });
368 1258 : }
369 :
370 : std::vector<SvcDiscoveryChannelHandler::CachedSvcInfo>
371 1928 : SvcDiscoveryChannelHandler::getCachedServices(const std::string& peerUri) const
372 : {
373 1928 : std::vector<CachedSvcInfo> result;
374 1928 : std::lock_guard lk(state_->mtx);
375 11596 : for (const auto& [dev, entry] : state_->cache) {
376 9672 : if (entry.peerUri == peerUri) {
377 1513 : for (const auto& svc : entry.services)
378 10 : result.push_back(CachedSvcInfo {dev, svc});
379 : }
380 : }
381 3856 : return result;
382 1928 : }
383 :
384 : void
385 0 : SvcDiscoveryChannelHandler::removeDevice(const DeviceId& deviceId)
386 : {
387 : {
388 0 : std::lock_guard lk(state_->mtx);
389 0 : state_->cache.erase(deviceId);
390 0 : }
391 0 : saveCache();
392 0 : }
393 :
394 : void
395 0 : SvcDiscoveryChannelHandler::saveCache() const
396 : {
397 0 : if (cachePath_.empty())
398 0 : return;
399 0 : std::error_code ec;
400 0 : std::filesystem::create_directories(cachePath_, ec);
401 0 : std::ofstream out(cachePath_ / SVC_CACHE_FILENAME, std::ios::binary | std::ios::trunc);
402 0 : if (out) {
403 0 : std::lock_guard lk(state_->mtx);
404 0 : msgpack::pack(out, state_->cache);
405 0 : } else {
406 0 : JAMI_WARNING("[SvcDiscovery] failed to save cache to disk");
407 : }
408 0 : }
409 :
410 : void
411 721 : SvcDiscoveryChannelHandler::loadCache()
412 : {
413 721 : if (cachePath_.empty())
414 0 : return;
415 721 : auto path = cachePath_ / SVC_CACHE_FILENAME;
416 721 : std::ifstream in(path, std::ios::binary);
417 721 : if (!in)
418 715 : return;
419 6 : std::string content((std::istreambuf_iterator<char>(in)), std::istreambuf_iterator<char>());
420 : try {
421 6 : auto oh = msgpack::unpack(content.data(), content.size());
422 6 : std::lock_guard lk(state_->mtx);
423 6 : oh.get().convert(state_->cache);
424 6 : } catch (const std::exception& e) {
425 0 : JAMI_WARNING("[SvcDiscovery] failed to load cache: {}", e.what());
426 0 : return;
427 0 : }
428 6 : JAMI_LOG("[SvcDiscovery] loaded {} cached device entries from disk", state_->cache.size());
429 1436 : }
430 :
431 : } // namespace jami
|
