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 2622 : sendMsg(const std::shared_ptr<dhtnet::ChannelSocket>& s, const T& msg)
35 : {
36 2622 : if (!s)
37 0 : return false;
38 2622 : msgpack::sbuffer buf;
39 2622 : msgpack::pack(buf, msg);
40 2622 : std::error_code ec;
41 2622 : auto sent = s->write(reinterpret_cast<const uint8_t*>(buf.data()), buf.size(), ec);
42 2622 : if (ec) {
43 0 : JAMI_WARNING("[SvcDiscovery] write error: {}", ec.message());
44 0 : return false;
45 : }
46 2622 : return sent == buf.size();
47 2622 : }
48 :
49 : constexpr const char* SVC_CACHE_FILENAME = "svc_discovery_cache.msgpack";
50 :
51 : } // namespace
52 :
53 796 : SvcDiscoveryChannelHandler::SvcDiscoveryChannelHandler(const std::shared_ptr<JamiAccount>& acc,
54 : dhtnet::ConnectionManager& cm,
55 796 : std::filesystem::path cachePath)
56 : : ChannelHandlerInterface()
57 796 : , account_(acc)
58 796 : , connectionManager_(cm)
59 796 : , state_(std::make_shared<State>())
60 1592 : , cachePath_(std::move(cachePath))
61 : {
62 796 : loadCache();
63 :
64 : // Install the default response callback that populates the cache.
65 796 : auto cachePathCopy = cachePath_;
66 1592 : state_->responseCb = [state = state_, cachePathCopy](const std::string& peerAccountUri,
67 : const std::string& peerDeviceId,
68 : const std::vector<svc_protocol::SvcInfo>& services) {
69 1313 : if (peerDeviceId.empty())
70 0 : return;
71 1313 : DeviceId dev;
72 : try {
73 1313 : dev = DeviceId(peerDeviceId);
74 0 : } catch (...) {
75 0 : return;
76 0 : }
77 1313 : CacheUpdateCb updateCb;
78 1313 : msgpack::sbuffer buf;
79 : {
80 1313 : std::lock_guard lk(state->mtx);
81 1313 : state->cache[dev] = State::CachedDeviceServices {peerAccountUri, services};
82 1313 : updateCb = state->cacheUpdateCb;
83 1313 : msgpack::pack(buf, state->cache);
84 1313 : }
85 : // Write to disk outside the lock
86 1313 : if (!cachePathCopy.empty()) {
87 1312 : std::error_code ec;
88 1312 : std::filesystem::create_directories(cachePathCopy, ec);
89 1313 : std::ofstream out(cachePathCopy / SVC_CACHE_FILENAME, std::ios::binary | std::ios::trunc);
90 1313 : if (out)
91 1313 : out.write(buf.data(), buf.size());
92 1313 : }
93 1313 : if (updateCb)
94 1313 : updateCb(peerAccountUri, dev, services);
95 2109 : };
96 796 : }
97 :
98 1592 : SvcDiscoveryChannelHandler::~SvcDiscoveryChannelHandler() = default;
99 :
100 : void
101 796 : SvcDiscoveryChannelHandler::onCacheUpdated(CacheUpdateCb cb)
102 : {
103 796 : std::lock_guard lk(state_->mtx);
104 796 : state_->cacheUpdateCb = std::move(cb);
105 796 : }
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 1310 : SvcDiscoveryChannelHandler::buildResponse(JamiAccount& account, const std::string& peerAccountUri)
116 : {
117 1310 : svc_protocol::SvcDiscResponse out;
118 1310 : if (const auto& id = account.identity().first)
119 1310 : out.device = id->getPublicKey().getLongId().toString();
120 0 : auto checker = [&account](const std::string& uri) {
121 0 : return account.isContact(uri);
122 1310 : };
123 1310 : auto visible = account.serviceManager().getVisibleServices(peerAccountUri, checker);
124 1310 : out.services.reserve(visible.size());
125 1315 : 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 2620 : return out;
136 1310 : }
137 :
138 : void
139 1310 : SvcDiscoveryChannelHandler::connect(const DeviceId& deviceId,
140 : const std::string& /*name*/,
141 : ConnectCb&& cb,
142 : const std::string& /*connectionType*/,
143 : bool /*forceNewConnection*/)
144 : {
145 1310 : auto userCb = std::make_shared<ConnectCb>(std::move(cb));
146 1310 : auto state = state_;
147 1310 : auto wacc = account_;
148 1310 : connectionManager_.connectDevice(deviceId,
149 2620 : std::string(svc_protocol::DiscoveryChannelName),
150 2620 : [userCb, state, wacc, this](std::shared_ptr<dhtnet::ChannelSocket> socket,
151 : const DeviceId& dev) {
152 1310 : 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 1305 : auto cert = socket->peerCertificate();
156 1305 : std::string peerAccountUri;
157 1305 : if (cert && cert->issuer)
158 1305 : peerAccountUri = cert->issuer->getId().toString();
159 : {
160 1305 : std::lock_guard lk(state->mtx);
161 1305 : state->channels[peerAccountUri].push_back(socket);
162 1305 : }
163 1305 : socket->onShutdown([state, ws = std::weak_ptr(socket), peerAccountUri](
164 : const std::error_code&) {
165 1305 : auto s = ws.lock();
166 1305 : if (!s)
167 0 : return;
168 1305 : std::lock_guard lk(state->mtx);
169 1305 : auto it = state->channels.find(peerAccountUri);
170 1305 : if (it != state->channels.end()) {
171 1305 : auto& vec = it->second;
172 1305 : vec.erase(std::remove(vec.begin(), vec.end(), s), vec.end());
173 1305 : if (vec.empty())
174 646 : state->channels.erase(it);
175 : }
176 1305 : });
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 1305 : installReader(socket, peerAccountUri);
181 3915 : if (!sendMsg(socket, svc_protocol::SvcDiscQuery {}))
182 0 : JAMI_WARNING("[SvcDiscovery] failed to send SvcDiscQuery to {}",
183 : peerAccountUri);
184 1305 : }
185 1310 : if (*userCb)
186 1310 : (*userCb)(socket, dev);
187 1310 : });
188 2620 : }
189 :
190 : bool
191 1305 : SvcDiscoveryChannelHandler::onRequest(const std::shared_ptr<dht::crypto::Certificate>& peer, const std::string& /*name*/)
192 : {
193 1305 : return peer && peer->issuer;
194 : }
195 :
196 : void
197 2605 : SvcDiscoveryChannelHandler::onReady(const std::shared_ptr<dht::crypto::Certificate>& peer,
198 : const std::string& /*name*/,
199 : std::shared_ptr<dhtnet::ChannelSocket> channel)
200 : {
201 2605 : 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 2603 : if (channel->isInitiator())
206 1305 : return;
207 1304 : if (!peer || !peer->issuer) {
208 0 : channel->shutdown();
209 0 : return;
210 : }
211 1304 : auto peerUri = peer->issuer->getId().toString();
212 : {
213 1304 : std::lock_guard lk(state_->mtx);
214 1304 : state_->channels[peerUri].push_back(channel);
215 1304 : }
216 1304 : auto state = state_;
217 1304 : channel->onShutdown([state, ws = std::weak_ptr(channel), peerUri](const std::error_code&) {
218 1304 : auto s = ws.lock();
219 1304 : if (!s)
220 0 : return;
221 1304 : std::lock_guard lk(state->mtx);
222 1304 : auto it = state->channels.find(peerUri);
223 1304 : if (it != state->channels.end()) {
224 1304 : auto& vec = it->second;
225 1304 : vec.erase(std::remove(vec.begin(), vec.end(), s), vec.end());
226 1304 : if (vec.empty())
227 647 : state->channels.erase(it);
228 : }
229 1303 : });
230 1304 : installReader(channel, peer->issuer->getId().toString());
231 1304 : }
232 :
233 : void
234 2608 : SvcDiscoveryChannelHandler::installReader(const std::shared_ptr<dhtnet::ChannelSocket>& channel,
235 : std::string peerAccountUri)
236 : {
237 2608 : auto reader = std::make_shared<msgpack::unpacker>();
238 2608 : reader->reserve_buffer(4096);
239 2607 : auto wacc = account_;
240 2606 : auto state = state_;
241 2608 : std::weak_ptr<dhtnet::ChannelSocket> wsock = channel;
242 :
243 2608 : channel->setOnRecv([reader, wacc, state, wsock, peerAccountUri = std::move(peerAccountUri)](const uint8_t* data,
244 : size_t size) -> ssize_t {
245 2616 : if (size == 0)
246 0 : return 0;
247 2616 : if (reader->buffer_capacity() < size)
248 0 : reader->reserve_buffer(size);
249 2616 : std::memcpy(reader->buffer(), data, size);
250 2616 : reader->buffer_consumed(size);
251 :
252 2616 : msgpack::object_handle oh;
253 5232 : while (reader->next(oh)) {
254 2616 : const auto& obj = oh.get();
255 2616 : const auto type = svc_protocol::peekType(obj);
256 2616 : const auto v = svc_protocol::peekVersion(obj);
257 2615 : auto sock = wsock.lock();
258 2615 : if (!sock)
259 0 : return static_cast<ssize_t>(size);
260 :
261 2615 : if (type == svc_protocol::MsgType::Query) {
262 1303 : auto acc = wacc.lock();
263 1303 : if (!acc) {
264 0 : sock->shutdown();
265 0 : continue;
266 : }
267 1303 : 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 1303 : auto resp = SvcDiscoveryChannelHandler::buildResponse(*acc, peerAccountUri);
274 1303 : JAMI_LOG("[SvcDiscovery] returning {} service(s) to peer={}", resp.services.size(), peerAccountUri);
275 1303 : if (!sendMsg(sock, resp))
276 0 : JAMI_WARNING("[SvcDiscovery] failed to send SvcDiscResponse to {}", peerAccountUri);
277 2615 : } else if (type == svc_protocol::MsgType::ServiceList) {
278 1298 : svc_protocol::SvcDiscResponse resp;
279 : try {
280 1299 : 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 1299 : ResponseCb cb;
286 : {
287 1299 : std::lock_guard lk(state->mtx);
288 1299 : cb = state->responseCb;
289 1299 : }
290 1299 : if (cb)
291 1299 : 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 1313 : } 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 2616 : }
323 2616 : return static_cast<ssize_t>(size);
324 2616 : });
325 2608 : }
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 1310 : SvcDiscoveryChannelHandler::refreshDevice(const std::string& /*peerUri*/, const DeviceId& deviceId)
361 : {
362 3930 : connect(deviceId,
363 2620 : std::string(svc_protocol::DiscoveryChannelName),
364 1310 : [](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 1310 : });
368 1310 : }
369 :
370 : std::vector<SvcDiscoveryChannelHandler::CachedSvcInfo>
371 2081 : SvcDiscoveryChannelHandler::getCachedServices(const std::string& peerUri) const
372 : {
373 2081 : std::vector<CachedSvcInfo> result;
374 2081 : std::lock_guard lk(state_->mtx);
375 11880 : for (const auto& [dev, entry] : state_->cache) {
376 9801 : if (entry.peerUri == peerUri) {
377 1589 : for (const auto& svc : entry.services)
378 10 : result.push_back(CachedSvcInfo {dev, svc});
379 : }
380 : }
381 4164 : return result;
382 2082 : }
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 796 : SvcDiscoveryChannelHandler::loadCache()
412 : {
413 796 : if (cachePath_.empty())
414 0 : return;
415 796 : auto path = cachePath_ / SVC_CACHE_FILENAME;
416 796 : std::ifstream in(path, std::ios::binary);
417 796 : if (!in)
418 790 : 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 1586 : }
430 :
431 : } // namespace jami
|
