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 "data_transfer.h"
19 :
20 : #include "base64.h"
21 : #include "fileutils.h"
22 : #include "manager.h"
23 : #include "client/jami_signal.h"
24 :
25 : #include <algorithm>
26 : #include <mutex>
27 : #include <cstdlib> // mkstemp
28 : #include <filesystem>
29 : #include <limits>
30 :
31 : #include <opendht/rng.h>
32 : #include <opendht/thread_pool.h>
33 :
34 : namespace jami {
35 :
36 : namespace {
37 :
38 : bool
39 58 : isExpectedFile(const std::filesystem::path& path, const std::string& sha3sum, std::size_t total)
40 : {
41 58 : std::error_code ec;
42 58 : return std::filesystem::file_size(path, ec) == total && fileutils::sha3File(path) == sha3sum;
43 : }
44 :
45 : bool
46 173 : isMissingPath(const std::filesystem::file_status& status, const std::error_code& ec)
47 : {
48 173 : return status.type() == std::filesystem::file_type::not_found || ec == std::errc::no_such_file_or_directory;
49 : }
50 :
51 : } // namespace
52 :
53 : libjami::DataTransferId
54 269 : generateUID(std::mt19937_64& engine)
55 : {
56 269 : return std::uniform_int_distribution<libjami::DataTransferId> {1, JAMI_ID_MAX_VAL}(engine);
57 : }
58 :
59 : namespace {
60 :
61 : constexpr size_t COMMIT_ID_SIZE = 40; // hex SHA-1
62 : constexpr size_t MAX_TID_SIZE = 20; // decimal uint64_t
63 :
64 : bool
65 582 : isLowerHex(std::string_view s)
66 : {
67 582 : return !s.empty() && std::all_of(s.begin(), s.end(), [](unsigned char c) {
68 23280 : return (c >= '0' && c <= '9') || (c >= 'a' && c <= 'f');
69 582 : });
70 : }
71 :
72 : bool
73 582 : isDecimal(std::string_view s)
74 : {
75 1158 : return !s.empty() && s.size() <= MAX_TID_SIZE
76 5401 : && std::all_of(s.begin(), s.end(), [](unsigned char c) { return c >= '0' && c <= '9'; });
77 : }
78 :
79 : /** Characters that are path separators or otherwise special in file names on some platform */
80 : bool
81 118 : isSafeExtension(std::string_view ext)
82 : {
83 118 : if (ext.empty() || ext.size() > fileutils::MAX_EXTENSION_SIZE)
84 98 : return false;
85 20 : return std::none_of(ext.begin(), ext.end(), [](unsigned char c) {
86 60 : return c < 0x20 || c == 0x7f || c == '/' || c == '\\' || c == ':' || c == '*' || c == '?' || c == '"'
87 120 : || c == '<' || c == '>' || c == '|' || c == '.';
88 20 : });
89 : }
90 :
91 : } // namespace
92 :
93 : std::string
94 119 : getFileId(const std::string& commitId, const std::string& tid, const std::string& displayName)
95 : {
96 119 : if (commitId.size() != COMMIT_ID_SIZE || !isLowerHex(commitId) || !isDecimal(tid))
97 11 : return {};
98 108 : auto extension = fileutils::getFileExtension(displayName);
99 108 : if (!isSafeExtension(extension))
100 196 : return fmt::format("{}_{}", commitId, tid);
101 20 : return fmt::format("{}_{}.{}", commitId, tid, extension);
102 : }
103 :
104 : bool
105 471 : isValidFileId(std::string_view fileId) noexcept
106 : {
107 471 : auto sep = fileId.find('_');
108 471 : if (sep != COMMIT_ID_SIZE || !isLowerHex(fileId.substr(0, sep)))
109 6 : return false;
110 465 : auto rest = fileId.substr(sep + 1);
111 465 : auto dot = rest.find('.');
112 465 : if (dot == std::string_view::npos)
113 449 : return isDecimal(rest);
114 16 : return isDecimal(rest.substr(0, dot)) && isSafeExtension(rest.substr(dot + 1));
115 : }
116 :
117 486 : FileInfo::FileInfo(const std::shared_ptr<dhtnet::ChannelSocket>& channel,
118 : const std::string& fileId,
119 : const std::string& interactionId,
120 486 : const libjami::DataTransferInfo& info)
121 486 : : fileId_(fileId)
122 486 : , interactionId_(interactionId)
123 486 : , info_(info)
124 972 : , channel_(channel)
125 486 : {}
126 :
127 : void
128 508 : FileInfo::emit(libjami::DataTransferEventCode code)
129 : {
130 508 : if (finishedCb_ && code >= libjami::DataTransferEventCode::finished)
131 267 : finishedCb_(uint32_t(code));
132 508 : if (interactionId_ != "") {
133 : // Else it's an internal transfer
134 37 : runOnMainThread([info = info_, iid = interactionId_, fid = fileId_, code]() {
135 37 : emitSignal<libjami::DataTransferSignal::DataTransferEvent>(info.accountId,
136 37 : info.conversationId,
137 37 : iid,
138 37 : fid,
139 : uint32_t(code));
140 37 : });
141 : }
142 508 : }
143 :
144 245 : OutgoingFile::OutgoingFile(const std::shared_ptr<dhtnet::ChannelSocket>& channel,
145 : const std::string& fileId,
146 : const std::string& interactionId,
147 : const libjami::DataTransferInfo& info,
148 : size_t start,
149 245 : size_t end)
150 : : FileInfo(channel, fileId, interactionId, info)
151 245 : , start_(start)
152 245 : , end_(end)
153 : {
154 245 : std::filesystem::path fpath(info_.path);
155 245 : if (!std::filesystem::is_regular_file(fpath)) {
156 434 : dht::ThreadPool::io().run([channel = std::move(channel_)] { channel->shutdown(); });
157 217 : return;
158 : }
159 28 : stream_.open(fpath, std::ios::binary | std::ios::in);
160 28 : if (!stream_ || !stream_.is_open()) {
161 0 : dht::ThreadPool::io().run([channel = std::move(channel_)] { channel->shutdown(); });
162 0 : return;
163 : }
164 245 : }
165 :
166 245 : OutgoingFile::~OutgoingFile()
167 : {
168 245 : if (stream_ && stream_.is_open())
169 0 : stream_.close();
170 245 : if (channel_) {
171 56 : dht::ThreadPool::io().run([channel = std::move(channel_)] { channel->shutdown(); });
172 : }
173 245 : }
174 :
175 : void
176 245 : OutgoingFile::process()
177 : {
178 245 : if (!channel_ or !stream_ or !stream_.is_open())
179 217 : return;
180 28 : auto correct = false;
181 28 : stream_.seekg(static_cast<long>(start_), std::ios::beg);
182 : try {
183 28 : std::vector<char> buffer(UINT16_MAX, 0);
184 28 : std::error_code ec;
185 28 : auto pos = start_;
186 262 : while (!stream_.eof()) {
187 237 : stream_.read(buffer.data(),
188 237 : end_ > start_ ? static_cast<long>(std::min(end_ - pos, buffer.size()))
189 237 : : static_cast<long>(buffer.size()));
190 237 : auto gcount = stream_.gcount();
191 237 : pos += gcount;
192 237 : channel_->write(reinterpret_cast<const uint8_t*>(buffer.data()), gcount, ec);
193 237 : if (ec)
194 3 : break;
195 : }
196 28 : if (!ec)
197 25 : correct = true;
198 28 : stream_.close();
199 28 : } catch (const std::exception& e) {
200 0 : JAMI_WARNING("Failed to read from stream: {}", e.what());
201 0 : }
202 28 : if (!isUserCancelled_) {
203 : // NOTE: emit(code) MUST be changed to improve handling of multiple destinations
204 : // But for now, we can just avoid to emit errors to the client, because for outgoing
205 : // transfer in a swarm, for outgoingFiles, we know that the file is ok. And the peer
206 : // will retry the transfer if they need, so we don't need to show errors.
207 28 : if (!interactionId_.empty() && !correct)
208 2 : return;
209 26 : auto code = correct ? libjami::DataTransferEventCode::finished : libjami::DataTransferEventCode::closed_by_peer;
210 26 : emit(code);
211 : }
212 : }
213 :
214 : void
215 0 : OutgoingFile::cancel()
216 : {
217 : // Remove link, not original file
218 0 : auto path = fileutils::get_data_dir() / "conversation_data" / info_.accountId / info_.conversationId / fileId_;
219 0 : if (std::filesystem::is_symlink(path))
220 0 : dhtnet::fileutils::remove(path);
221 0 : isUserCancelled_ = true;
222 0 : emit(libjami::DataTransferEventCode::closed_by_host);
223 0 : }
224 :
225 241 : IncomingFile::IncomingFile(const std::shared_ptr<dhtnet::ChannelSocket>& channel,
226 : const libjami::DataTransferInfo& info,
227 : const std::string& fileId,
228 : const std::string& interactionId,
229 : const std::string& sha3Sum,
230 241 : const std::filesystem::path& temporaryPath)
231 : : FileInfo(channel, fileId, interactionId, info)
232 241 : , sha3Sum_(sha3Sum)
233 482 : , path_(temporaryPath)
234 : {
235 241 : stream_.open(path_, std::ios::binary | std::ios::out | std::ios::app);
236 241 : if (!stream_)
237 0 : return;
238 :
239 241 : emit(libjami::DataTransferEventCode::ongoing);
240 0 : }
241 :
242 241 : IncomingFile::~IncomingFile()
243 : {
244 : {
245 241 : std::lock_guard<std::mutex> lk(streamMtx_);
246 241 : if (stream_ && stream_.is_open())
247 0 : stream_.close();
248 241 : }
249 241 : if (channel_)
250 480 : dht::ThreadPool::io().run([channel = std::move(channel_)] { channel->shutdown(); });
251 241 : }
252 :
253 : void
254 1 : IncomingFile::cancel()
255 : {
256 1 : isUserCancelled_ = true;
257 : {
258 1 : std::lock_guard<std::mutex> lk(streamMtx_);
259 1 : if (stream_.is_open())
260 1 : stream_.close();
261 1 : }
262 1 : std::error_code ec;
263 1 : std::filesystem::remove(path_, ec);
264 1 : if (ec)
265 0 : JAMI_WARNING("Unable to remove canceled partial file {}: {}", path_, ec.message());
266 1 : emit(libjami::DataTransferEventCode::closed_by_peer);
267 1 : if (channel_)
268 2 : dht::ThreadPool::io().run([channel = std::move(channel_)] { channel->shutdown(); });
269 1 : }
270 :
271 : void
272 241 : IncomingFile::process()
273 : {
274 241 : if (!stream_.is_open()) {
275 0 : emit(libjami::DataTransferEventCode::invalid_pathname);
276 0 : if (channel_)
277 0 : dht::ThreadPool::io().run([channel = std::move(channel_)] { channel->shutdown(); });
278 0 : return;
279 : }
280 241 : channel_->setOnRecv([w = weak_from_this()](const uint8_t* buf, size_t len) {
281 225 : if (auto shared = w.lock()) {
282 224 : std::lock_guard<std::mutex> lk(shared->streamMtx_);
283 224 : if (!shared->stream_.is_open())
284 0 : return -1;
285 224 : shared->stream_.write(reinterpret_cast<const char*>(buf), static_cast<long>(len));
286 224 : if (!shared->stream_)
287 0 : return -1;
288 224 : shared->info_.bytesProgress = shared->stream_.tellp();
289 224 : return static_cast<int>(len);
290 449 : }
291 : // Data received after destruction
292 1 : JAMI_ERROR("{} bytes received after IncomingFile destruction.", len);
293 1 : return -1;
294 : });
295 241 : channel_->onShutdown([w = weak_from_this()](const std::error_code& /*error_code*/) {
296 241 : auto shared = w.lock();
297 241 : if (!shared)
298 1 : return;
299 : {
300 240 : std::lock_guard<std::mutex> lk(shared->streamMtx_);
301 240 : if (shared->stream_ && shared->stream_.is_open())
302 240 : shared->stream_.close();
303 240 : }
304 240 : auto correct = shared->sha3Sum_.empty();
305 240 : std::error_code ec;
306 240 : if (!correct) {
307 17 : if (shared->isUserCancelled_) {
308 0 : std::filesystem::remove(shared->path_, ec);
309 17 : } else if (shared->info_.bytesProgress < shared->info_.totalSize) {
310 1 : JAMI_WARNING("Channel for {} shut down before transfer was complete (progress: {}/{})",
311 : shared->info_.path,
312 : shared->info_.bytesProgress,
313 : shared->info_.totalSize);
314 16 : } else if (shared->info_.totalSize != 0 && shared->info_.bytesProgress > shared->info_.totalSize) {
315 0 : JAMI_WARNING("Removing {} larger than announced: {}/{}",
316 : shared->path_,
317 : shared->info_.bytesProgress,
318 : shared->info_.totalSize);
319 0 : std::filesystem::remove(shared->path_, ec);
320 : } else {
321 16 : auto sha3Sum = fileutils::sha3File(shared->path_);
322 16 : if (shared->sha3Sum_ == sha3Sum) {
323 15 : JAMI_LOG("New file received: {}", shared->info_.path);
324 15 : correct = true;
325 : } else {
326 1 : JAMI_WARNING(
327 : "Removing {} with expected size ({} bytes) but invalid sha3sum (expected: {}, actual: {})",
328 : shared->path_,
329 : shared->info_.totalSize,
330 : shared->sha3Sum_,
331 : sha3Sum);
332 1 : std::filesystem::remove(shared->path_, ec);
333 : }
334 16 : }
335 17 : if (ec) {
336 0 : JAMI_ERROR("Failed to remove file {}: {}", shared->path_, ec.message());
337 : }
338 : }
339 240 : auto installed = false;
340 240 : if (correct) {
341 238 : if (shared->installCb_) {
342 10 : installed = shared->installCb_(shared->path_);
343 : } else {
344 228 : std::filesystem::rename(shared->path_, shared->info_.path, ec);
345 228 : installed = !ec;
346 228 : if (ec)
347 0 : JAMI_ERROR("Failed to rename file from {} to {}: {}",
348 : shared->path_,
349 : shared->info_.path,
350 : ec.message());
351 : }
352 : }
353 240 : if (shared->isUserCancelled_)
354 0 : return;
355 478 : auto code = !correct ? libjami::DataTransferEventCode::closed_by_host
356 238 : : installed ? libjami::DataTransferEventCode::finished
357 : : libjami::DataTransferEventCode::invalid_pathname;
358 240 : shared->emit(code);
359 240 : dht::ThreadPool::io().run([s = std::move(shared)] {});
360 241 : });
361 : }
362 :
363 : //==============================================================================
364 :
365 : class TransferManager::Impl
366 : {
367 : public:
368 1778 : Impl(const std::string& accountId, const std::string& accountUri, const std::string& to, const std::mt19937_64& rand)
369 1778 : : accountId_(accountId)
370 1778 : , accountUri_(accountUri)
371 1778 : , to_(to)
372 1778 : , rand_(rand)
373 : {
374 1778 : if (!to_.empty()) {
375 916 : conversationDataPath_ = fileutils::get_data_dir() / accountId_ / "conversation_data" / to_;
376 916 : dhtnet::fileutils::check_dir(conversationDataPath_);
377 916 : waitingPath_ = conversationDataPath_ / "waiting";
378 : }
379 1778 : profilesPath_ = fileutils::get_data_dir() / accountId_ / "profiles";
380 1778 : accountProfilePath_ = fileutils::get_data_dir() / accountId / "profile.vcf";
381 1778 : loadWaiting();
382 1778 : }
383 :
384 1778 : ~Impl()
385 : {
386 1778 : std::lock_guard lk {mapMutex_};
387 1997 : for (auto& [channel, _] : outgoings_) {
388 438 : dht::ThreadPool::io().run([c = std::move(channel)] { c->shutdown(); });
389 : }
390 1778 : outgoings_.clear();
391 1778 : incomings_.clear();
392 1778 : vcards_.clear();
393 1778 : }
394 :
395 1778 : void loadWaiting()
396 : {
397 : try {
398 : // read file
399 3551 : auto file = fileutils::loadFile(waitingPath_);
400 : // load values
401 5 : msgpack::object_handle oh = msgpack::unpack((const char*) file.data(), file.size());
402 5 : std::lock_guard lk {mapMutex_};
403 5 : oh.get().convert(waitingIds_);
404 5 : auto changed = false;
405 7 : for (auto it = waitingIds_.begin(); it != waitingIds_.end();) {
406 2 : const auto& request = it->second;
407 2 : auto destination = std::filesystem::path(request.path);
408 4 : if (it->first != request.fileId || !isValidFileId(request.fileId)
409 4 : || (!destination.empty() && destination.is_relative())) {
410 0 : it = waitingIds_.erase(it);
411 0 : changed = true;
412 : } else {
413 2 : ++it;
414 : }
415 2 : }
416 5 : if (changed)
417 0 : saveWaiting();
418 1778 : } catch (const std::exception& e) {
419 1773 : return;
420 1773 : }
421 : }
422 98 : void saveWaiting()
423 : {
424 98 : std::ofstream file(waitingPath_, std::ios::trunc | std::ios::binary);
425 98 : msgpack::pack(file, waitingIds_);
426 98 : }
427 :
428 : std::string accountId_ {};
429 : std::string accountUri_ {};
430 : std::string to_ {};
431 : std::filesystem::path waitingPath_ {};
432 : std::filesystem::path profilesPath_ {};
433 : std::filesystem::path accountProfilePath_ {};
434 : std::filesystem::path conversationDataPath_ {};
435 :
436 : std::mutex mapMutex_ {};
437 : std::map<std::string, std::weak_ptr<std::mutex>> waitingMutexes_ {};
438 : std::map<std::string, WaitingRequest> waitingIds_ {};
439 : std::map<std::shared_ptr<dhtnet::ChannelSocket>, std::shared_ptr<OutgoingFile>> outgoings_ {};
440 : std::map<std::string, std::shared_ptr<IncomingFile>> incomings_ {};
441 : std::map<std::pair<std::string, std::string>, std::shared_ptr<IncomingFile>> vcards_ {};
442 :
443 : std::mt19937_64 rand_;
444 : };
445 :
446 1778 : TransferManager::TransferManager(const std::string& accountId,
447 : const std::string& accountUri,
448 : const std::string& to,
449 1778 : const std::mt19937_64& rand)
450 1778 : : pimpl_ {std::make_unique<Impl>(accountId, accountUri, to, rand)}
451 1778 : {}
452 :
453 1778 : TransferManager::~TransferManager() {}
454 :
455 : void
456 247 : TransferManager::transferFile(const std::shared_ptr<dhtnet::ChannelSocket>& channel,
457 : const std::string& fileId,
458 : const std::string& interactionId,
459 : const std::string& path,
460 : size_t start,
461 : size_t end,
462 : OnFinishedCb onFinished)
463 : {
464 247 : std::lock_guard lk {pimpl_->mapMutex_};
465 247 : if (pimpl_->outgoings_.find(channel) != pimpl_->outgoings_.end())
466 2 : return;
467 245 : libjami::DataTransferInfo info;
468 245 : info.accountId = pimpl_->accountId_;
469 245 : info.conversationId = pimpl_->to_;
470 245 : info.path = path;
471 245 : auto f = std::make_shared<OutgoingFile>(channel, fileId, interactionId, info, start, end);
472 245 : f->onFinished([w = weak(), channel, onFinished = std::move(onFinished)](uint32_t code) {
473 26 : if (code == uint32_t(libjami::DataTransferEventCode::finished) && onFinished) {
474 4 : onFinished();
475 : }
476 : // schedule destroy outgoing transfer as not needed
477 26 : dht::ThreadPool().computation().run([w, channel] {
478 26 : if (auto sthis_ = w.lock()) {
479 26 : auto& pimpl = sthis_->pimpl_;
480 26 : std::lock_guard lk {pimpl->mapMutex_};
481 26 : auto itO = pimpl->outgoings_.find(channel);
482 26 : if (itO != pimpl->outgoings_.end())
483 26 : pimpl->outgoings_.erase(itO);
484 52 : }
485 26 : });
486 26 : });
487 245 : auto [outFile, _] = pimpl_->outgoings_.emplace(channel, std::move(f));
488 245 : dht::ThreadPool::io().run([w = std::weak_ptr<OutgoingFile>(outFile->second)] {
489 245 : if (auto of = w.lock())
490 245 : of->process();
491 245 : });
492 247 : }
493 :
494 : bool
495 36 : TransferManager::cancel(const std::string& fileId)
496 : {
497 36 : std::shared_ptr<IncomingFile> incoming;
498 36 : std::filesystem::path partialPath;
499 36 : auto canceled = false;
500 : {
501 36 : std::lock_guard lk {pimpl_->mapMutex_};
502 : // Remove from waiting, this avoid auto-download
503 36 : auto itW = pimpl_->waitingIds_.find(fileId);
504 36 : if (itW != pimpl_->waitingIds_.end()) {
505 36 : auto destination = std::filesystem::path(itW->second.path);
506 36 : if (destination.empty())
507 0 : partialPath = temporaryPath(fileId, path(fileId));
508 36 : else if (destination.is_absolute())
509 36 : partialPath = temporaryPath(fileId, destination);
510 36 : pimpl_->waitingIds_.erase(itW);
511 36 : JAMI_LOG("Cancel {}", fileId);
512 36 : pimpl_->saveWaiting();
513 36 : canceled = true;
514 36 : }
515 36 : auto itC = pimpl_->incomings_.find(fileId);
516 36 : if (itC != pimpl_->incomings_.end())
517 1 : incoming = itC->second;
518 36 : }
519 36 : if (incoming) {
520 1 : incoming->cancel();
521 1 : return true;
522 : }
523 35 : if (!partialPath.empty()) {
524 35 : std::error_code ec;
525 35 : std::filesystem::remove(partialPath, ec);
526 35 : if (ec)
527 0 : JAMI_WARNING("Unable to remove canceled partial file {}: {}", partialPath, ec.message());
528 : }
529 35 : return canceled;
530 36 : }
531 :
532 : bool
533 3 : TransferManager::info(const std::string& fileId, std::string& path, int64_t& total, int64_t& progress) const noexcept
534 : {
535 3 : std::unique_lock lk {pimpl_->mapMutex_};
536 3 : if (pimpl_->to_.empty())
537 0 : return false;
538 :
539 3 : auto itI = pimpl_->incomings_.find(fileId);
540 3 : auto itW = pimpl_->waitingIds_.find(fileId);
541 3 : std::filesystem::path transferPath;
542 : try {
543 3 : transferPath = this->path(fileId);
544 3 : path = transferPath.string();
545 0 : } catch (const std::filesystem::filesystem_error& e) {
546 0 : JAMI_WARNING("Unable to resolve transfer path for {}: {}", fileId, e.what());
547 0 : progress = 0;
548 0 : return false;
549 0 : }
550 3 : if (itI != pimpl_->incomings_.end()) {
551 0 : total = itI->second->info().totalSize;
552 0 : progress = itI->second->info().bytesProgress;
553 0 : return true;
554 : }
555 :
556 3 : std::error_code ec;
557 3 : if (std::filesystem::is_regular_file(transferPath, ec)) {
558 1 : auto fileSize = std::filesystem::file_size(transferPath, ec);
559 1 : if (ec) {
560 0 : JAMI_WARNING("Unable to read transfer file size for {}: {}", path, ec.message());
561 0 : progress = 0;
562 0 : return false;
563 : }
564 1 : progress = static_cast<int64_t>(
565 1 : std::min<uintmax_t>(fileSize, static_cast<uintmax_t>(std::numeric_limits<int64_t>::max())));
566 1 : if (itW != pimpl_->waitingIds_.end()) {
567 0 : total = static_cast<int64_t>(itW->second.totalSize);
568 : } else {
569 : // If not waiting it's finished
570 1 : total = progress;
571 : }
572 1 : return true;
573 : }
574 2 : if (ec && ec != std::errc::no_such_file_or_directory) {
575 1 : JAMI_WARNING("Unable to inspect transfer path {}: {}", path, ec.message());
576 : }
577 2 : if (itW != pimpl_->waitingIds_.end()) {
578 0 : total = static_cast<int64_t>(itW->second.totalSize);
579 0 : progress = 0;
580 0 : return true;
581 : }
582 : // Else we don't know infos there.
583 2 : progress = 0;
584 2 : return false;
585 3 : }
586 :
587 : bool
588 24 : TransferManager::indexFile(const std::string& fileId,
589 : const std::filesystem::path& candidate,
590 : const std::string& sha3sum,
591 : std::size_t total,
592 : bool independent)
593 : {
594 24 : return installIndex(fileId, candidate, sha3sum, total, independent, false);
595 : }
596 :
597 : bool
598 84 : TransferManager::installIndex(const std::string& fileId,
599 : const std::filesystem::path& candidate,
600 : const std::string& sha3sum,
601 : std::size_t total,
602 : bool independent,
603 : bool verifyCandidate)
604 : {
605 84 : if (!isValidFileId(fileId)) {
606 0 : JAMI_WARNING("Refusing to index file transfer with invalid id '{}'", fileId);
607 0 : return false;
608 : }
609 84 : auto canonicalPath = path(fileId);
610 : {
611 84 : std::lock_guard fileLock(dhtnet::fileutils::getFileLock(canonicalPath));
612 84 : std::error_code ec;
613 84 : auto canonicalStatus = std::filesystem::symlink_status(canonicalPath, ec);
614 84 : auto canonicalMissing = isMissingPath(canonicalStatus, ec);
615 84 : auto canonicalIsLink = canonicalStatus.type() == std::filesystem::file_type::symlink;
616 84 : auto canonicalIsValid = !canonicalMissing && !ec && isExpectedFile(canonicalPath, sha3sum, total);
617 84 : if (!canonicalIsValid || (independent && canonicalIsLink)) {
618 82 : if (candidate == canonicalPath || (verifyCandidate && !isExpectedFile(candidate, sha3sum, total)))
619 56 : return false;
620 : // Only a missing or stale link may be replaced; existing content is preserved.
621 26 : if (!canonicalMissing && (ec || !canonicalIsLink)) {
622 0 : JAMI_WARNING("Refusing to replace existing file transfer index {}", canonicalPath);
623 0 : return false;
624 : }
625 : libjami::DataTransferId stagingId;
626 : {
627 26 : std::lock_guard lk {pimpl_->mapMutex_};
628 26 : stagingId = generateUID(pimpl_->rand_);
629 26 : }
630 52 : auto stagedPath = canonicalPath.parent_path() / fmt::format(".jami-index-{}-{}.tmp", fileId, stagingId);
631 26 : auto stagedStatus = std::filesystem::symlink_status(stagedPath, ec);
632 26 : if (!isMissingPath(stagedStatus, ec)) {
633 0 : JAMI_WARNING("Refusing to replace occupied file transfer staging path {}", stagedPath);
634 0 : return false;
635 : }
636 26 : if (independent)
637 15 : std::filesystem::create_hard_link(candidate, stagedPath, ec);
638 : else
639 11 : std::filesystem::create_symlink(candidate, stagedPath, ec);
640 26 : if (ec) {
641 0 : JAMI_WARNING("Unable to link file transfer {} at {}, copying it: {}", fileId, stagedPath, ec.message());
642 0 : std::filesystem::copy_file(candidate, stagedPath, ec);
643 : // A link shares the verified content; only a copy can differ from it.
644 0 : if (!ec && !isExpectedFile(stagedPath, sha3sum, total))
645 0 : ec = std::make_error_code(std::errc::io_error);
646 0 : if (ec) {
647 0 : JAMI_ERROR("Unable to copy file transfer {} to {}: {}", fileId, stagedPath, ec.message());
648 0 : std::error_code removeError;
649 0 : std::filesystem::remove(stagedPath, removeError);
650 0 : return false;
651 : }
652 : }
653 26 : std::filesystem::rename(stagedPath, canonicalPath, ec);
654 26 : if (ec) {
655 0 : JAMI_ERROR("Unable to install file transfer index {} at {}: {}", fileId, canonicalPath, ec.message());
656 0 : std::error_code removeError;
657 0 : std::filesystem::remove(stagedPath, removeError);
658 0 : return false;
659 : }
660 26 : }
661 84 : }
662 :
663 28 : std::lock_guard lk {pimpl_->mapMutex_};
664 28 : if (pimpl_->waitingIds_.erase(fileId) != 0)
665 11 : pimpl_->saveWaiting();
666 28 : return true;
667 84 : }
668 :
669 : bool
670 3 : TransferManager::exportFile(const std::string& fileId,
671 : const std::filesystem::path& destination,
672 : const std::string& sha3sum,
673 : std::size_t total)
674 : {
675 3 : std::lock_guard fileLock(dhtnet::fileutils::getFileLock(destination));
676 3 : std::error_code ec;
677 3 : if (std::filesystem::equivalent(path(fileId), destination, ec) || isExpectedFile(destination, sha3sum, total))
678 2 : return true;
679 1 : auto status = std::filesystem::symlink_status(destination, ec);
680 1 : if (!isMissingPath(status, ec)) {
681 0 : JAMI_WARNING("Refusing to overwrite existing file {}", destination);
682 0 : return false;
683 : }
684 1 : auto source = std::filesystem::canonical(path(fileId), ec);
685 1 : if (ec) {
686 0 : JAMI_ERROR("Unable to resolve file transfer index {}: {}", fileId, ec.message());
687 0 : return false;
688 : }
689 1 : std::filesystem::create_hard_link(source, destination, ec);
690 1 : if (ec) {
691 0 : std::filesystem::copy_file(source, destination, ec);
692 0 : if (ec) {
693 0 : JAMI_ERROR("Unable to export file transfer {} to {}: {}", fileId, destination, ec.message());
694 0 : std::error_code removeError;
695 0 : std::filesystem::remove(destination, removeError);
696 0 : return false;
697 : }
698 : }
699 1 : return true;
700 3 : }
701 :
702 : bool
703 10 : TransferManager::installTransfer(const std::string& fileId,
704 : const std::filesystem::path& partial,
705 : const std::filesystem::path& destination,
706 : const std::string& sha3sum,
707 : std::size_t total)
708 : {
709 10 : const auto isIndex = destination.lexically_normal() == path(fileId).lexically_normal();
710 10 : std::lock_guard fileLock(dhtnet::fileutils::getFileLock(destination));
711 10 : std::error_code ec;
712 10 : auto status = std::filesystem::symlink_status(destination, ec);
713 10 : auto replaceable = isMissingPath(status, ec);
714 10 : if (!replaceable) {
715 0 : if (isExpectedFile(destination, sha3sum, total)) {
716 : // Redundant download: the destination already holds the file.
717 0 : std::filesystem::remove(partial, ec);
718 0 : return isIndex || indexFile(fileId, destination, sha3sum, total);
719 : }
720 : // Only a stale index link may be replaced; existing content is preserved.
721 0 : replaceable = isIndex && status.type() == std::filesystem::file_type::symlink;
722 : }
723 10 : if (!replaceable) {
724 0 : JAMI_WARNING("Refusing to overwrite existing file {}", destination);
725 0 : std::filesystem::remove(partial, ec);
726 0 : return false;
727 : }
728 10 : std::filesystem::rename(partial, destination, ec);
729 10 : if (ec) {
730 0 : JAMI_ERROR("Unable to install file transfer {} at {}: {}", fileId, destination, ec.message());
731 0 : std::filesystem::remove(partial, ec);
732 0 : return false;
733 : }
734 10 : if (isIndex || indexFile(fileId, destination, sha3sum, total))
735 10 : return true;
736 0 : std::filesystem::remove(destination, ec);
737 0 : return false;
738 10 : }
739 :
740 : TransferManager::WaitResult
741 285 : TransferManager::waitForTransfer(const std::string& fileId,
742 : const std::string& interactionId,
743 : const std::string& sha3sum,
744 : const std::string& path,
745 : std::size_t total)
746 : {
747 285 : if (!isValidFileId(fileId)) {
748 0 : JAMI_WARNING("Refusing to wait for file transfer with invalid id '{}'", fileId);
749 0 : return WaitResult::conflict;
750 : }
751 285 : if (!path.empty() && std::filesystem::path(path).is_relative())
752 0 : return WaitResult::conflict;
753 285 : auto canonicalPath = this->path(fileId);
754 285 : auto destination = path.empty() ? canonicalPath : std::filesystem::path(path);
755 285 : const auto isIndex = destination.lexically_normal() == canonicalPath.lexically_normal();
756 285 : std::shared_ptr<std::mutex> waitingMutex;
757 : {
758 285 : std::lock_guard lk(pimpl_->mapMutex_);
759 551 : std::erase_if(pimpl_->waitingMutexes_, [](const auto& entry) { return entry.second.expired(); });
760 285 : auto& mutex = pimpl_->waitingMutexes_[fileId];
761 285 : waitingMutex = mutex.lock();
762 285 : if (!waitingMutex) {
763 131 : waitingMutex = std::make_shared<std::mutex>();
764 131 : mutex = waitingMutex;
765 : }
766 285 : }
767 285 : std::lock_guard waitingLock(*waitingMutex);
768 : {
769 285 : std::lock_guard lk(pimpl_->mapMutex_);
770 285 : auto itW = pimpl_->waitingIds_.find(fileId);
771 285 : if (itW != pimpl_->waitingIds_.end()) {
772 231 : auto waited = itW->second.path.empty() ? canonicalPath : std::filesystem::path(itW->second.path);
773 462 : auto matches = itW->second.interactionId == interactionId && itW->second.sha3sum == sha3sum
774 462 : && waited.lexically_normal() == destination.lexically_normal()
775 693 : && itW->second.totalSize == total;
776 231 : if (!matches)
777 225 : return WaitResult::conflict;
778 6 : if (pimpl_->incomings_.contains(fileId))
779 0 : return WaitResult::waiting;
780 231 : }
781 285 : }
782 60 : if (installIndex(fileId, destination, sha3sum, total, false, true))
783 4 : return (isIndex || exportFile(fileId, destination, sha3sum, total)) ? WaitResult::complete
784 4 : : WaitResult::conflict;
785 56 : std::lock_guard lk(pimpl_->mapMutex_);
786 56 : if (pimpl_->waitingIds_.contains(fileId))
787 4 : return WaitResult::waiting;
788 52 : std::error_code ec;
789 52 : auto status = std::filesystem::symlink_status(destination, ec);
790 52 : if (!isMissingPath(status, ec) && !(isIndex && status.type() == std::filesystem::file_type::symlink)) {
791 2 : JAMI_WARNING("Refusing to overwrite existing file {}", destination);
792 2 : return WaitResult::conflict;
793 : }
794 : // A partial file left by a previous attempt is resumed by the transfer.
795 50 : pimpl_->waitingIds_[fileId] = {fileId, interactionId, sha3sum, path, total};
796 50 : pimpl_->saveWaiting();
797 50 : return WaitResult::waiting;
798 335 : }
799 :
800 : void
801 13 : TransferManager::onIncomingFileTransfer(const std::string& fileId,
802 : const std::shared_ptr<dhtnet::ChannelSocket>& channel,
803 : size_t start)
804 : {
805 13 : if (!isValidFileId(fileId)) {
806 0 : dht::ThreadPool().io().run([channel] { channel->shutdown(); });
807 0 : return;
808 : }
809 13 : std::unique_lock lk(pimpl_->mapMutex_);
810 : // Check if not already an incoming file for this id and that we are waiting this file
811 13 : auto itC = pimpl_->incomings_.find(fileId);
812 13 : if (itC != pimpl_->incomings_.end()) {
813 0 : dht::ThreadPool().io().run([channel] { channel->shutdown(); });
814 0 : return;
815 : }
816 13 : auto itW = pimpl_->waitingIds_.find(fileId);
817 13 : if (itW == pimpl_->waitingIds_.end()) {
818 0 : dht::ThreadPool().io().run([channel] { channel->shutdown(); });
819 0 : return;
820 : }
821 :
822 13 : libjami::DataTransferInfo info;
823 13 : info.accountId = pimpl_->accountId_;
824 13 : info.conversationId = pimpl_->to_;
825 13 : info.path = itW->second.path;
826 13 : info.totalSize = static_cast<int64_t>(itW->second.totalSize);
827 13 : info.bytesProgress = static_cast<int64_t>(start);
828 :
829 : // Receive into a private partial file next to the destination; the destination and the
830 : // index entry are only touched once the content has been verified.
831 13 : if (info.path.empty())
832 1 : info.path = path(fileId).string();
833 13 : const std::filesystem::path destinationPath(info.path);
834 13 : const auto expectedSize = itW->second.totalSize;
835 13 : const auto expectedSha3 = itW->second.sha3sum;
836 :
837 13 : auto ifile = std::make_shared<IncomingFile>(std::move(channel),
838 : info,
839 : fileId,
840 13 : itW->second.interactionId,
841 : expectedSha3,
842 26 : temporaryPath(fileId, destinationPath));
843 13 : auto res = pimpl_->incomings_.emplace(fileId, std::move(ifile));
844 13 : if (res.second) {
845 26 : res.first->second->onInstall(
846 26 : [w = weak(), fileId, destinationPath, expectedSha3, expectedSize](const std::filesystem::path& partial) {
847 10 : if (auto sthis = w.lock())
848 10 : return sthis->installTransfer(fileId, partial, destinationPath, expectedSha3, expectedSize);
849 0 : return false;
850 : });
851 13 : res.first->second->onFinished([w = weak(), fileId](uint32_t code) {
852 13 : if (auto sthis = w.lock()) {
853 13 : auto& pimpl = sthis->pimpl_;
854 13 : std::lock_guard lk {pimpl->mapMutex_};
855 13 : pimpl->incomings_.erase(fileId);
856 13 : if ((code == uint32_t(libjami::DataTransferEventCode::finished)
857 3 : || code == uint32_t(libjami::DataTransferEventCode::invalid_pathname))
858 16 : && pimpl->waitingIds_.erase(fileId) != 0) {
859 1 : pimpl->saveWaiting();
860 : }
861 26 : }
862 13 : });
863 13 : auto incoming = res.first->second;
864 13 : lk.unlock();
865 13 : incoming->process();
866 13 : }
867 13 : }
868 :
869 : std::filesystem::path
870 424 : TransferManager::path(const std::string& fileId) const
871 : {
872 424 : return pimpl_->conversationDataPath_ / fileId;
873 : }
874 :
875 : std::filesystem::path
876 68 : TransferManager::temporaryPath(const std::string& fileId, const std::filesystem::path& destination) const
877 : {
878 136 : return destination.parent_path() / fmt::format(".jami-{}-{}-{}.tmp", pimpl_->accountId_, pimpl_->to_, fileId);
879 : }
880 :
881 : void
882 232 : TransferManager::onIncomingProfile(const std::shared_ptr<dhtnet::ChannelSocket>& channel, const std::string& sha3Sum)
883 : {
884 232 : if (!channel)
885 0 : return;
886 :
887 232 : auto chName = channel->name();
888 232 : std::string_view name = chName;
889 232 : auto sep = name.find_last_of('?');
890 232 : if (sep != std::string::npos)
891 6 : name = name.substr(0, sep);
892 :
893 232 : auto lastSep = name.find_last_of('/');
894 232 : auto fileId = name.substr(lastSep + 1);
895 :
896 232 : auto deviceId = channel->deviceId().toString();
897 232 : auto cert = channel->peerCertificate();
898 232 : if (!cert || !cert->issuer || fileId.find(".vcf") == std::string::npos)
899 3 : return;
900 :
901 235 : auto uri = fileId == "profile.vcf" ? cert->issuer->getId().toString()
902 458 : : std::string(fileId.substr(0, fileId.size() - 4 /*.vcf*/));
903 :
904 229 : std::lock_guard lk(pimpl_->mapMutex_);
905 229 : auto idx = std::make_pair(deviceId, uri);
906 : // Check if not already an incoming file for this id and that we are waiting this file
907 229 : auto itV = pimpl_->vcards_.find(idx);
908 229 : if (itV != pimpl_->vcards_.end()) {
909 2 : dht::ThreadPool().io().run([channel] { channel->shutdown(); });
910 1 : return;
911 : }
912 :
913 228 : auto tid = generateUID(pimpl_->rand_);
914 228 : libjami::DataTransferInfo info;
915 228 : info.accountId = pimpl_->accountId_;
916 228 : info.conversationId = pimpl_->to_;
917 :
918 228 : auto recvDir = fileutils::get_cache_dir() / pimpl_->accountId_ / "vcard";
919 228 : dhtnet::fileutils::recursive_mkdir(recvDir);
920 456 : info.path = (recvDir / fmt::format("{:s}_{:s}_{}", deviceId, uri, tid)).string();
921 :
922 228 : auto ifile = std::make_shared<IncomingFile>(std::move(channel),
923 : info,
924 : "profile.vcf",
925 : "",
926 : sha3Sum,
927 456 : std::filesystem::path(info.path + ".tmp"));
928 228 : auto res = pimpl_->vcards_.emplace(idx, std::move(ifile));
929 228 : if (res.second) {
930 912 : res.first->second->onFinished([w = weak(),
931 228 : uri = std::move(uri),
932 228 : deviceId = std::move(deviceId),
933 228 : accountId = pimpl_->accountId_,
934 228 : cert = std::move(cert),
935 : path = info.path](uint32_t code) {
936 1140 : dht::ThreadPool().computation().run([w,
937 228 : uri = std::move(uri),
938 228 : deviceId = std::move(deviceId),
939 228 : accountId = std::move(accountId),
940 228 : path = std::move(path),
941 : code] {
942 228 : if (auto sthis_ = w.lock()) {
943 228 : auto& pimpl = sthis_->pimpl_;
944 :
945 228 : auto destPath = sthis_->profilePath(uri);
946 : try {
947 : // Move profile to destination path
948 228 : std::lock_guard lock(dhtnet::fileutils::getFileLock(destPath));
949 228 : dhtnet::fileutils::recursive_mkdir(destPath.parent_path());
950 228 : std::filesystem::rename(path, destPath);
951 228 : if (!pimpl->accountUri_.empty() && uri == pimpl->accountUri_) {
952 : // If this is the account profile, link or copy it to the account profile path
953 3 : if (!fileutils::createFileLink(pimpl->accountProfilePath_, destPath)) {
954 0 : std::error_code ec;
955 0 : std::filesystem::copy_file(destPath, pimpl->accountProfilePath_, ec);
956 : }
957 : }
958 228 : } catch (const std::exception& e) {
959 0 : JAMI_ERROR("{}", e.what());
960 0 : }
961 :
962 228 : std::lock_guard lk {pimpl->mapMutex_};
963 228 : auto itO = pimpl->vcards_.find({deviceId, uri});
964 228 : if (itO != pimpl->vcards_.end())
965 228 : pimpl->vcards_.erase(itO);
966 228 : if (code == uint32_t(libjami::DataTransferEventCode::finished)) {
967 228 : emitSignal<libjami::ConfigurationSignal::ProfileReceived>(accountId, uri, destPath.string());
968 : }
969 456 : }
970 228 : });
971 228 : });
972 228 : res.first->second->process();
973 : }
974 243 : }
975 :
976 : std::filesystem::path
977 243 : TransferManager::profilePath(const std::string& contactId) const
978 : {
979 486 : return pimpl_->profilesPath_ / fmt::format("{}.vcf", base64::encode(contactId));
980 : }
981 :
982 : std::vector<WaitingRequest>
983 1892 : TransferManager::waitingRequests() const
984 : {
985 1892 : std::vector<WaitingRequest> res;
986 1892 : std::lock_guard lk(pimpl_->mapMutex_);
987 1925 : for (const auto& [fileId, req] : pimpl_->waitingIds_) {
988 33 : auto itC = pimpl_->incomings_.find(fileId);
989 33 : if (itC == pimpl_->incomings_.end())
990 33 : res.emplace_back(req);
991 : }
992 3784 : return res;
993 1892 : }
994 :
995 : bool
996 16 : TransferManager::isWaiting(const std::string& fileId) const
997 : {
998 16 : std::lock_guard lk(pimpl_->mapMutex_);
999 32 : return pimpl_->waitingIds_.find(fileId) != pimpl_->waitingIds_.end();
1000 16 : }
1001 :
1002 : } // namespace jami
|
