diff --git a/bitchatTests/Nostr/TorEgressVerifierTests.swift b/bitchatTests/Nostr/TorEgressVerifierTests.swift index fd4c0202..9f3a4854 100644 --- a/bitchatTests/Nostr/TorEgressVerifierTests.swift +++ b/bitchatTests/Nostr/TorEgressVerifierTests.swift @@ -19,20 +19,72 @@ struct TorEgressVerifierTests { private let lock = NSLock() private var _now = Date(timeIntervalSince1970: 1_000_000) private var _result: TorEgressVerifier.ProbeResult = .verifiedTor - private(set) var probeCount = 0 + private var _hanging = false + private var _gated = false + private var _released = false + private var _probeCount = 0 + private var _cancelledCount = 0 var now: Date { lock.lock(); defer { lock.unlock() }; return _now } + var probeCount: Int { + lock.lock(); defer { lock.unlock() }; return _probeCount + } + /// Number of hung probes that observed cooperative cancellation. + var cancelledCount: Int { + lock.lock(); defer { lock.unlock() }; return _cancelledCount + } func advance(_ seconds: TimeInterval) { lock.lock(); _now = _now.addingTimeInterval(seconds); lock.unlock() } func setResult(_ r: TorEgressVerifier.ProbeResult) { lock.lock(); _result = r; lock.unlock() } + /// When `true`, probes park forever and only exit via cooperative + /// cancellation — models a canary request wedged by + /// `waitsForConnectivity` deferring the request timer. + func setHanging(_ hanging: Bool) { + lock.lock(); _hanging = hanging; lock.unlock() + } + /// When `true`, probes wait for `release()` before returning — models + /// a slow-but-completing canary for join-semantics tests. + func setGated(_ gated: Bool) { + lock.lock(); _gated = gated; lock.unlock() + } + func release() { + lock.lock(); _released = true; lock.unlock() + } + /// Suspends until at least `n` probes have started. + func waitUntilProbeCount(atLeast n: Int) async { + while probeCount < n { + try? await Task.sleep(nanoseconds: 1_000_000) + } + } func makeProbe() -> @Sendable () async -> TorEgressVerifier.ProbeResult { return { [self] in - lock.lock(); probeCount += 1; let r = _result; lock.unlock() + lock.lock() + _probeCount += 1 + let r = _result + let hang = _hanging + let gated = _gated + lock.unlock() + if hang { + // Park until cancelled (verifier watchdog or invalidate()); + // cancellation-responsive so no task outlives the test. + while !Task.isCancelled { + try? await Task.sleep(nanoseconds: 2_000_000) + } + lock.lock(); _cancelledCount += 1; lock.unlock() + return .unreachable("hung probe cancelled") + } + if gated { + while !Task.isCancelled { + lock.lock(); let released = _released; lock.unlock() + if released { break } + try? await Task.sleep(nanoseconds: 1_000_000) + } + } return r } } @@ -41,8 +93,19 @@ struct TorEgressVerifierTests { } } - private func makeVerifier(_ h: Harness, ttl: TimeInterval = 300, minRetry: TimeInterval = 5) -> TorEgressVerifier { - TorEgressVerifier(ttl: ttl, minRetryInterval: minRetry, now: h.nowProvider(), probe: h.makeProbe()) + private func makeVerifier( + _ h: Harness, + ttl: TimeInterval = 300, + minRetry: TimeInterval = 5, + probeTimeout: TimeInterval = TorEgressVerifier.defaultProbeTimeout + ) -> TorEgressVerifier { + TorEgressVerifier( + ttl: ttl, + minRetryInterval: minRetry, + probeTimeout: probeTimeout, + now: h.nowProvider(), + probe: h.makeProbe() + ) } @Test("verifiedTor allows and is cached within TTL (single probe)") @@ -217,4 +280,124 @@ struct TorEgressVerifierTests { _ = await v.verify() #expect(await v.lastProbeResult() == .notTor) } + + // MARK: - Liveness (probe timeout watchdog + invalidate cancellation) + + @Test("a probe that never completes is bounded by probeTimeout and fails closed") + func hungProbeIsBoundedByTimeout() async { + let h = Harness() + h.setHanging(true) + let v = makeVerifier(h, ttl: 300, minRetry: 0, probeTimeout: 0.05) + + // Without the watchdog this would wedge: waitsForConnectivity can + // defer the request timer, leaving the canary bounded only by the + // 7-day resource timeout. + #expect(await v.verify() == false) + let last = await v.lastProbeResult() + switch last { + case .unreachable: + break // fail-closed timeout verdict recorded + default: + Issue.record("expected .unreachable after probe timeout, got \(String(describing: last))") + } + #expect(v.hasFreshVerification == false) + + // The hung probe task itself was cancelled (URLSession task would be + // torn down), not abandoned. + while h.cancelledCount < 1 { + try? await Task.sleep(nanoseconds: 1_000_000) + } + #expect(h.cancelledCount == 1) + + // The in-flight slot was cleared: the next verify() starts a fresh + // probe (does not join the hung one) and recovers. + h.setHanging(false) + h.setResult(.verifiedTor) + #expect(await v.verify() == true) + #expect(h.probeCount == 2) + #expect(v.hasFreshVerification == true) + } + + @Test("invalidate() cancels the in-flight probe and the awaiting caller fails closed") + func invalidateCancelsInFlightProbe() async { + let h = Harness() + h.setHanging(true) + // Long (real-time) timeout and throttle: only invalidate() can + // unblock the caller, and only invalidate() clearing the throttle + // lets the follow-up probe run without advancing the clock. + let v = makeVerifier(h, ttl: 300, minRetry: 600, probeTimeout: 600) + + let first = Task { await v.verify() } + await h.waitUntilProbeCount(atLeast: 1) + await v.invalidate() + + // The awaiting caller resolves promptly (no 600s wait) and refuses. + #expect(await first.value == false) + #expect(v.hasFreshVerification == false) + + // The hung probe observed cancellation (a Tor restart genuinely + // shakes the wedged canary request). + while h.cancelledCount < 1 { + try? await Task.sleep(nanoseconds: 1_000_000) + } + + // Recovery: a fresh verify() runs a NEW probe — it neither joins the + // cancelled one nor inherits its throttle/last-result state. + h.setHanging(false) + h.setResult(.verifiedTor) + #expect(await v.verify() == true) + #expect(h.probeCount == 2) + #expect(v.hasFreshVerification == true) + } + + @Test("recovery after a hung probe survives invalidate + Tor restart cycle") + func hungThenInvalidatedThenRecovers() async { + let h = Harness() + h.setHanging(true) + let v = makeVerifier(h, ttl: 300, minRetry: 5, probeTimeout: 600) + + // Wedge one probe, then simulate a Tor restart mid-flight. + let wedged = Task { await v.verify() } + await h.waitUntilProbeCount(atLeast: 1) + await v.invalidate() + #expect(await wedged.value == false) + + // Canary still down right after restart: fresh probe, fail closed — + // and the throttle applies to the FRESH result (bounded retry intact). + h.setHanging(false) + h.setResult(.unreachable("circuit not built")) + #expect(await v.verify() == false) + #expect(h.probeCount == 2) + h.advance(1) + #expect(await v.verify() == false) + #expect(h.probeCount == 2) // throttled, no hammering + + // Canary returns after the retry window: verification recovers. + h.setResult(.verifiedTor) + h.advance(5) + #expect(await v.verify() == true) + #expect(v.hasFreshVerification == true) + } + + @Test("concurrent verify() callers still share a single in-flight probe") + func concurrentCallersShareOneProbe() async { + let h = Harness() + h.setResult(.verifiedTor) + h.setGated(true) + let v = makeVerifier(h, ttl: 300, minRetry: 0) + + let t1 = Task { await v.verify() } + await h.waitUntilProbeCount(atLeast: 1) + let t2 = Task { await v.verify() } + // Give t2 a chance to join the in-flight probe before releasing it. + // Either way the invariant holds: t2 joins the shared probe, or (if + // scheduled after completion) hits the fresh TTL cache — exactly one + // probe runs. + try? await Task.sleep(nanoseconds: 20_000_000) + h.release() + + #expect(await t1.value == true) + #expect(await t2.value == true) + #expect(h.probeCount == 1) + } } diff --git a/localPackages/Arti/Sources/TorEgressVerifier.swift b/localPackages/Arti/Sources/TorEgressVerifier.swift index b8645be5..4e1ae4fe 100644 --- a/localPackages/Arti/Sources/TorEgressVerifier.swift +++ b/localPackages/Arti/Sources/TorEgressVerifier.swift @@ -44,6 +44,20 @@ import Foundation /// gate, GeoRelayDirectory backoff) re-probe and succeed once the canary /// is reachable again. /// +/// Liveness: +/// - Every probe is hard-bounded by `probeTimeout` via an independent async +/// watchdog. The proxied session sets `waitsForConnectivity = true`, which +/// can defer the per-request timer indefinitely, leaving the request +/// bounded only by the default 7-day resource timeout — without the +/// watchdog a hung canary would wedge every subsequent `verify()` caller. +/// On timeout the probe task is cancelled (cooperatively cancelling the +/// underlying `URLSessionTask`), the verdict is the fail-closed +/// `.unreachable`, and the in-flight slot is cleared so the next +/// `verify()` (after the retry throttle) starts a fresh probe. +/// - `invalidate()` cancels any in-flight probe and clears all cached state +/// (including the retry throttle), so a Tor restart/dormant/shutdown +/// genuinely resets the verifier: a hung probe cannot survive it. +/// /// The probe is injectable so the policy/caching logic is unit-tested without a /// live network (see `TorEgressVerifierTests`). public actor TorEgressVerifier { @@ -56,6 +70,11 @@ public actor TorEgressVerifier { case unreachable(String) } + /// Hard upper bound for a single canary probe, enforced independently of + /// URLSession timers (see the "Liveness" section of the type doc). Matches + /// the live probe's per-request timeout. + public static let defaultProbeTimeout: TimeInterval = 20 + private let probe: @Sendable () async -> ProbeResult private let now: @Sendable () -> Date private let ttl: TimeInterval @@ -63,11 +82,18 @@ public actor TorEgressVerifier { /// persistent `.unreachable` cannot hammer the canary endpoint on every /// reconnect burst. private let minRetryInterval: TimeInterval + /// Outer wall-clock bound on a single probe (watchdog; fail-closed). + private let probeTimeout: TimeInterval private var lastVerifiedAt: Date? private var lastProbeAt: Date? private var lastResult: ProbeResult? private var inFlight: Task? + /// Bumped whenever a new probe starts or `invalidate()` runs. A completing + /// probe only records its outcome (cache/throttle) and clears `inFlight` + /// if its generation is still current, so a cancelled/superseded probe + /// cannot clobber state owned by a fresh one (actor-reentrancy safety). + private var probeGeneration = 0 /// Lock-protected mirror of "verified within TTL" so synchronous gates /// (e.g. `NostrRelayManager`'s connect path) can consult the cache without @@ -95,18 +121,26 @@ public actor TorEgressVerifier { public init( ttl: TimeInterval, minRetryInterval: TimeInterval = 5.0, + probeTimeout: TimeInterval = TorEgressVerifier.defaultProbeTimeout, now: @escaping @Sendable () -> Date = Date.init, probe: @escaping @Sendable () async -> ProbeResult ) { self.ttl = ttl self.minRetryInterval = minRetryInterval + self.probeTimeout = probeTimeout self.now = now self.probe = probe } /// Drop any cached verification (e.g. after a Tor restart or when the - /// network path changes). The next `verify()` re-probes. + /// network path changes) AND cancel any in-flight probe. The next + /// `verify()` starts a fresh probe — it neither joins the cancelled one + /// nor is throttled by its outcome, so a probe hung from before a Tor + /// restart cannot wedge callers after it. public func invalidate() { + probeGeneration += 1 + inFlight?.cancel() + inFlight = nil lastVerifiedAt = nil lastProbeAt = nil lastResult = nil @@ -137,10 +171,15 @@ public actor TorEgressVerifier { } if let inFlight { return await inFlight.value } - let task = Task { await self.runProbe() } + probeGeneration += 1 + let generation = probeGeneration + let task = Task { await self.runProbe(generation: generation) } inFlight = task let allowed = await task.value - inFlight = nil + // Only clear the slot if this probe is still the current one: an + // `invalidate()` while we were suspended has already cleared it and a + // newer probe may occupy it (do not clobber the fresh task). + if probeGeneration == generation { inFlight = nil } return allowed } @@ -158,8 +197,13 @@ public actor TorEgressVerifier { } } - private func runProbe() async -> Bool { - let result = await probe() + private func runProbe(generation: Int) async -> Bool { + let result = await boundedProbe() + // Superseded by `invalidate()` (Tor restart/dormant/shutdown) while the + // probe ran: its verdict predates the reset, so discard it — recording + // it would re-seed the throttle/cache that invalidate() just cleared. + // Fail closed for the callers that were awaiting this probe. + guard generation == probeGeneration else { return false } lastProbeAt = now() lastResult = result switch result { @@ -185,6 +229,95 @@ public actor TorEgressVerifier { return false } } + + /// Runs the injected probe raced against `probeTimeout`, guaranteeing a + /// result in bounded time regardless of URLSession timer behavior (the + /// proxied session's `waitsForConnectivity` can defer the per-request + /// timeout indefinitely). Whichever side loses the race is cancelled: + /// - on timeout, the probe task is cancelled (URLSession's async APIs + /// cancel the underlying `URLSessionTask` cooperatively) and the result + /// is the fail-closed `.unreachable`; + /// - on completion, the watchdog's sleep is cancelled so no timer lingers. + /// Cancelling the enclosing task (`invalidate()`) resolves immediately as + /// `.unreachable` and cancels both sides. + /// + /// `nonisolated` so the race body never re-enters the actor; it touches + /// only immutable `Sendable` state. + nonisolated private func boundedProbe() async -> ProbeResult { + let probe = self.probe + let timeout = self.probeTimeout + let race = ProbeRace() + return await withTaskCancellationHandler { + await withCheckedContinuation { (continuation: CheckedContinuation) in + race.install(continuation) + let probeTask = Task { race.finish(await probe()) } + let watchdog = Task { + try? await Task.sleep(nanoseconds: UInt64(max(0, timeout) * 1_000_000_000)) + guard !Task.isCancelled else { return } + race.finish(.unreachable("probe timed out after \(Int(timeout))s")) + } + race.register(probeTask: probeTask, watchdog: watchdog) + } + } onCancel: { + race.finish(.unreachable("probe cancelled")) + } + } + + /// Resolve-once rendezvous for the probe/watchdog race. Lock-protected + /// (never held across an await); the first `finish()` wins, resumes the + /// continuation exactly once, and cancels both tasks. + private final class ProbeRace: @unchecked Sendable { + private let lock = NSLock() + private var continuation: CheckedContinuation? + private var pendingResult: ProbeResult? + private var resolved = false + private var probeTask: Task? + private var watchdog: Task? + + func install(_ continuation: CheckedContinuation) { + lock.lock() + if let result = pendingResult { + // finish() ran before the continuation existed (e.g. the + // enclosing task was already cancelled): resolve immediately. + pendingResult = nil + lock.unlock() + continuation.resume(returning: result) + return + } + self.continuation = continuation + lock.unlock() + } + + func register(probeTask: Task, watchdog: Task) { + lock.lock() + if resolved { + lock.unlock() + probeTask.cancel() + watchdog.cancel() + return + } + self.probeTask = probeTask + self.watchdog = watchdog + lock.unlock() + } + + func finish(_ result: ProbeResult) { + lock.lock() + guard !resolved else { lock.unlock(); return } + resolved = true + let continuation = self.continuation + self.continuation = nil + if continuation == nil { pendingResult = result } + let probeTask = self.probeTask + let watchdog = self.watchdog + self.probeTask = nil + self.watchdog = nil + lock.unlock() + probeTask?.cancel() + watchdog?.cancel() + continuation?.resume(returning: result) + } + } } // MARK: - Live probe @@ -198,9 +331,13 @@ public extension TorEgressVerifier { /// /// Follow-up (see PR): make the canary endpoint configurable and add an /// onion-service canary so verification does not depend on a single host. + /// Note: the per-request `timeoutInterval` below is best-effort only — the + /// proxied session's `waitsForConnectivity` can defer it. The authoritative + /// bound is the verifier's `probeTimeout` watchdog, whose cancellation + /// propagates into `session.data(for:)` and cancels the URLSessionTask. static func liveProbe( endpoint: URL = URL(string: "https://check.torproject.org/api/ip")!, - timeout: TimeInterval = 20 + timeout: TimeInterval = TorEgressVerifier.defaultProbeTimeout ) -> @Sendable () async -> ProbeResult { return { var request = URLRequest(url: endpoint)