diff --git a/libp2p/nameresolving.nim b/libp2p/nameresolving.nim new file mode 100644 index 0000000000..188a3a671c --- /dev/null +++ b/libp2p/nameresolving.nim @@ -0,0 +1,34 @@ +# SPDX-License-Identifier: Apache-2.0 OR MIT +# Copyright (c) Status Research & Development GmbH + +## Choose a DNS resolver with explicit servers or an OS-backed resolver: +## +## ```nim +## let resolver = newNameResolver(rng) +## let addresses = await resolver.resolveMAddress(address) +## await resolver.stop() +## ``` +## +## Passing a resolver to `SwitchBuilder.withNameResolver` keeps its lifecycle +## caller-managed. Stop dependent switches before awaiting `SystemResolver.stop()`. +## +## The OS resolver requires threads. Its TXT queries use discovered system DNS +## servers, with DefaultDnsServers as a fallback when discovery finds none. + +{.push raises: [].} + +import chronos +import ./nameresolving/[nameresolver, dnsresolver, systemresolver] +import ./crypto/rng + +export nameresolver, dnsresolver, systemresolver + +proc newNameResolver*(rng: Rng, nameServers: seq[TransportAddress]): DnsResolver = + ## Use only the supplied DNS servers. An empty list produces no DNS results. + DnsResolver.new(nameServers, rng) + +proc newNameResolver*(rng: Rng): SystemResolver = + ## Resolve IPs through the OS. Discovers TXT nameservers synchronously during + ## construction, which can block the calling thread while reading system config. + ## Requires threads. The caller must await stop() to release its workers. + SystemResolver.new(rng) diff --git a/libp2p/nameresolving/systemresolver.nim b/libp2p/nameresolving/systemresolver.nim new file mode 100644 index 0000000000..0fb1978282 --- /dev/null +++ b/libp2p/nameresolving/systemresolver.nim @@ -0,0 +1,359 @@ +# SPDX-License-Identifier: Apache-2.0 OR MIT +# Copyright (c) Status Research & Development GmbH + +## Resolve IPs through the OS, including hosts files and search domains. +## Blocking getaddrinfo calls run on worker threads; TXT queries use a fallback +## NameResolver. Call stop() to release workers, and start() to reuse a resolver. + +{.push raises: [].} + +import std/[atomics, locks] +import chronos, chronos/threadsync, chronicles +import nameresolver, dnsresolver +import ../crypto/rng + +logScope: + topics = "libp2p systemresolver" + +const + MaxSystemAddresses = 100 + IPv4MappedMarkerByte = 0xff'u8 + +type + RequestState {.pure.} = enum + Pending + Done + Discarded + + # Everything the worker threads touch lives in the shared heap and is + # plain data - no GC'd Nim types may cross the thread boundary. + SharedRequest = object + next: ptr SharedRequest + host: array[256, char] # null-terminated, copied at submission + port: uint16 + domain: Domain + state: Atomic[int] # RequestState + results: array[MaxSystemAddresses, TransportAddress] + resultsLen: int + + SharedState = object + lock: Lock + queueHead, queueTail: ptr SharedRequest + completedHead: ptr SharedRequest # singly-linked via `next` + running: Atomic[bool] + workersRunning: Atomic[int] + requestSignal, responseSignal: ThreadSignalPtr + when defined(libp2p_testing): + workerEntered, workerRelease: ThreadSignalPtr + workerGateReleased: Atomic[bool] + + PendingRequest = object + req: ptr SharedRequest + fut: Future[seq[TransportAddress]].Raising([CancelledError]) + + SystemResolver* = ref object of NameResolver + ## Resolves names through the OS resolver (getaddrinfo), offloaded to + ## lazily-created worker threads. + txtResolver: NameResolver + shared: ptr SharedState + workers: seq[Thread[ptr SharedState]] + workerCount: int + pending: seq[PendingRequest] + dispatcherFut: Future[void].Raising([]) + stopFut: Future[void].Raising([]) + stopped: bool + when defined(libp2p_testing): + workerEntered, workerRelease: ThreadSignalPtr + workerGateReleased: bool + +proc isV4Mapped(address: TransportAddress): bool = + if address.family != AddressFamily.IPv6: + return false + for i in 0 ..< 10: + if address.address_v6[i] != 0: + return false + address.address_v6[10] == IPv4MappedMarkerByte and + address.address_v6[11] == IPv4MappedMarkerByte + +proc workerLoop(shared: ptr SharedState) {.thread.} = + {.cast(gcsafe).}: + while shared[].running.load(): + discard shared[].requestSignal.waitSync() + while true: + var req: ptr SharedRequest + shared[].lock.acquire() + if shared[].running.load(): + req = shared[].queueHead + if not req.isNil: + shared[].queueHead = req.next + if shared[].queueHead.isNil: + shared[].queueTail = nil + shared[].lock.release() + if req.isNil: + break + + if RequestState(req.state.load()) == RequestState.Pending: + let host = $cast[cstring](addr req.host[0]) + when defined(libp2p_testing): + let workerRelease = shared[].workerRelease + if not workerRelease.isNil: + shared[].workerRelease = nil + discard shared[].workerEntered.fireSync() + let waitResult = workerRelease.waitSync(2.seconds) + shared[].workerGateReleased.store(waitResult.isOk and waitResult.get()) + try: + let resolved = resolveTAddress(host, Port(req.port), req.domain) + for address in resolved: + # getaddrinfo can return mapped literals even without AI_V4MAPPED. + if req.domain == Domain.AF_INET6 and address.isV4Mapped(): + continue + if req.resultsLen == MaxSystemAddresses: + break + req.results[req.resultsLen] = address + inc req.resultsLen + var expected = ord(RequestState.Pending) + discard req.state.compareExchange(expected, ord(RequestState.Done)) + except TransportAddressError: + var expected = ord(RequestState.Pending) + discard req.state.compareExchange(expected, ord(RequestState.Discarded)) + + shared[].lock.acquire() + req.next = shared[].completedHead + shared[].completedHead = req + shared[].lock.release() + discard shared[].responseSignal.fireSync() + + # Wake another idle worker during shutdown, then notify the event-loop + # thread that this worker no longer touches shared state. + discard shared[].requestSignal.fireSync() + discard shared[].workersRunning.fetchSub(1) + discard shared[].responseSignal.fireSync() + +proc finishRequest(self: SystemResolver, req: ptr SharedRequest) = + ## Complete (or skip, when cancelled) the future waiting on `req` and + ## free the shared request. Event-loop thread only. + var idx = -1 + for i, p in self.pending: + if p.req == req: + idx = i + break + if idx >= 0: + let fut = self.pending[idx].fut + self.pending.del(idx) + if not fut.finished(): + let deliver = RequestState(req.state.load()) == RequestState.Done + var res = newSeqOfCap[TransportAddress](if deliver: req.resultsLen else: 0) + if deliver: + for i in 0 ..< req.resultsLen: + res.add(req.results[i]) + fut.complete(res) + deallocShared(req) + +proc drainCompleted(self: SystemResolver) = + let shared = self.shared + shared[].lock.acquire() + var cur = shared[].completedHead + shared[].completedHead = nil + shared[].lock.release() + while not cur.isNil: + let req = cur + cur = req.next + self.finishRequest(req) + +proc dispatchLoop(self: SystemResolver) {.async: (raises: []).} = + while not self.stopped: + try: + await self.shared[].responseSignal.wait() + except AsyncError, CancelledError: + continue + self.drainCompleted() + +proc startWorkers(self: SystemResolver) {.raises: [TransportAddressError].} = + # Allocate workers lazily so unused resolvers consume no threads. + if not self.shared.isNil: + return + + let shared = cast[ptr SharedState](allocShared0(sizeof(SharedState))) + initLock(shared[].lock) + shared[].running.store(true) + shared[].workersRunning.store(0) + when defined(libp2p_testing): + shared[].workerEntered = self.workerEntered + shared[].workerRelease = self.workerRelease + shared[].workerGateReleased.store(false) + shared[].requestSignal = ThreadSignalPtr.new().valueOr: + deinitLock(shared[].lock) + deallocShared(shared) + raise + newException(TransportAddressError, "Failed to create request signal: " & error) + shared[].responseSignal = ThreadSignalPtr.new().valueOr: + discard shared[].requestSignal.close() + deinitLock(shared[].lock) + deallocShared(shared) + raise + newException(TransportAddressError, "Failed to create response signal: " & error) + + # createThread passes the address of the Thread object itself to pthread, so + # every object must already occupy its final, stable sequence slot. + var workers = newSeq[Thread[ptr SharedState]](self.workerCount) + var started = 0 + try: + for i in 0 ..< workers.len: + createThread(workers[i], workerLoop, shared) + discard shared[].workersRunning.fetchAdd(1) + inc started + except ResourceExhaustedError as exc: + workers.setLen(started) + shared[].running.store(false) + discard shared[].requestSignal.fireSync() + joinThreads(workers) + discard shared[].requestSignal.close() + discard shared[].responseSignal.close() + deinitLock(shared[].lock) + deallocShared(shared) + raise newException( + TransportAddressError, "Failed to spawn resolver worker: " & exc.msg, exc + ) + + self.shared = shared + self.workers = workers + self.dispatcherFut = self.dispatchLoop() + +proc new*( + T: type SystemResolver, rng: Rng, txtResolver: NameResolver = nil, workers: int = 2 +): T = + ## Create a SystemResolver which starts `workers` worker threads on first use. + ## `txtResolver` handles TXT queries (getaddrinfo cannot); when nil, a + ## `DnsResolver` over the system nameservers is used. Discovery is synchronous + ## during construction and can block the calling thread while reading system config. + doAssert workers >= 1 + + T( + txtResolver: + if txtResolver.isNil: + NameResolver(DnsResolver.new(getSystemNameServers(), rng)) + else: + txtResolver, + workerCount: workers, + ) + +when defined(libp2p_testing): + proc setWorkerTestGate*(self: SystemResolver, entered, release: ThreadSignalPtr) = + ## Delay the first worker lookup so shutdown behavior can be tested. + doAssert self.shared.isNil + doAssert not entered.isNil and not release.isNil + self.workerEntered = entered + self.workerRelease = release + self.workerGateReleased = false + + proc workerTestGateReleased*(self: SystemResolver): bool = + self.workerGateReleased + +method resolveIp*( + self: SystemResolver, address: string, port: Port, domain: Domain = Domain.AF_UNSPEC +): Future[seq[TransportAddress]] {. + async: (raises: [CancelledError, TransportAddressError]) +.} = + if self.stopped or address.len == 0 or address.len > 255 or '\0' in address: + return @[] + + if self.shared.isNil: + self.startWorkers() + + let + shared = self.shared + req = cast[ptr SharedRequest](allocShared0(sizeof(SharedRequest))) + fut = Future[seq[TransportAddress]].Raising([CancelledError]).init( + "systemresolver.resolveIp" + ) + copyMem(addr req.host[0], address.cstring, address.len) + req.port = uint16(port) + req.domain = domain + req.state.store(ord(RequestState.Pending)) + + proc cancellation(udata: pointer) {.gcsafe, raises: [].} = + var expected = ord(RequestState.Pending) + discard req.state.compareExchange(expected, ord(RequestState.Discarded)) + + fut.cancelCallback = cancellation + self.pending.add(PendingRequest(req: req, fut: fut)) + + shared[].lock.acquire() + if shared[].queueTail.isNil: + shared[].queueHead = req + else: + shared[].queueTail.next = req + shared[].queueTail = req + shared[].lock.release() + discard shared[].requestSignal.fireSync() + + trace "Resolving via getaddrinfo", address, port = uint16(port), domain + return await fut + +method resolveTxt*( + self: SystemResolver, address: string +): Future[seq[string]] {.async: (raises: [CancelledError]).} = + if self.stopped: + return @[] + return await self.txtResolver.resolveTxt(address) + +proc start*(self: SystemResolver) {.gcsafe.} = + ## Worker resources are created lazily by the next IP resolution. + if self.shared.isNil: + self.stopped = false + self.stopFut = nil + +proc stopWorkers(self: SystemResolver) {.async: (raises: []).} = + self.stopped = true + + if self.shared.isNil: + return + + let shared = self.shared + shared[].lock.acquire() + shared[].running.store(false) + shared[].lock.release() + discard shared[].requestSignal.fireSync() + + # Stop the dispatcher before using its signal to await worker exits. + if not self.dispatcherFut.isNil: + discard shared[].responseSignal.fireSync() + await noCancel self.dispatcherFut + self.dispatcherFut = nil + + # A worker may still be inside getaddrinfo, but waiting for it must not + # block the event loop. Every worker signals after decrementing this count. + while shared[].workersRunning.load() > 0: + try: + await shared[].responseSignal.wait() + except AsyncError, CancelledError: + continue + self.drainCompleted() + + # All workers have exited, so joining only releases their thread handles. + joinThreads(self.workers) + when defined(libp2p_testing): + self.workerGateReleased = shared[].workerGateReleased.load() + + # No worker can touch the shared state anymore: complete whatever is + # left, with results where the worker managed to produce them. + self.drainCompleted() + var cur = shared[].queueHead + while not cur.isNil: + let req = cur + cur = req.next + self.finishRequest(req) + + discard shared[].requestSignal.close() + discard shared[].responseSignal.close() + deinitLock(shared[].lock) + deallocShared(shared) + self.shared = nil + self.workers = @[] + +proc stop*(self: SystemResolver) {.async: (raises: []).} = + ## Wait for active OS lookups, discard queued lookups, and release workers. + ## Concurrent callers await the same cleanup; the event loop stays responsive. + if self.stopFut.isNil: + self.stopFut = self.stopWorkers() + await noCancel self.stopFut diff --git a/tests/libp2p/test_system_resolve.nim b/tests/libp2p/test_system_resolve.nim new file mode 100644 index 0000000000..f5b7109d58 --- /dev/null +++ b/tests/libp2p/test_system_resolve.nim @@ -0,0 +1,182 @@ +# SPDX-License-Identifier: Apache-2.0 OR MIT +# Copyright (c) Status Research & Development GmbH + +{.used.} + +import std/[sequtils, strutils, tables] +import chronos, chronos/threadsync +import ../../libp2p/nameresolving +import ../tools/[unittest, crypto, multiaddress, resolver, switch_builder] + +suite "System name resolution": + teardown: + checkTrackers() + + asyncTest "switch resolves localhost with a caller-managed system resolver": + let resolver = newNameResolver(rng()) + let source = + makeStandardSwitchBuilder(TcpAutoAddress).withNameResolver(resolver).build() + let destination = makeStandardSwitch(TcpAutoAddress) + defer: + await source.stop() + await destination.stop() + await resolver.stop() + let addresses = await resolver.resolveMAddress(ma("/dns4/localhost/tcp/4001")) + check ma("/ip4/127.0.0.1/tcp/4001") in addresses + await source.stop() + check (await resolver.resolveIp("localhost", 4001.Port)).len > 0 + await resolver.stop() + await destination.start() + await source.start() + check (await resolver.resolveIp("localhost", 4001.Port)).len == 0 + resolver.start() + let address = + ma("/dns4/localhost") & destination.peerInfo.addrs[0][1 ..^ 1].tryGet() + await source.connect(destination.peerInfo.peerId, @[address]) + check source.isConnected(destination.peerInfo.peerId) + await source.stop() + check (await resolver.resolveIp("localhost", 4001.Port)).len > 0 + + asyncTest "factory preserves explicit DNS servers": + for servers in [@[initTAddress("192.0.2.1:5353")], newSeq[TransportAddress]()]: + let resolver = newNameResolver(rng(), servers) + check resolver.nameServers == servers + if servers.len == 0: + check (await resolver.resolveIp("localhost", 4001.Port)).len == 0 + check (await resolver.resolveTxt("_dnsaddr.test.io")).len == 0 + + asyncTest "requested family and port are preserved": + let resolver = SystemResolver.new(rng()) + defer: + await resolver.stop() + let ipv4 = await resolver.resolveIp("localhost", 4001.Port, Domain.AF_INET) + check: + ipv4.len > 0 + ipv4.allIt(it.family == AddressFamily.IPv4 and it.port == 4001.Port) + check (await resolver.resolveIp("::1", 4002.Port, Domain.AF_INET6)) == + @[initTAddress("[::1]:4002")] + + asyncTest "dns6 rejects IPv4 and mapped IPv4 literals": + let resolver = SystemResolver.new(rng()) + defer: + await resolver.stop() + for host in ["127.0.0.1", "::ffff:127.0.0.1"]: + check (await resolver.resolveIp(host, 0.Port, Domain.AF_INET6)).len == 0 + check (await resolver.resolveMAddress(ma("/dns6/" & host & "/tcp/4001"))).len == 0 + + asyncTest "invalid host input returns no addresses": + let resolver = SystemResolver.new(rng()) + defer: + await resolver.stop() + for host in ["", repeat('a', 256), "localhost\0.invalid"]: + check (await resolver.resolveIp(host, 0.Port)).len == 0 + + asyncTest "TXT queries use the supplied fallback": + let fallback = MockResolver.new() + fallback.txtResponses["_dnsaddr.test.io"] = @["dnsaddr=/ip4/127.0.0.1/tcp/4001"] + let resolver = SystemResolver.new(rng(), txtResolver = fallback) + defer: + await resolver.stop() + check (await resolver.resolveTxt("_dnsaddr.test.io")) == + @["dnsaddr=/ip4/127.0.0.1/tcp/4001"] + + asyncTest "concurrent requests preserve their ports": + let resolver = SystemResolver.new(rng()) + defer: + await resolver.stop() + let futures = (0 ..< 16).mapIt(resolver.resolveIp("localhost", Port(4001 + it))) + for i, future in futures: + let addresses = await future + check: + addresses.len > 0 + addresses.allIt(it.port == Port(4001 + i)) + + asyncTest "a stalled lookup does not block the event loop": + let + entered = ThreadSignalPtr.new().expect("create worker-entered signal") + release = ThreadSignalPtr.new().expect("create worker-release signal") + resolver = SystemResolver.new(rng(), workers = 1) + resolver.setWorkerTestGate(entered, release) + defer: + discard release.fireSync() + await resolver.stop() + discard entered.close() + discard release.close() + + let lookup = resolver.resolveIp("localhost", 4001.Port) + await entered.wait() + await sleepAsync(10.milliseconds) + check not lookup.finished() + discard release.fireSync() + check (await lookup).len > 0 + + asyncTest "shutdown yields to the event loop and discards the backlog": + let + entered = ThreadSignalPtr.new().expect("create worker-entered signal") + release = ThreadSignalPtr.new().expect("create worker-release signal") + resolver = SystemResolver.new(rng(), workers = 1) + resolver.setWorkerTestGate(entered, release) + defer: + discard release.fireSync() + await resolver.stop() + discard entered.close() + discard release.close() + + let inFlight = resolver.resolveIp("localhost", 4001.Port) + await entered.wait() + let queued = (0 ..< 8).mapIt(resolver.resolveIp("localhost", Port(4002 + it))) + proc releaseWorker() {.async: (raises: [CancelledError]).} = + await sleepAsync(10.milliseconds) + discard release.fireSync() + + let releaseFuture = releaseWorker() + let closing = resolver.stop() + let alsoClosing = resolver.stop() + check: + not closing.finished() + not alsoClosing.finished() + await alsoClosing + await closing + await releaseFuture + check resolver.workerTestGateReleased() + for lookup in queued: + check (await lookup).len == 0 + discard await inFlight + + asyncTest "cancelling active and queued requests leaves workers usable": + let + entered = ThreadSignalPtr.new().expect("create worker-entered signal") + release = ThreadSignalPtr.new().expect("create worker-release signal") + resolver = SystemResolver.new(rng(), workers = 1) + resolver.setWorkerTestGate(entered, release) + defer: + discard release.fireSync() + await resolver.stop() + discard entered.close() + discard release.close() + + let inFlight = resolver.resolveIp("localhost", 4001.Port) + await entered.wait() + let queued = resolver.resolveIp("localhost", 4002.Port) + await queued.cancelAndWait() + await inFlight.cancelAndWait() + discard release.fireSync() + check (await resolver.resolveIp("localhost", 4003.Port)).len > 0 + + asyncTest "stop is idempotent and start reopens the worker pool": + let resolver = SystemResolver.new(rng()) + check (await resolver.resolveIp("localhost", 4001.Port)).len > 0 + await resolver.stop() + await resolver.stop() + check (await resolver.resolveIp("localhost", 4001.Port)).len == 0 + resolver.start() + check (await resolver.resolveIp("localhost", 4002.Port)).len > 0 + await resolver.stop() + + asyncTest "stop and reopen before workers are created": + let resolver = SystemResolver.new(rng()) + await resolver.stop() + check (await resolver.resolveTxt("_dnsaddr.test.io")).len == 0 + resolver.start() + check (await resolver.resolveIp("localhost", 4001.Port)).len > 0 + await resolver.stop()