peer_discovery: scope DNS-SD traffic by interface

This commit is contained in:
Adrien Beraud committed 2026-08-12 16:30:44 -04:00
1 parent c58c14d49a
commit dac5f0fc05
11 files changed
+428 -286

No files matched your search

+5 -1
View File
@@ -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()
+7 -2
View File
@@ -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,
+1 -1
View File
@@ -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<PeerDiscovery>();
: std::make_shared<PeerDiscovery>(nullptr, context.logger);
#else
std::cerr << "Peer discovery requested but OpenDHT built without peer discovery support." << std::endl;
#endif
-40
View File
@@ -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 <asio/ip/udp.hpp>
#include <functional>
#include <memory>
namespace asio {
class io_context;
}
namespace dht::mdns {
class MdnsTransport
{
public:
using ReceiveCallback = std::function<void(const Packet&, const asio::ip::udp::endpoint&)>;
MdnsTransport(std::shared_ptr<asio::io_context> ioContext, std::shared_ptr<Logger> 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<dns_sd::AddressData> addresses() const;
private:
class Impl;
std::shared_ptr<Impl> impl_;
};
} // namespace dht::mdns
@@ -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<Service> 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;
}
@@ -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 <chrono>
#include <functional>
+2 -5
View File
@@ -145,10 +145,7 @@ public:
return static_cast<uint16_t>((high << 8) | low);
}
uint32_t read32()
{
return (static_cast<uint32_t>(read16()) << 16) | read16();
}
uint32_t read32() { return (static_cast<uint32_t>(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");
+42 -21
View File
@@ -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 <array>
#include <cstddef>
@@ -136,38 +132,63 @@ using Packet = std::vector<uint8_t>;
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
@@ -6,6 +6,7 @@
#include <asio.hpp>
#include <algorithm>
#include <array>
#include <cerrno>
#include <cstring>
#include <future>
#include <mutex>
@@ -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<InterfaceAddress>
interfaceAddresses([[maybe_unused]] asio::io_context& ioContext)
{
asio::ip::address address;
unsigned index {};
bool multicast {true};
bool loopback {false};
};
std::vector<Interface>
interfaces([[maybe_unused]] asio::io_context& ioContext)
{
std::vector<Interface> result;
std::vector<InterfaceAddress> result;
#ifdef _WIN32
ULONG size = 15000;
std::vector<uint8_t> buffer(size);
auto* adapters = reinterpret_cast<IP_ADAPTER_ADDRESSES*>(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<IP_ADAPTER_ADDRESSES*>(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<const sockaddr_in*>(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<const sockaddr_in6*>(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<const sockaddr_in*>(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<const sockaddr_in6*>(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<NetworkInterface>
groupInterfaces(const std::vector<InterfaceAddress>& addresses)
{
std::vector<NetworkInterface> 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<MdnsTransport::Impl>
{
public:
Impl(std::shared_ptr<asio::io_context> ioContext, std::shared_ptr<Logger> 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<uint8_t, MAX_PACKET_SIZE> 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<asio::ip::udp::endpoint> destination)
void send(Packet packet, std::optional<asio::ip::udp::endpoint> destination, std::optional<InterfaceId> interface)
{
auto sharedPacket = std::make_shared<Packet>(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<std::promise<void>>();
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<dns_sd::AddressData> addresses() const
std::vector<NetworkInterface> interfaces() const
{
std::lock_guard lock(interfacesMutex_);
std::vector<dns_sd::AddressData> 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<NetworkInterface> 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<Socket>(*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<Interface> 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<Socket>& 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<asio::io_context> ioContext_;
std::shared_ptr<Logger> logger_;
mutable std::mutex interfacesMutex_;
std::vector<Interface> interfaces_;
std::vector<NetworkInterface> interfaces_;
std::vector<std::shared_ptr<Socket>> sockets_;
ReceiveCallback callback_;
};
@@ -372,8 +401,7 @@ MdnsTransport::MdnsTransport(std::shared_ptr<asio::io_context> ioContext, std::s
: impl_(std::make_shared<Impl>(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<dns_sd::AddressData>
MdnsTransport::addresses() const
std::vector<NetworkInterface>
MdnsTransport::interfaces() const
{
return impl_->addresses();
return impl_->interfaces();
}
} // namespace dht::mdns
+74
View File
@@ -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 <asio/ip/address.hpp>
#include <asio/ip/udp.hpp>
#include <cstdint>
#include <functional>
#include <memory>
#include <optional>
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<asio::ip::address_v4> ipv4Outbound;
std::vector<dns_sd::AddressData> addresses;
};
OPENDHT_PUBLIC std::vector<NetworkInterface> groupInterfaces(const std::vector<InterfaceAddress>& addresses);
struct ReceiveContext
{
asio::ip::udp::endpoint source;
InterfaceId interface {};
unsigned ipv6Index {};
};
class MdnsTransport
{
public:
using ReceiveCallback = std::function<void(const Packet&, const ReceiveContext&)>;
MdnsTransport(std::shared_ptr<asio::io_context> ioContext, std::shared_ptr<Logger> 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<NetworkInterface> interfaces() const;
private:
class Impl;
std::shared_ptr<Impl> impl_;
};
} // namespace dht::mdns
@@ -9,6 +9,7 @@
#include <asio.hpp>
#include <chrono>
#include <cstring>
#include <map>
#include <mutex>
#include <optional>
#include <random>
@@ -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<std::chrono::steady_clock::duration>(CONNECTIVITY_RETRY_MAX));
connectivityRetry_ = std::min(connectivityRetry_ * 2,
std::chrono::duration_cast<std::chrono::steady_clock::duration>(
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<mdns::dns_sd::Service> 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<mdns::dns_sd::Service> 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<int> {20, 120}(random));
auto timer = std::make_shared<asio::steady_timer>(*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<NetId> 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<NetId> discoveryNetwork_;
PeerDiscoveredCallback discoveryCallback_;
std::optional<mdns::dns_sd::Service> published_;
mdns::dns_sd::Cache cache_;
std::map<mdns::InterfaceId, mdns::dns_sd::Cache> caches_;
};
PeerDiscovery::PeerDiscovery(std::shared_ptr<asio::io_context> ioContext, std::shared_ptr<Logger> logger)