diff --git a/CMakeLists.txt b/CMakeLists.txt index 2c0a9047..69c6911a 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -502,7 +502,11 @@ list (APPEND opendht_HEADERS ) if (OPENDHT_PEER_DISCOVERY) - list (APPEND opendht_SOURCES src/mdns.cpp src/dns_sd.cpp src/mdns_transport.cpp src/peer_discovery.cpp) + list (APPEND opendht_SOURCES + src/peer_discovery/mdns.cpp + src/peer_discovery/dns_sd.cpp + src/peer_discovery/mdns_transport.cpp + src/peer_discovery/peer_discovery.cpp) list (APPEND opendht_HEADERS include/opendht/peer_discovery.h) add_definitions(-DOPENDHT_PEER_DISCOVERY) endif() diff --git a/meson.build b/meson.build index d53a5051..221c167f 100644 --- a/meson.build +++ b/meson.build @@ -123,7 +123,12 @@ if get_option('push_notifications').enabled() add_project_arguments('-DOPENDHT_PUSH_NOTIFICATIONS', language: 'cpp') endif if get_option('peer_discovery').enabled() - opendht_src += ['src/mdns.cpp', 'src/dns_sd.cpp', 'src/mdns_transport.cpp', 'src/peer_discovery.cpp'] + opendht_src += [ + 'src/peer_discovery/mdns.cpp', + 'src/peer_discovery/dns_sd.cpp', + 'src/peer_discovery/mdns_transport.cpp', + 'src/peer_discovery/peer_discovery.cpp', + ] add_project_arguments('-DOPENDHT_PEER_DISCOVERY', language: 'cpp') if host_machine.system() == 'windows' deps += meson.get_compiler('cpp').find_library('iphlpapi') @@ -132,7 +137,7 @@ endif opendht = library( 'opendht', opendht_src, - include_directories: opendht_inc, + include_directories: [opendht_inc, opendht_interface_inc], dependencies: deps, cpp_args: ['-DOPENDHT_BUILD', '-Dopendht_EXPORTS'], install: true, diff --git a/src/dhtrunner.cpp b/src/dhtrunner.cpp index a33f6ef0..4511ff9f 100644 --- a/src/dhtrunner.cpp +++ b/src/dhtrunner.cpp @@ -237,7 +237,7 @@ DhtRunner::run(const Config& config, Context&& context) if (config.peer_discovery or config.peer_publish) { #ifdef OPENDHT_PEER_DISCOVERY peerDiscovery_ = context.peerDiscovery ? std::move(context.peerDiscovery) - : std::make_shared(); + : std::make_shared(nullptr, context.logger); #else std::cerr << "Peer discovery requested but OpenDHT built without peer discovery support." << std::endl; #endif diff --git a/src/mdns_transport.h b/src/mdns_transport.h deleted file mode 100644 index 5b93b674..00000000 --- a/src/mdns_transport.h +++ /dev/null @@ -1,40 +0,0 @@ -// Copyright (c) 2014-2026 Savoir-faire Linux Inc. -// SPDX-License-Identifier: MIT -#pragma once - -#include "dns_sd.h" -#include "logger.h" - -#include -#include -#include - -namespace asio { -class io_context; -} - -namespace dht::mdns { - -class MdnsTransport -{ -public: - using ReceiveCallback = std::function; - - MdnsTransport(std::shared_ptr ioContext, std::shared_ptr logger = {}); - ~MdnsTransport(); - - void start(ReceiveCallback callback); - void send(Packet packet); - void send(Packet packet, const asio::ip::udp::endpoint& destination); - void sendAndWait(Packet packet); - void refresh(); - void stop(); - - std::vector addresses() const; - -private: - class Impl; - std::shared_ptr impl_; -}; - -} // namespace dht::mdns \ No newline at end of file diff --git a/src/dns_sd.cpp b/src/peer_discovery/dns_sd.cpp similarity index 78% rename from src/dns_sd.cpp rename to src/peer_discovery/dns_sd.cpp index d65ceb78..bc2a5a8c 100644 --- a/src/dns_sd.cpp +++ b/src/peer_discovery/dns_sd.cpp @@ -11,6 +11,7 @@ namespace dht::mdns::dns_sd { namespace { constexpr std::string_view SERVICE_NAME = "_opendht._udp.local."; +constexpr std::string_view SERVICE_ENUMERATION_NAME = "_services._dns-sd._udp.local."; constexpr std::string_view SERVICE_SUFFIX = "._opendht._udp.local."; constexpr std::string_view HOST_SUFFIX = ".local."; @@ -55,10 +56,10 @@ makeAdvertisement(const Service& service, uint32_t serviceTtl, uint32_t hostTtl) Message message; message.response = true; message.authoritative = true; - message.answers.push_back( - {serviceName(), Type::PTR, CLASS_IN, false, serviceTtl, PtrData {instance}}); - message.additionals.push_back( - {instance, Type::SRV, CLASS_IN, true, hostTtl, SrvData {0, 0, service.port, host}}); + message.answers.push_back({serviceName(), Type::PTR, CLASS_IN, false, serviceTtl, PtrData {instance}}); + message.additionals.push_back({ + instance, Type::SRV, CLASS_IN, true, hostTtl, SrvData {0, 0, service.port, host} + }); TxtData txt; txt.strings.emplace_back(txtString("txtvers=1")); @@ -150,14 +151,13 @@ bool matches(const Question& question, const ResourceRecord& record) { return question.name == record.name and (question.type == Type::ANY or question.type == record.type) - and (question.classCode == CLASS_ANY or question.classCode == record.classCode); + and (question.classCode == CLASS_ANY or question.classCode == record.classCode); } bool sameRecord(const ResourceRecord& lhs, const ResourceRecord& rhs) { - return lhs.name == rhs.name and lhs.type == rhs.type and lhs.classCode == rhs.classCode - and lhs.data == rhs.data; + return lhs.name == rhs.name and lhs.type == rhs.type and lhs.classCode == rhs.classCode and lhs.data == rhs.data; } bool @@ -194,27 +194,28 @@ public: if (not record->cacheFlush) continue; const auto alreadyFlushed = std::any_of(flushed.begin(), flushed.end(), [&](const ResourceRecord* other) { - return other->name == record->name - and other->type == record->type and other->classCode == record->classCode; + return other->name == record->name and other->type == record->type + and other->classCode == record->classCode; }); if (not alreadyFlushed) { flushed.push_back(record); - entries_.erase( - std::remove_if(entries_.begin(), entries_.end(), [&](const Entry& entry) { - return entry.record.name == record->name and entry.record.type == record->type - and entry.record.classCode == record->classCode; - }), - entries_.end()); + entries_.erase(std::remove_if(entries_.begin(), + entries_.end(), + [&](const Entry& entry) { + return entry.record.name == record->name + and entry.record.type == record->type + and entry.record.classCode == record->classCode; + }), + entries_.end()); } } for (const auto* record : records) { if (record->ttl == 0) { - entries_.erase( - std::remove_if(entries_.begin(), entries_.end(), [&](const Entry& entry) { - return sameRecord(entry.record, *record); - }), - entries_.end()); + entries_.erase(std::remove_if(entries_.begin(), + entries_.end(), + [&](const Entry& entry) { return sameRecord(entry.record, *record); }), + entries_.end()); continue; } const auto expires = now_() + std::chrono::seconds(record->ttl); @@ -232,9 +233,10 @@ public: auto current = collect(); std::vector discovered; for (const auto& service : current) { - if (std::find_if(emitted_.begin(), emitted_.end(), [&](const Service& previous) { - return equal(previous, service); - }) == emitted_.end()) + if (std::find_if(emitted_.begin(), + emitted_.end(), + [&](const Service& previous) { return equal(previous, service); }) + == emitted_.end()) discovered.push_back(service); } emitted_ = std::move(current); @@ -259,15 +261,15 @@ private: static bool equal(const Service& lhs, const Service& rhs) { return lhs.nodeId == rhs.nodeId and lhs.network == rhs.network and lhs.port == rhs.port - and lhs.addresses == rhs.addresses; + and lhs.addresses == rhs.addresses; } void expire() { const auto now = now_(); - entries_.erase(std::remove_if(entries_.begin(), entries_.end(), [&](const Entry& entry) { - return entry.expires <= now; - }), + entries_.erase(std::remove_if(entries_.begin(), + entries_.end(), + [&](const Entry& entry) { return entry.expires <= now; }), entries_.end()); } @@ -383,18 +385,29 @@ respond(const Message& query, const Service& service) response.response = true; response.authoritative = true; + const ResourceRecord serviceType { + Name::fromString(SERVICE_ENUMERATION_NAME), + Type::PTR, + CLASS_IN, + false, + SERVICE_RECORD_TTL, + PtrData {serviceName()}, + }; + auto addMatching = [&](const ResourceRecord& record) { const auto question = std::find_if(query.questions.begin(), query.questions.end(), [&](const Question& q) { return matches(q, record); }); if (question == query.questions.end() or (not question->unicastResponse and isKnownAnswer(query, record))) return; - if (std::find_if(response.answers.begin(), response.answers.end(), [&](const ResourceRecord& answer) { - return sameRecord(answer, record); - }) == response.answers.end()) + if (std::find_if(response.answers.begin(), + response.answers.end(), + [&](const ResourceRecord& answer) { return sameRecord(answer, record); }) + == response.answers.end()) response.answers.push_back(record); }; + addMatching(serviceType); for (const auto& record : advertisement.answers) addMatching(record); for (const auto& record : advertisement.additionals) @@ -402,14 +415,23 @@ respond(const Message& query, const Service& service) if (response.answers.empty()) return {}; + const auto metaOnly = std::all_of(response.answers.begin(), response.answers.end(), [&](const auto& answer) { + return sameRecord(answer, serviceType); + }); + if (metaOnly) + return response; + response.additionals = advertisement.additionals; - response.additionals.erase( - std::remove_if(response.additionals.begin(), response.additionals.end(), [&](const ResourceRecord& record) { - return std::any_of(response.answers.begin(), response.answers.end(), [&](const ResourceRecord& answer) { - return sameRecord(answer, record); - }); - }), - response.additionals.end()); + response.additionals.erase(std::remove_if(response.additionals.begin(), + response.additionals.end(), + [&](const ResourceRecord& record) { + return std::any_of(response.answers.begin(), + response.answers.end(), + [&](const ResourceRecord& answer) { + return sameRecord(answer, record); + }); + }), + response.additionals.end()); return response; } diff --git a/src/dns_sd.h b/src/peer_discovery/dns_sd.h similarity index 94% rename from src/dns_sd.h rename to src/peer_discovery/dns_sd.h index 00307f52..a6e54468 100644 --- a/src/dns_sd.h +++ b/src/peer_discovery/dns_sd.h @@ -4,13 +4,8 @@ #include "mdns.h" -#if __has_include("infohash.h") -#include "infohash.h" -#include "utils.h" -#else #include "opendht/infohash.h" #include "opendht/utils.h" -#endif #include #include diff --git a/src/mdns.cpp b/src/peer_discovery/mdns.cpp similarity index 99% rename from src/mdns.cpp rename to src/peer_discovery/mdns.cpp index 41a9f106..105b7a9d 100644 --- a/src/mdns.cpp +++ b/src/peer_discovery/mdns.cpp @@ -145,10 +145,7 @@ public: return static_cast((high << 8) | low); } - uint32_t read32() - { - return (static_cast(read16()) << 16) | read16(); - } + uint32_t read32() { return (static_cast(read16()) << 16) | read16(); } Name readName() { @@ -493,7 +490,7 @@ Packet encode(const Message& message) { const auto recordCount = message.questions.size() + message.answers.size() + message.authorities.size() - + message.additionals.size(); + + message.additionals.size(); if (recordCount > MAX_RECORD_COUNT) throw ParseError("too many DNS records"); diff --git a/src/mdns.h b/src/peer_discovery/mdns.h similarity index 70% rename from src/mdns.h rename to src/peer_discovery/mdns.h index baf3338a..b95a900b 100644 --- a/src/mdns.h +++ b/src/peer_discovery/mdns.h @@ -2,11 +2,7 @@ // SPDX-License-Identifier: MIT #pragma once -#if __has_include("def.h") -#include "def.h" -#else #include "opendht/def.h" -#endif #include #include @@ -136,38 +132,63 @@ using Packet = std::vector; OPENDHT_PUBLIC Packet encode(const Message& message); OPENDHT_PUBLIC Message decode(const uint8_t* packet, size_t size); -inline bool operator==(const Question& lhs, const Question& rhs) +inline bool +operator==(const Question& lhs, const Question& rhs) { return lhs.name == rhs.name and lhs.type == rhs.type and lhs.classCode == rhs.classCode - and lhs.unicastResponse == rhs.unicastResponse; + and lhs.unicastResponse == rhs.unicastResponse; } -inline bool operator==(const AData& lhs, const AData& rhs) { return lhs.address == rhs.address; } -inline bool operator==(const AaaaData& lhs, const AaaaData& rhs) { return lhs.address == rhs.address; } -inline bool operator==(const PtrData& lhs, const PtrData& rhs) { return lhs.name == rhs.name; } -inline bool operator==(const TxtData& lhs, const TxtData& rhs) { return lhs.strings == rhs.strings; } -inline bool operator==(const SrvData& lhs, const SrvData& rhs) +inline bool +operator==(const AData& lhs, const AData& rhs) +{ + return lhs.address == rhs.address; +} +inline bool +operator==(const AaaaData& lhs, const AaaaData& rhs) +{ + return lhs.address == rhs.address; +} +inline bool +operator==(const PtrData& lhs, const PtrData& rhs) +{ + return lhs.name == rhs.name; +} +inline bool +operator==(const TxtData& lhs, const TxtData& rhs) +{ + return lhs.strings == rhs.strings; +} +inline bool +operator==(const SrvData& lhs, const SrvData& rhs) { return lhs.priority == rhs.priority and lhs.weight == rhs.weight and lhs.port == rhs.port - and lhs.target == rhs.target; + and lhs.target == rhs.target; } -inline bool operator==(const NsecData& lhs, const NsecData& rhs) +inline bool +operator==(const NsecData& lhs, const NsecData& rhs) { return lhs.nextDomain == rhs.nextDomain and lhs.types == rhs.types; } -inline bool operator==(const UnknownData& lhs, const UnknownData& rhs) { return lhs.bytes == rhs.bytes; } - -inline bool operator==(const ResourceRecord& lhs, const ResourceRecord& rhs) +inline bool +operator==(const UnknownData& lhs, const UnknownData& rhs) { - return lhs.name == rhs.name and lhs.type == rhs.type and lhs.classCode == rhs.classCode - and lhs.cacheFlush == rhs.cacheFlush and lhs.ttl == rhs.ttl and lhs.data == rhs.data; + return lhs.bytes == rhs.bytes; } -inline bool operator==(const Message& lhs, const Message& rhs) +inline bool +operator==(const ResourceRecord& lhs, const ResourceRecord& rhs) +{ + return lhs.name == rhs.name and lhs.type == rhs.type and lhs.classCode == rhs.classCode + and lhs.cacheFlush == rhs.cacheFlush and lhs.ttl == rhs.ttl and lhs.data == rhs.data; +} + +inline bool +operator==(const Message& lhs, const Message& rhs) { return lhs.id == rhs.id and lhs.response == rhs.response and lhs.authoritative == rhs.authoritative - and lhs.truncated == rhs.truncated and lhs.questions == rhs.questions and lhs.answers == rhs.answers - and lhs.authorities == rhs.authorities and lhs.additionals == rhs.additionals; + and lhs.truncated == rhs.truncated and lhs.questions == rhs.questions and lhs.answers == rhs.answers + and lhs.authorities == rhs.authorities and lhs.additionals == rhs.additionals; } } // namespace dht::mdns \ No newline at end of file diff --git a/src/mdns_transport.cpp b/src/peer_discovery/mdns_transport.cpp similarity index 54% rename from src/mdns_transport.cpp rename to src/peer_discovery/mdns_transport.cpp index 6d19ed1c..cffdf1cc 100644 --- a/src/mdns_transport.cpp +++ b/src/peer_discovery/mdns_transport.cpp @@ -6,6 +6,7 @@ #include #include #include +#include #include #include #include @@ -24,25 +25,16 @@ constexpr in_port_t MDNS_PORT = 5353; constexpr std::string_view MDNS_IPV4 = "224.0.0.251"; constexpr std::string_view MDNS_IPV6 = "ff02::fb"; -struct Interface +std::vector +interfaceAddresses([[maybe_unused]] asio::io_context& ioContext) { - asio::ip::address address; - unsigned index {}; - bool multicast {true}; - bool loopback {false}; -}; - -std::vector -interfaces([[maybe_unused]] asio::io_context& ioContext) -{ - std::vector result; + std::vector result; #ifdef _WIN32 ULONG size = 15000; std::vector buffer(size); auto* adapters = reinterpret_cast(buffer.data()); auto status = GetAdaptersAddresses(AF_UNSPEC, - GAA_FLAG_SKIP_ANYCAST | GAA_FLAG_SKIP_MULTICAST - | GAA_FLAG_SKIP_DNS_SERVER, + GAA_FLAG_SKIP_ANYCAST | GAA_FLAG_SKIP_MULTICAST | GAA_FLAG_SKIP_DNS_SERVER, nullptr, adapters, &size); @@ -50,16 +42,17 @@ interfaces([[maybe_unused]] asio::io_context& ioContext) buffer.resize(size); adapters = reinterpret_cast(buffer.data()); status = GetAdaptersAddresses(AF_UNSPEC, - GAA_FLAG_SKIP_ANYCAST | GAA_FLAG_SKIP_MULTICAST - | GAA_FLAG_SKIP_DNS_SERVER, + GAA_FLAG_SKIP_ANYCAST | GAA_FLAG_SKIP_MULTICAST | GAA_FLAG_SKIP_DNS_SERVER, nullptr, adapters, &size); } if (status == NO_ERROR) { + InterfaceId id = 0; for (auto* adapter = adapters; adapter; adapter = adapter->Next) { if (adapter->OperStatus != IfOperStatusUp) continue; + ++id; const auto multicast = not(adapter->Flags & IP_ADAPTER_NO_MULTICAST); const auto loopback = adapter->IfType == IF_TYPE_SOFTWARE_LOOPBACK; for (auto* current = adapter->FirstUnicastAddress; current; current = current->Next) { @@ -68,13 +61,16 @@ interfaces([[maybe_unused]] asio::io_context& ioContext) const auto* address = reinterpret_cast(socketAddress); asio::ip::address_v4::bytes_type bytes; std::memcpy(bytes.data(), &address->sin_addr, bytes.size()); - result.push_back({asio::ip::address_v4(bytes), adapter->IfIndex, multicast, loopback}); + result.push_back({id, adapter->IfIndex, asio::ip::address_v4(bytes), multicast, loopback}); } else if (socketAddress->sa_family == AF_INET6) { const auto* address = reinterpret_cast(socketAddress); asio::ip::address_v6::bytes_type bytes; std::memcpy(bytes.data(), &address->sin6_addr, bytes.size()); - result.push_back( - {asio::ip::address_v6(bytes, address->sin6_scope_id), adapter->Ipv6IfIndex, multicast, loopback}); + result.push_back({id, + adapter->Ipv6IfIndex, + asio::ip::address_v6(bytes, address->sin6_scope_id), + multicast, + loopback}); } } } @@ -91,41 +87,87 @@ interfaces([[maybe_unused]] asio::io_context& ioContext) const auto* address = reinterpret_cast(current->ifa_addr); asio::ip::address_v4::bytes_type bytes; std::memcpy(bytes.data(), &address->sin_addr, bytes.size()); - result.push_back({asio::ip::address_v4(bytes), - if_nametoindex(current->ifa_name), + const auto index = if_nametoindex(current->ifa_name); + result.push_back({index, + index, + asio::ip::address_v4(bytes), bool(current->ifa_flags & IFF_MULTICAST), bool(current->ifa_flags & IFF_LOOPBACK)}); } else if (family == AF_INET6) { const auto* address = reinterpret_cast(current->ifa_addr); asio::ip::address_v6::bytes_type bytes; std::memcpy(bytes.data(), &address->sin6_addr, bytes.size()); - result.push_back({asio::ip::address_v6(bytes, address->sin6_scope_id), - if_nametoindex(current->ifa_name), + const auto index = if_nametoindex(current->ifa_name); + result.push_back({index, + index, + asio::ip::address_v6(bytes, address->sin6_scope_id), bool(current->ifa_flags & IFF_MULTICAST), bool(current->ifa_flags & IFF_LOOPBACK)}); } } freeifaddrs(addresses); #endif - result.erase(std::remove_if(result.begin(), result.end(), [](const Interface& interface) { - return interface.address.is_unspecified() or interface.address.is_multicast(); - }), - result.end()); - std::stable_sort(result.begin(), result.end(), [](const Interface& lhs, const Interface& rhs) { - return lhs.loopback < rhs.loopback; - }); return result; } } // namespace +std::vector +groupInterfaces(const std::vector& addresses) +{ + std::vector result; + const auto hasNonLoopback = std::any_of(addresses.begin(), addresses.end(), [](const InterfaceAddress& address) { + return address.multicast and not address.loopback and not address.address.is_unspecified() + and not address.address.is_multicast(); + }); + + for (const auto& address : addresses) { + if (address.id == 0 or not address.multicast or (hasNonLoopback and address.loopback) + or address.address.is_unspecified() or address.address.is_multicast()) + continue; + + auto scope = std::find_if(result.begin(), result.end(), [&](const NetworkInterface& interface) { + return interface.id == address.id; + }); + if (scope == result.end()) { + NetworkInterface interface; + interface.id = address.id; + result.push_back(std::move(interface)); + scope = std::prev(result.end()); + } + + dns_sd::AddressData record; + if (address.address.is_v4()) { + scope->ipv4Index = address.index; + if (not scope->ipv4Outbound) + scope->ipv4Outbound = address.address.to_v4(); + record = AData {address.address.to_v4().to_bytes()}; + } else { + const auto bytes = address.address.to_v6().to_bytes(); + const auto mapped = std::all_of(bytes.begin(), bytes.begin() + 10, [](uint8_t byte) { return byte == 0; }) + and bytes[10] == 0xff and bytes[11] == 0xff; + if (mapped) + continue; + scope->ipv6Index = address.index; + record = AaaaData {bytes}; + } + if (std::find(scope->addresses.begin(), scope->addresses.end(), record) == scope->addresses.end()) + scope->addresses.emplace_back(std::move(record)); + } + result.erase(std::remove_if(result.begin(), + result.end(), + [](const NetworkInterface& interface) { return interface.addresses.empty(); }), + result.end()); + return result; +} + class MdnsTransport::Impl : public std::enable_shared_from_this { public: Impl(std::shared_ptr ioContext, std::shared_ptr logger) : ioContext_(std::move(ioContext)) , logger_(std::move(logger)) - , interfaces_(interfaces(*ioContext_)) + , interfaces_(groupInterfaces(interfaceAddresses(*ioContext_))) {} struct Socket @@ -138,6 +180,8 @@ public: std::array receiveBuffer {}; asio::ip::udp::endpoint sender; asio::ip::udp::endpoint multicast; + InterfaceId interface {}; + unsigned ipv6Index {}; }; void start(ReceiveCallback callback) @@ -146,44 +190,43 @@ public: configure(); } - void send(Packet packet) - { - send(std::move(packet), {}); - } + void send(Packet packet) { send(std::move(packet), {}, {}); } - void send(Packet packet, std::optional destination) + void send(Packet packet, std::optional destination, std::optional interface) { auto sharedPacket = std::make_shared(std::move(packet)); - asio::post(*ioContext_, [weak = weak_from_this(), sharedPacket, destination] { + asio::post(*ioContext_, [weak = weak_from_this(), sharedPacket, destination, interface] { const auto self = weak.lock(); if (not self) return; for (const auto& state : self->sockets_) { + if (interface and state->interface != *interface) + continue; const auto target = destination.value_or(state->multicast); if (target.address().is_v4() != state->multicast.address().is_v4()) continue; - state->socket.async_send_to( - asio::buffer(*sharedPacket), - target, - [logger = self->logger_, sharedPacket](const asio::error_code& error, size_t) { - if (error and error != asio::error::operation_aborted and logger) - logger->warn("Unable to send mDNS packet: {}", error.message()); - }); + state->socket.async_send_to(asio::buffer(*sharedPacket), + target, + [logger = self->logger_, sharedPacket](const asio::error_code& error, + size_t) { + if (error and error != asio::error::operation_aborted and logger) + logger->warn("Unable to send mDNS packet: {}", error.message()); + }); } }); } - void sendAndWait(Packet packet) + void sendAndWait(Packet packet, InterfaceId interface) { if (ioContext_->get_executor().running_in_this_thread()) { - sendNow(packet); + sendNow(packet, interface); return; } auto completed = std::make_shared>(); auto future = completed->get_future(); - asio::post(*ioContext_, [weak = weak_from_this(), packet = std::move(packet), completed] { + asio::post(*ioContext_, [weak = weak_from_this(), packet = std::move(packet), interface, completed] { if (const auto self = weak.lock()) - self->sendNow(packet); + self->sendNow(packet, interface); completed->set_value(); }); future.wait_for(std::chrono::milliseconds(250)); @@ -193,7 +236,7 @@ public: { { std::lock_guard lock(interfacesMutex_); - interfaces_ = interfaces(*ioContext_); + interfaces_ = groupInterfaces(interfaceAddresses(*ioContext_)); } asio::post(*ioContext_, [weak = weak_from_this()] { if (const auto self = weak.lock()) @@ -217,34 +260,18 @@ public: future.wait_for(std::chrono::milliseconds(250)); } - std::vector addresses() const + std::vector interfaces() const { std::lock_guard lock(interfacesMutex_); - std::vector result; - const auto hasNonLoopback = std::any_of(interfaces_.begin(), interfaces_.end(), [](const Interface& interface) { - return not interface.loopback; - }); - for (const auto& interface : interfaces_) { - if (hasNonLoopback and interface.loopback) - continue; - if (interface.address.is_v4()) { - result.emplace_back(AData {interface.address.to_v4().to_bytes()}); - } else { - const auto bytes = interface.address.to_v6().to_bytes(); - const auto mapped = std::all_of(bytes.begin(), bytes.begin() + 10, [](uint8_t byte) { - return byte == 0; - }) and bytes[10] == 0xff and bytes[11] == 0xff; - if (not mapped) - result.emplace_back(AaaaData {bytes}); - } - } - return result; + return interfaces_; } private: - void sendNow(const Packet& packet) + void sendNow(const Packet& packet, InterfaceId interface) { for (const auto& state : sockets_) { + if (state->interface != interface) + continue; asio::error_code error; state->socket.send_to(asio::buffer(packet), state->multicast, 0, error); if (error and logger_) @@ -261,11 +288,20 @@ private: void configure() { stopSockets(); - configure(asio::ip::udp::v4(), asio::ip::make_address(MDNS_IPV4)); - configure(asio::ip::udp::v6(), asio::ip::make_address(MDNS_IPV6)); + std::vector currentInterfaces; + { + std::lock_guard lock(interfacesMutex_); + currentInterfaces = interfaces_; + } + for (const auto& interface : currentInterfaces) { + if (interface.ipv4Outbound) + configure(interface, asio::ip::udp::v4(), asio::ip::make_address(MDNS_IPV4)); + if (interface.ipv6Index) + configure(interface, asio::ip::udp::v6(), asio::ip::make_address(MDNS_IPV6)); + } } - void configure(asio::ip::udp protocol, const asio::ip::address& multicast) + void configure(const NetworkInterface& interface, asio::ip::udp protocol, const asio::ip::address& multicast) { auto state = std::make_shared(*ioContext_); asio::error_code error; @@ -273,6 +309,15 @@ private: if (error) return logSocketError(protocol, error); state->socket.set_option(asio::ip::udp::socket::reuse_address(true), error); +#if defined(SO_REUSEPORT) && !defined(_WIN32) + if (not error) { + const int enabled = 1; + if (::setsockopt(state->socket.native_handle(), SOL_SOCKET, SO_REUSEPORT, &enabled, sizeof(enabled)) != 0) + error.assign(errno, asio::error::get_system_category()); + } +#endif + if (error) + return logSocketError(protocol, error); if (protocol == asio::ip::udp::v6()) state->socket.set_option(asio::ip::v6_only(true), error); state->socket.bind({protocol, MDNS_PORT}, error); @@ -282,64 +327,48 @@ private: if (error) return logSocketError(protocol, error); state->socket.set_option(asio::ip::multicast::enable_loopback(true), error); - - std::vector currentInterfaces; - { - std::lock_guard lock(interfacesMutex_); - currentInterfaces = interfaces_; - } - size_t joined = 0; - for (const auto& interface : currentInterfaces) { - if (not interface.multicast or interface.address.is_v4() != (protocol == asio::ip::udp::v4())) - continue; - if (protocol == asio::ip::udp::v4()) - state->socket.set_option(asio::ip::multicast::join_group(multicast.to_v4(), - interface.address.to_v4()), - error); - else - state->socket.set_option(asio::ip::multicast::join_group(multicast.to_v6(), interface.index), error); + if (protocol == asio::ip::udp::v4()) { + state->socket.set_option(asio::ip::multicast::join_group(multicast.to_v4(), *interface.ipv4Outbound), error); if (not error) - joined++; - } - if (joined == 0) { - state->socket.set_option(asio::ip::multicast::join_group(multicast), error); - if (error) - return logSocketError(protocol, error); + state->socket.set_option(asio::ip::multicast::outbound_interface(*interface.ipv4Outbound), error); + } else { + state->socket.set_option(asio::ip::multicast::join_group(multicast.to_v6(), interface.ipv6Index), error); + if (not error) + state->socket.set_option(asio::ip::multicast::outbound_interface(interface.ipv6Index), error); } + if (error) + return logSocketError(protocol, error); - const auto outbound = std::find_if(currentInterfaces.begin(), currentInterfaces.end(), [&](const Interface& interface) { - return interface.multicast and not interface.loopback - and interface.address.is_v4() == (protocol == asio::ip::udp::v4()); - }); - if (outbound != currentInterfaces.end()) { - if (protocol == asio::ip::udp::v4()) - state->socket.set_option(asio::ip::multicast::outbound_interface(outbound->address.to_v4()), error); - else - state->socket.set_option(asio::ip::multicast::outbound_interface(outbound->index), error); - } - state->multicast = {multicast, MDNS_PORT}; + if (protocol == asio::ip::udp::v6()) + state->multicast = {asio::ip::address_v6(multicast.to_v6().to_bytes(), interface.ipv6Index), MDNS_PORT}; + else + state->multicast = {multicast, MDNS_PORT}; + state->interface = interface.id; + state->ipv6Index = interface.ipv6Index; sockets_.push_back(state); receive(state); } void receive(const std::shared_ptr& state) { - state->socket.async_receive_from( - asio::buffer(state->receiveBuffer), - state->sender, - [weak = weak_from_this(), state](const asio::error_code& error, size_t size) { - const auto self = weak.lock(); - if (not self) - return; - if (not error and self->callback_) { - Packet packet(state->receiveBuffer.begin(), state->receiveBuffer.begin() + size); - self->callback_(packet, state->sender); - } else if (error != asio::error::operation_aborted and self->logger_) { - self->logger_->warn("Unable to receive mDNS packet: {}", error.message()); - } - if (state->socket.is_open()) - self->receive(state); - }); + state->socket.async_receive_from(asio::buffer(state->receiveBuffer), + state->sender, + [weak = weak_from_this(), state](const asio::error_code& error, size_t size) { + const auto self = weak.lock(); + if (not self) + return; + if (not error and self->callback_) { + Packet packet(state->receiveBuffer.begin(), + state->receiveBuffer.begin() + size); + self->callback_(packet, + {state->sender, state->interface, state->ipv6Index}); + } else if (error != asio::error::operation_aborted and self->logger_) { + self->logger_->warn("Unable to receive mDNS packet: {}", + error.message()); + } + if (state->socket.is_open()) + self->receive(state); + }); } void stopSockets() @@ -363,7 +392,7 @@ private: std::shared_ptr ioContext_; std::shared_ptr logger_; mutable std::mutex interfacesMutex_; - std::vector interfaces_; + std::vector interfaces_; std::vector> sockets_; ReceiveCallback callback_; }; @@ -372,8 +401,7 @@ MdnsTransport::MdnsTransport(std::shared_ptr ioContext, std::s : impl_(std::make_shared(std::move(ioContext), std::move(logger))) {} -MdnsTransport::~MdnsTransport() -= default; +MdnsTransport::~MdnsTransport() = default; void MdnsTransport::start(ReceiveCallback callback) @@ -388,15 +416,21 @@ MdnsTransport::send(Packet packet) } void -MdnsTransport::send(Packet packet, const asio::ip::udp::endpoint& destination) +MdnsTransport::send(Packet packet, InterfaceId interface) { - impl_->send(std::move(packet), destination); + impl_->send(std::move(packet), {}, interface); } void -MdnsTransport::sendAndWait(Packet packet) +MdnsTransport::send(Packet packet, const asio::ip::udp::endpoint& destination, InterfaceId interface) { - impl_->sendAndWait(std::move(packet)); + impl_->send(std::move(packet), destination, interface); +} + +void +MdnsTransport::sendAndWait(Packet packet, InterfaceId interface) +{ + impl_->sendAndWait(std::move(packet), interface); } void @@ -412,10 +446,10 @@ MdnsTransport::stop() impl_->stop(); } -std::vector -MdnsTransport::addresses() const +std::vector +MdnsTransport::interfaces() const { - return impl_->addresses(); + return impl_->interfaces(); } } // namespace dht::mdns \ No newline at end of file diff --git a/src/peer_discovery/mdns_transport.h b/src/peer_discovery/mdns_transport.h new file mode 100644 index 00000000..00f4b702 --- /dev/null +++ b/src/peer_discovery/mdns_transport.h @@ -0,0 +1,74 @@ +// Copyright (c) 2014-2026 Savoir-faire Linux Inc. +// SPDX-License-Identifier: MIT +#pragma once + +#include "dns_sd.h" + +#include "opendht/logger.h" + +#include +#include +#include +#include +#include +#include + +namespace asio { +class io_context; +} + +namespace dht::mdns { + +using InterfaceId = uint64_t; + +struct InterfaceAddress +{ + InterfaceId id {}; + unsigned index {}; + asio::ip::address address; + bool multicast {true}; + bool loopback {false}; +}; + +struct NetworkInterface +{ + InterfaceId id {}; + unsigned ipv4Index {}; + unsigned ipv6Index {}; + std::optional ipv4Outbound; + std::vector addresses; +}; + +OPENDHT_PUBLIC std::vector groupInterfaces(const std::vector& addresses); + +struct ReceiveContext +{ + asio::ip::udp::endpoint source; + InterfaceId interface {}; + unsigned ipv6Index {}; +}; + +class MdnsTransport +{ +public: + using ReceiveCallback = std::function; + + MdnsTransport(std::shared_ptr ioContext, std::shared_ptr logger = {}); + ~MdnsTransport(); + + void start(ReceiveCallback callback); + void send(Packet packet); + void send(Packet packet, InterfaceId interface); + void send(Packet packet, const asio::ip::udp::endpoint& destination, InterfaceId interface); + void sendAndWait(Packet packet, InterfaceId interface); + void refresh(); + void stop(); + + std::vector interfaces() const; + +private: + class Impl; + std::shared_ptr impl_; +}; + +} // namespace dht::mdns \ No newline at end of file diff --git a/src/peer_discovery.cpp b/src/peer_discovery/peer_discovery.cpp similarity index 71% rename from src/peer_discovery.cpp rename to src/peer_discovery/peer_discovery.cpp index 2026bbf7..8be1b5f7 100644 --- a/src/peer_discovery.cpp +++ b/src/peer_discovery/peer_discovery.cpp @@ -9,6 +9,7 @@ #include #include #include +#include #include #include #include @@ -21,9 +22,7 @@ constexpr auto CONNECTIVITY_RETRY = std::chrono::seconds(10); constexpr auto CONNECTIVITY_RETRY_MAX = std::chrono::minutes(1); SockAddr -socketAddress(const mdns::dns_sd::AddressData& data, - in_port_t port, - const asio::ip::udp::endpoint& source) +socketAddress(const mdns::dns_sd::AddressData& data, in_port_t port, const mdns::ReceiveContext& context) { SockAddr address; std::visit( @@ -36,8 +35,8 @@ socketAddress(const mdns::dns_sd::AddressData& data, address.setFamily(AF_INET6); std::memcpy(&address.getIPv6().sin6_addr, record.address.data(), record.address.size()); const asio::ip::address_v6 parsed(record.address); - if (parsed.is_link_local() and source.address().is_v6()) - address.getIPv6().sin6_scope_id = source.address().to_v6().scope_id(); + if (parsed.is_link_local()) + address.getIPv6().sin6_scope_id = context.ipv6Index; } }, data); @@ -61,11 +60,12 @@ public: void start() { - transport_.start([weak = weak_from_this()](const mdns::Packet& packet, - const asio::ip::udp::endpoint& source) { + transport_.start([weak = weak_from_this()](const mdns::Packet& packet, const mdns::ReceiveContext& context) { if (const auto self = weak.lock()) - self->receive(packet, source); + self->receive(packet, context); }); + if (logger_) + logger_->debug("[peer discovery {:p}] Started.", fmt::ptr(this)); } void startDiscovery(NetId network, PeerDiscoveredCallback callback) @@ -83,13 +83,14 @@ public: void startPublish(const InfoHash& nodeId, NetId network, in_port_t port) { - mdns::dns_sd::Service service {nodeId, network, port, transport_.addresses()}; - const auto message = mdns::dns_sd::announcement(service); + mdns::dns_sd::Service service {nodeId, network, port, {}}; + if (transport_.interfaces().empty()) + throw mdns::ParseError("DNS-SD service requires a multicast interface"); { std::lock_guard lock(mutex_); published_ = std::move(service); } - transport_.send(mdns::encode(message)); + announce(); asio::post(*ioContext_, [weak = weak_from_this()] { const auto self = weak.lock(); if (not self) @@ -129,7 +130,7 @@ public: self->announceTimer_.cancel(); }); if (service) - transport_.send(mdns::encode(mdns::dns_sd::goodbye(*service))); + advertise(*service, true, false); return bool(service); } @@ -147,7 +148,7 @@ public: self->announceTimer_.cancel(); }); if (service) - transport_.sendAndWait(mdns::encode(mdns::dns_sd::goodbye(*service))); + advertise(*service, true, true); transport_.stop(); } @@ -172,6 +173,7 @@ public: private: void handleConnectivityChanged() { + caches_.clear(); transport_.refresh(); refreshPublication(); browse(); @@ -183,9 +185,9 @@ private: self->handleConnectivityChanged(); } }); - connectivityRetry_ = std::min( - connectivityRetry_ * 2, - std::chrono::duration_cast(CONNECTIVITY_RETRY_MAX)); + connectivityRetry_ = std::min(connectivityRetry_ * 2, + std::chrono::duration_cast( + CONNECTIVITY_RETRY_MAX)); } void browse() @@ -197,6 +199,8 @@ private: } if (not active) return; + if (logger_) + logger_->debug("[peer discovery {:p}] Browsing.", fmt::ptr(this)); transport_.send(mdns::encode(mdns::dns_sd::browseQuery())); browseTimer_.expires_after(BROWSE_INTERVAL); browseTimer_.async_wait([weak = weak_from_this()](const asio::error_code& error) { @@ -214,8 +218,10 @@ private: std::lock_guard lock(mutex_); service = published_; } + if (logger_) + logger_->debug("[peer discovery {:p}] Announcing.", fmt::ptr(this)); if (service) - transport_.send(mdns::encode(mdns::dns_sd::announcement(*service))); + advertise(*service, false, false); } void refreshPublication() @@ -223,30 +229,43 @@ private: std::optional service; { std::lock_guard lock(mutex_); - if (published_) { - published_->addresses = transport_.addresses(); + if (published_) service = published_; - } } if (service) - transport_.send(mdns::encode(mdns::dns_sd::announcement(*service))); + advertise(*service, false, false); } - void receive(const mdns::Packet& packet, const asio::ip::udp::endpoint& source) + void advertise(const mdns::dns_sd::Service& service, bool goodbye, bool wait) { + for (const auto& interface : transport_.interfaces()) { + auto scoped = service; + scoped.addresses = interface.addresses; + auto packet = mdns::encode(goodbye ? mdns::dns_sd::goodbye(scoped) : mdns::dns_sd::announcement(scoped)); + if (wait) + transport_.sendAndWait(std::move(packet), interface.id); + else + transport_.send(std::move(packet), interface.id); + } + } + + void receive(const mdns::Packet& packet, const mdns::ReceiveContext& context) + { + if (logger_) + logger_->debug("[peer discovery {:p}] Got packet of size {}.", fmt::ptr(this), packet.size()); try { const auto message = mdns::decode(packet.data(), packet.size()); if (message.response) - discover(message, source); + discover(message, context); else - answer(message, source); + answer(message, context); } catch (const std::exception& error) { if (logger_) logger_->warn("Ignoring invalid mDNS packet: {}", error.what()); } } - void answer(const mdns::Message& query, const asio::ip::udp::endpoint& source) + void answer(const mdns::Message& query, const mdns::ReceiveContext& context) { std::optional service; { @@ -254,30 +273,40 @@ private: service = published_; } if (service) { + const auto interfaces = transport_.interfaces(); + const auto interface = std::find_if(interfaces.begin(), interfaces.end(), [&](const auto& candidate) { + return candidate.id == context.interface; + }); + if (interface == interfaces.end()) + return; + service->addresses = interface->addresses; if (auto response = mdns::dns_sd::respond(query, *service)) { - const auto unicast = source.port() != 5353 - or std::any_of(query.questions.begin(), query.questions.end(), [](const mdns::Question& question) { - return question.unicastResponse; - }); - if (source.port() != 5353) + const auto unicast = context.source.port() != 5353 + or std::any_of(query.questions.begin(), + query.questions.end(), + [](const mdns::Question& question) { + return question.unicastResponse; + }); + if (context.source.port() != 5353) response->questions = query.questions; auto packet = mdns::encode(*response); - const auto sharedAnswer = std::any_of( - response->answers.begin(), response->answers.end(), [](const mdns::ResourceRecord& record) { - return not record.cacheFlush; - }); + const auto sharedAnswer = std::any_of(response->answers.begin(), + response->answers.end(), + [](const mdns::ResourceRecord& record) { + return not record.cacheFlush; + }); if (not sharedAnswer) { if (unicast) - transport_.send(std::move(packet), source); + transport_.send(std::move(packet), context.source, context.interface); else - transport_.send(std::move(packet)); + transport_.send(std::move(packet), context.interface); return; } static thread_local std::mt19937 random {std::random_device {}()}; const auto delay = std::chrono::milliseconds(std::uniform_int_distribution {20, 120}(random)); auto timer = std::make_shared(*ioContext_, delay); - timer->async_wait([weak = weak_from_this(), timer, packet = std::move(packet), source, unicast]( + timer->async_wait([weak = weak_from_this(), timer, packet = std::move(packet), context, unicast]( const asio::error_code& error) mutable { if (error) return; @@ -285,15 +314,15 @@ private: if (not self) return; if (unicast) - self->transport_.send(std::move(packet), source); + self->transport_.send(std::move(packet), context.source, context.interface); else - self->transport_.send(std::move(packet)); + self->transport_.send(std::move(packet), context.interface); }); } } } - void discover(const mdns::Message& message, const asio::ip::udp::endpoint& source) + void discover(const mdns::Message& message, const mdns::ReceiveContext& context) { PeerDiscoveredCallback callback; std::optional network; @@ -308,11 +337,12 @@ private: if (not callback or not network) return; - for (const auto& service : cache_.update(message)) { + const auto cache = caches_.try_emplace(context.interface).first; + for (const auto& service : cache->second.update(message)) { if (service.network != *network or (localNode and service.nodeId == *localNode)) continue; for (const auto& address : service.addresses) - callback(service.nodeId, socketAddress(address, service.port, source)); + callback(service.nodeId, socketAddress(address, service.port, context)); } } @@ -327,7 +357,7 @@ private: std::optional discoveryNetwork_; PeerDiscoveredCallback discoveryCallback_; std::optional published_; - mdns::dns_sd::Cache cache_; + std::map caches_; }; PeerDiscovery::PeerDiscovery(std::shared_ptr ioContext, std::shared_ptr logger)