From c390d57ad76acbd66334530985cd62a887861569 Mon Sep 17 00:00:00 2001 From: Kaede Akino Date: Sat, 19 Sep 2026 13:01:14 +0800 Subject: [PATCH] Support reconnect in port forwarding --- .../SSH/PortForwardingController.swift | 191 ++++-- .../SSH/SSHForwardingSession.swift | 77 ++- Memory/ARCHITECTURE.md | 24 +- TFI/App/MobileAppDependencies.swift | 2 + TFI/App/MobileAppModel.swift | 2 +- TFITests/MobileForwardingLifecycleTests.swift | 58 ++ TFMTests/ArchitectureBoundaryTests.swift | 5 +- TFMTests/PortForwardingJourneyTests.swift | 18 +- TFMTests/PortForwardingReconnectTests.swift | 582 ++++++++++++++++++ TFMTests/SSHForwardingWireTests.swift | 70 ++- TFMTests/SSHKeyboardInteractiveTests.swift | 2 +- 11 files changed, 945 insertions(+), 86 deletions(-) create mode 100644 TFMTests/PortForwardingReconnectTests.swift diff --git a/Frameworks/TermindRemoteInfrastructure/SSH/PortForwardingController.swift b/Frameworks/TermindRemoteInfrastructure/SSH/PortForwardingController.swift index cef1e329..870034ad 100644 --- a/Frameworks/TermindRemoteInfrastructure/SSH/PortForwardingController.swift +++ b/Frameworks/TermindRemoteInfrastructure/SSH/PortForwardingController.swift @@ -10,6 +10,7 @@ public nonisolated enum PortForwardingRuntimeState: Hashable, Sendable { case awaitingTrust case awaitingAuthentication case connecting + case reconnecting case starting case running(activeConnections: Int) case failed(String) @@ -33,6 +34,7 @@ public nonisolated enum PortForwardingRuntimeState: Hashable, Sendable { case .awaitingTrust: String(localized: "Awaiting trust") case .awaitingAuthentication: String(localized: "Awaiting authentication") case .connecting: String(localized: "Connecting") + case .reconnecting: String(localized: "Reconnecting") case .starting: String(localized: "Starting") case .running(let count): count == 0 @@ -49,7 +51,7 @@ public nonisolated enum PortForwardingRuntimeState: Hashable, Sendable { @Observable public final class PortForwardingConnection { public let hostID: Host.ID - public let connection = RemoteSessionConnectionController() + public let connection: RemoteSessionConnectionController public var journey: ConnectionJourney { connection.journey } public private(set) var authenticationPrompt: SSHAuthenticationPrompt? public fileprivate(set) var hostKeyRequestID: UUID? @@ -59,17 +61,21 @@ public final class PortForwardingConnection { fileprivate var session: (any SSHForwardingSessionProtocol)? fileprivate var task: Task? fileprivate var connected = false + fileprivate var ready = false + fileprivate var silentAttempt = false + fileprivate var configuration: SSHConnectionConfiguration? fileprivate var connectionCounts: [PortForwardingRule.ID: Int] = [:] - fileprivate init(host: Host, route: [Host]) { + fileprivate init(host: Host, route: [Host], reconnect: SessionReconnectController) { hostID = host.id self.route = route + connection = RemoteSessionConnectionController(reconnect: reconnect) } fileprivate func setPrompt(_ prompt: SSHAuthenticationPrompt?) { authenticationPrompt = prompt } } -/// Runtime intent exists only between explicit Start and Stop/failure. No defaults or retry timers. +/// Explicit Start owns runtime intent. Established connections recover silently until Stop or a terminal failure. @Observable public final class PortForwardingController { public typealias SessionFactory = @@ -79,20 +85,26 @@ public final class PortForwardingController { private let hosts: HostLibraryModel private let makeSession: SessionFactory + private let makeReconnectController: @MainActor () -> SessionReconnectController public private(set) var states: [PortForwardingRule.ID: PortForwardingRuntimeState] = [:] private var connections: [Host.ID: PortForwardingConnection] = [:] private var presentations: [PortForwardingRule.ID: PortForwardingConnection] = [:] private var generations: [PortForwardingRule.ID: UUID] = [:] private var ruleTasks: [PortForwardingRule.ID: Task] = [:] + private var cleanupTasks: [Host.ID: Task] = [:] private var savingAuthentication: PortForwardingConnection? public init( hosts: HostLibraryModel, makeSession: @escaping SessionFactory = { SSHForwardingSession(connectionHandler: $0, failureHandler: $1) + }, + makeReconnectController: @escaping @MainActor () -> SessionReconnectController = { + SessionReconnectController() } ) { self.hosts = hosts + self.makeReconnectController = makeReconnectController self.makeSession = makeSession } @@ -107,7 +119,8 @@ public final class PortForwardingController { if let existing = connections[host.id], !activeIDs(existing).isEmpty { context = existing } else { - context = PortForwardingConnection(host: host, route: hosts.route(for: host)) + context = PortForwardingConnection( + host: host, route: hosts.route(for: host), reconnect: makeReconnectController()) connections[host.id] = context } generations[id] = UUID() @@ -115,6 +128,8 @@ public final class PortForwardingController { states[id] = .preparing if context.connected { startRule(rule, in: context) + } else if context.connection.isRecovering { + states[id] = .reconnecting } else if context.task == nil && context.authenticationPrompt == nil { open(context) } else { @@ -133,8 +148,8 @@ public final class PortForwardingController { let session = context.session let previous = ruleTasks[id] ruleTasks[id] = Task { - if let previous { await previous.value } if let session { await session.stop(id) } + if let previous { await previous.value } } if activeIDs(context).isEmpty { retire(context) } } @@ -237,18 +252,24 @@ public final class PortForwardingController { } private func open(_ context: PortForwardingConnection) { - guard let host = hosts.host(context.hostID), !activeIDs(context).isEmpty else { return } + guard let host = hosts.host(context.hostID), connections[host.id] === context, + !activeIDs(context).isEmpty, context.task == nil, !context.connected + else { return } + let silently = context.connection.isRecovering let attempt = UUID() context.attemptID = attempt - context.journey.begin( - host: host, route: hosts.route(for: host), purpose: .portForwarding, - username: context.submissions[host.id]?.username ?? hosts.identity(for: host)?.username) - update(context, state: .preparing) + context.silentAttempt = silently + if !silently { + context.journey.begin( + host: host, route: hosts.route(for: host), purpose: .portForwarding, + username: context.submissions[host.id]?.username ?? hosts.identity(for: host)?.username) + update(context, state: .preparing) + } let session = makeSession( - { [weak self, weak context] id, count in + { [weak self, weak context] id, generation, count in Task { @MainActor in guard let self, let context, self.isCurrent(context, attempt: attempt), - self.generations[id] != nil, self.presentations[id] === context + self.generations[id] == generation, self.presentations[id] === context else { return } context.connectionCounts[id] = count if case .running = self.states[id], self.states[id] != .running(activeConnections: count) { @@ -259,21 +280,34 @@ public final class PortForwardingController { { [weak self, weak context] message in Task { @MainActor in guard let self, let context, self.isCurrent(context, attempt: attempt) else { return } - self.fail(context, message: message) + self.recover(context, message: message) } }) context.session = session + let cleanup = cleanupTasks[context.hostID] context.task = Task { [weak self, weak context] in guard let self, let context else { return } do { - let preparation = try hosts.prepareSSHConnection( - for: host, submissions: context.submissions, - temporaryHostKeys: context.connection.temporaryHostKeys) - let configuration = try await preparation.resolve(trustPolicy: .allowChallenge) + await cleanup?.value + try Task.checkCancellation() + let configuration: SSHConnectionConfiguration + if silently { + guard let snapshot = context.configuration else { throw TermindSSHError.missingCredential } + configuration = snapshot + } else { + let preparation = try hosts.prepareSSHConnection( + for: host, submissions: context.submissions, + temporaryHostKeys: context.connection.temporaryHostKeys) + configuration = try await preparation.resolve(trustPolicy: .allowChallenge) + } guard isCurrent(context, attempt: attempt) else { throw CancellationError() } - update(context, state: .connecting) - try await session.connect( - SSHConnectionRequest( + if silently { + try await session.connect( + SSHConnectionRequest(configuration: configuration, interactionPolicy: .forbidden) + .enforcingInteractionPolicy(), progress: nil) + } else { + update(context, state: .connecting) + let request = SSHConnectionRequest( configuration: configuration, authorizeHostKey: { [weak self, weak context] challenge in guard let self, let context else { throw CancellationError() } @@ -282,17 +316,23 @@ public final class PortForwardingController { keyboardInteractive: { [weak self, weak context] challenge in guard let self, let context else { throw CancellationError() } return try await self.answer(challenge, in: context, attempt: attempt) - }), - progress: { [weak self, weak context] event in - Task { @MainActor in - guard let self, let context, self.isCurrent(context, attempt: attempt) else { return } - context.journey.receive(event) - } - }) + }) + try await session.connect( + request, + progress: { [weak self, weak context] event in + Task { @MainActor in + guard let self, let context, self.isCurrent(context, attempt: attempt) else { return } + context.journey.receive(event) + } + }) + } try Task.checkCancellation() guard isCurrent(context, attempt: attempt) else { throw CancellationError() } context.task = nil context.connected = true + if !silently { + context.configuration = hosts.configurationForAdditionalConnection(from: configuration) + } context.submissions.removeAll() context.connection.temporaryHostKeys.removeAll() for id in activeIDs(context) { @@ -304,7 +344,13 @@ public final class PortForwardingController { context.task = nil context.attemptID = nil context.session = nil - if let ssh = error as? TermindSSHError, + if silently { + if Self.requiresManualRetry(error) { + fail(context, message: error.localizedDescription, presentJourney: false) + } else { + recover(context, message: error.localizedDescription) + } + } else if let ssh = error as? TermindSSHError, ssh.isLocalAuthenticationRequirement || ssh.shouldRequestAuthentication { requestAuthentication(context, error: ssh) @@ -368,51 +414,112 @@ public final class PortForwardingController { } private func startRule(_ rule: PortForwardingRule, in context: PortForwardingConnection) { - guard let token = generations[rule.id], let session = context.session else { return } + guard let token = generations[rule.id], let session = context.session, let attempt = context.attemptID else { + return + } + let silently = context.silentAttempt context.connectionCounts[rule.id] = 0 - states[rule.id] = .starting + states[rule.id] = silently ? .reconnecting : .starting let previous = ruleTasks[rule.id] ruleTasks[rule.id] = Task { [weak self] in if let previous { await previous.value } - guard let self, generations[rule.id] == token else { return } + guard let self, generations[rule.id] == token, isCurrent(context, attempt: attempt) else { return } do { - try await session.start(rule) - guard generations[rule.id] == token else { return } + try await session.start(rule, generation: token) + guard generations[rule.id] == token, isCurrent(context, attempt: attempt) else { + await session.stop(rule.id) + return + } states[rule.id] = .running(activeConnections: context.connectionCounts[rule.id] ?? 0) - context.journey.succeed() + if !context.ready { + context.ready = true + context.connection.markConnected() + } + if !silently { context.journey.succeed() } } catch { - guard generations[rule.id] == token else { return } + guard generations[rule.id] == token, isCurrent(context, attempt: attempt) else { return } + // Transport cleanup can finish a pending remote listen before its failure callback arrives. + if error is CancellationError + || (error as? TermindSSHError).map({ + if case .disconnected = $0 { return true } + return false + }) == true + { + recover(context, message: error.localizedDescription) + return + } generations[rule.id] = nil states[rule.id] = .failed(error.localizedDescription) if activeIDs(context).isEmpty { retire(context) - context.journey.fail(error.localizedDescription) + if !silently { context.journey.fail(error.localizedDescription) } } } } } - private func fail(_ context: PortForwardingConnection, message: String) { + private static func requiresManualRetry(_ error: any Error) -> Bool { + if error is SSHInteractionRequired || error is CancellationError { return true } + guard let ssh = error as? TermindSSHError else { return false } + if ssh.isLocalAuthenticationRequirement || ssh.shouldRequestAuthentication || ssh.shouldRequestHostKey { + return true + } + switch ssh { + case .untrustedHost, .mismatchedHostKey, .hostKeyChanged: return true + default: return false + } + } + + private func recover(_ context: PortForwardingConnection, message: String) { + guard context.connection.hasConnectedOnce, context.configuration != nil else { + fail(context, message: message) + return + } + closeTransport(context) + context.journey.hide() + update(context, state: .reconnecting) + context.connection.scheduleRecovery(after: message) { [weak self, weak context] in + guard let self, let context, self.connections[context.hostID] === context, + context.connection.isRecovering + else { return } + self.open(context) + } + } + + private func fail(_ context: PortForwardingConnection, message: String, presentJourney: Bool = true) { for id in activeIDs(context) { generations[id] = nil states[id] = .failed(message) } retire(context) - context.journey.fail(message) + if presentJourney { context.journey.fail(message) } } - private func retire(_ context: PortForwardingConnection) { + private func closeTransport(_ context: PortForwardingConnection) { context.attemptID = nil + let opening = context.task context.task?.cancel() context.task = nil context.connected = false + context.ready = false + context.connectionCounts.removeAll() + let session = context.session + context.session = nil + let previous = cleanupTasks[context.hostID] + cleanupTasks[context.hostID] = Task { + await previous?.value + await opening?.value + await session?.disconnect() + } + } + + private func retire(_ context: PortForwardingConnection) { + closeTransport(context) context.setPrompt(nil) context.hostKeyRequestID = nil context.submissions.removeAll() + context.configuration = nil context.connection.disconnect() - let session = context.session - context.session = nil - if let session { Task { await session.disconnect() } } if connections[context.hostID] === context { connections[context.hostID] = nil } } } diff --git a/Frameworks/TermindRemoteInfrastructure/SSH/SSHForwardingSession.swift b/Frameworks/TermindRemoteInfrastructure/SSH/SSHForwardingSession.swift index 52e4e485..0cfcd70b 100644 --- a/Frameworks/TermindRemoteInfrastructure/SSH/SSHForwardingSession.swift +++ b/Frameworks/TermindRemoteInfrastructure/SSH/SSHForwardingSession.swift @@ -15,13 +15,13 @@ nonisolated struct SSHForwardingStatistics: Sendable { public protocol SSHForwardingSessionProtocol: Actor { func connect(_ request: SSHConnectionRequest, progress: SSHRouteProgressHandler?) async throws - func start(_ rule: PortForwardingRule) async throws + func start(_ rule: PortForwardingRule, generation: UUID) async throws func stop(_ id: PortForwardingRule.ID) async func disconnect() async } public actor SSHForwardingSession: SSHForwardingSessionProtocol { - public typealias ConnectionHandler = @Sendable (PortForwardingRule.ID, Int) -> Void + public typealias ConnectionHandler = @Sendable (PortForwardingRule.ID, UUID, Int) -> Void public typealias FailureHandler = @Sendable (String) -> Void private let worker: SSHForwardingWorker @@ -39,8 +39,8 @@ public actor SSHForwardingSession: SSHForwardingSessionProtocol { try await worker.connect(request, progress: progress) } - public func start(_ rule: PortForwardingRule) async throws { - try await worker.start(rule) + public func start(_ rule: PortForwardingRule, generation: UUID = UUID()) async throws { + try await worker.start(rule, generation: generation) } public func stop(_ id: PortForwardingRule.ID) async { @@ -61,6 +61,7 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable { private final class RuleContext: @unchecked Sendable { let rule: PortForwardingRule + let generation: UUID var socket: Int32 = -1 var events: SSHSocketEvents? var remoteListener: OpaquePointer? @@ -70,7 +71,10 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable { var stop: (@Sendable () -> Void)? var reportedCount = -1 - init(rule: PortForwardingRule) { self.rule = rule } + init(rule: PortForwardingRule, generation: UUID) { + self.rule = rule + self.generation = generation + } } private final class Flow: @unchecked Sendable { @@ -103,6 +107,7 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable { } private let queue = DispatchQueue(label: "com.akinokaede.termind.ssh.forwarding") + private let closingSockets = DispatchGroup() private let connectionHandler: SSHForwardingSession.ConnectionHandler private let failureHandler: SSHForwardingSession.FailureHandler private var socket: Int32 = -1 @@ -140,7 +145,7 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable { deinit { cleanup() } func connect(_ request: SSHConnectionRequest, progress: SSHRouteProgressHandler?) async throws { - await queue.perform { self.cleanup() } + await disconnect() let opened = try await SSHSessionWorker.openAuthenticatedSession( using: request, timeout: 20, progress: progress) do { @@ -156,12 +161,12 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable { } try Task.checkCancellation() } catch { - await queue.perform { self.cleanup() } + await disconnect() throw error } } - func start(_ rule: PortForwardingRule) async throws { + func start(_ rule: PortForwardingRule, generation: UUID) async throws { try await withCheckedThrowingContinuation { continuation in queue.async { [self] in do { @@ -170,7 +175,7 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable { continuation.resume() return } - let context = RuleContext(rule: rule) + let context = RuleContext(rule: rule, generation: generation) switch rule.kind { case .local, .dynamic: do { context.socket = try makeListener(host: rule.bindHost, port: rule.bindPort) } catch { @@ -212,7 +217,14 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable { } } - func disconnect() async { await queue.perform { self.cleanup() } } + func disconnect() async { + await withCheckedContinuation { continuation in + queue.async { [self] in + cleanup() + closingSockets.notify(queue: queue) { continuation.resume() } + } + } + } func statistics() async -> SSHForwardingStatistics { await withCheckedContinuation { continuation in @@ -387,15 +399,25 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable { } return true } - if libssh2_session_last_errno(session) == LIBSSH2_ERROR_EAGAIN { + let code = libssh2_session_last_errno(session) + if code == LIBSSH2_ERROR_EAGAIN { blocked(session) return false } listening = nil if rules[context.rule.id] === context { rules[context.rule.id] = nil } - context.start?.resume( - throwing: SSHSessionWorker.sessionError( - session, prefix: String(localized: "Unable to start remote forwarding"))) + let error: TermindSSHError + switch code { + case LIBSSH2_ERROR_SOCKET_NONE, LIBSSH2_ERROR_SOCKET_SEND, LIBSSH2_ERROR_SOCKET_RECV, + LIBSSH2_ERROR_SOCKET_DISCONNECT, LIBSSH2_ERROR_SOCKET_TIMEOUT, LIBSSH2_ERROR_BAD_SOCKET, + LIBSSH2_ERROR_TIMEOUT: + // A lost transport must not turn a pending listener into a rule-local failure. + error = .disconnected + default: + error = SSHSessionWorker.sessionError( + session, prefix: String(localized: "Unable to start remote forwarding")) + } + context.start?.resume(throwing: error) context.start = nil context.stop?() context.stop = nil @@ -571,7 +593,7 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable { private func report(_ context: RuleContext) { guard !context.stopped, context.start == nil, context.reportedCount != context.flows.count else { return } context.reportedCount = context.flows.count - connectionHandler(context.rule.id, context.reportedCount) + connectionHandler(context.rule.id, context.generation, context.reportedCount) } private func close(_ flow: Flow) { @@ -587,7 +609,11 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable { let descriptor = flow.stream.socket flow.stream.socket = -1 if let events = flow.events { - events.cancel { Darwin.close(descriptor) } + closingSockets.enter() + events.cancel { [closingSockets] in + Darwin.close(descriptor) + closingSockets.leave() + } flow.events = nil } else if descriptor >= 0 { Darwin.close(descriptor) @@ -613,9 +639,11 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable { let descriptor = context.socket context.socket = -1 context.events = nil - events.cancel { + closingSockets.enter() + events.cancel { [closingSockets] in Darwin.close(descriptor) completion() + closingSockets.leave() } } else { completion() @@ -685,7 +713,15 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable { ready = SSHReadyQueue() pendingOpens = SSHReadyQueue() if let transport = routeTransport { - if let events = transportEvents { events.cancel { transport.close() } } else { transport.close() } + if let events = transportEvents { + closingSockets.enter() + events.cancel { [closingSockets] in + transport.close() + closingSockets.leave() + } + } else { + transport.close() + } } transportEvents = nil routeTransport = nil @@ -782,7 +818,10 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable { if candidate >= 0 { var yes: Int32 = 1 setsockopt(candidate, SOL_SOCKET, SO_NOSIGPIPE, &yes, socklen_t(MemoryLayout.size)) - if Darwin.bind(candidate, info.ai_addr, info.ai_addrlen) == 0, + // Closed streams can remain in TIME_WAIT after the listener is released. + // REUSEADDR permits rebinding then without sharing an active listener. + if setsockopt(candidate, SOL_SOCKET, SO_REUSEADDR, &yes, socklen_t(MemoryLayout.size)) == 0, + Darwin.bind(candidate, info.ai_addr, info.ai_addrlen) == 0, Darwin.listen(candidate, 128) == 0 { makeNonblocking(candidate) diff --git a/Memory/ARCHITECTURE.md b/Memory/ARCHITECTURE.md index 1f03dd54..e67b605e 100644 --- a/Memory/ARCHITECTURE.md +++ b/Memory/ARCHITECTURE.md @@ -197,13 +197,23 @@ window-owned multiple/marquee selection with Stop (disabled without active rules Delete. Start/retry always target one rule; neither platform has bulk start or Start All/Stop All. Mobile pushes detail with trailing Start/Stop/Retry. -Only explicit actions start/retry rules: launch, foreground, saved/synced edits and failures never do; -legacy enabled defaults are unused. Background expiration/termination stop resources; navigation owns -neither transport nor prompts. Shared host sessions share authentication; new connections require fresh -challenges. Stopping one rule preserves peers; stopping the last cancels opening/prompts. Listener failure -is rule-local; transport failure fails all without restart. Host/jump-route edits and rule replacement -stop affected runs; authentication Continue/Save may continue its requested opening. Connection/rule -generations reject stale results, and stop/start serializes listener cleanup. +Only explicit actions create rule-running intent; launch, foreground and saved/synced edits never do. +After a listener has run, transport loss preserves that intent and silently recovers all active rules on +one shared Host connection. Retry delays are 1/2/5/10/30 seconds, capped at 30, reset after 60 stable seconds. +Recovery reuses an in-memory authenticated configuration with durable trust refreshed after initial login; +once-only trust and keyboard-interactive answers never authorize another connection. Like SFTP workers, +the complete route forbids app authentication/trust forms, permits system key authorization and does not +gate eligibility on saved credentials. Authentication/trust/interaction failures end recovery with an inline +error; only explicit Retry opens Journey again. Recovery shows Reconnecting and Stop, never Journey or alerts. +Initial failures still need manual Retry. Listener failure is rule-local; old TCP streams are not replayed. + +Legacy enabled defaults are unused. Background expiration/termination stop resources and retries; navigation +owns neither transport nor prompts. Stopping one rule preserves peers; stopping the last cancels opening, +prompts and retries and releases credential snapshots. Host/jump-route edits and rule replacement stop affected +runs; authentication Continue/Save may continue its requested opening. Connection/rule generations reject stale +results, including pending listener completions, and replacement sessions await old transport/listener cleanup. +Disconnect returns only after socket cancellation handlers close descriptors. Local/SOCKS listeners use +SO_REUSEADDR to permit rebinding after closed streams enter TIME_WAIT, never SO_REUSEPORT or shared live binds. Forwarding/jump bridges use DispatchSource readiness and exactly one serial queue per libssh2 session. Stream pumps retain write chunks across EAGAIN, use a 256 KiB per-direction quantum and 256 KiB/session-pass diff --git a/TFI/App/MobileAppDependencies.swift b/TFI/App/MobileAppDependencies.swift index e499c4d8..2c353073 100644 --- a/TFI/App/MobileAppDependencies.swift +++ b/TFI/App/MobileAppDependencies.swift @@ -1,6 +1,7 @@ import Foundation import SwiftData import TermindApplication +import TermindRemoteInfrastructure @MainActor struct MobileAppDependencies { @@ -14,6 +15,7 @@ struct MobileAppDependencies { var container: ModelContainer? = nil var workspaceStore: MobileWorkspaceStore? = nil var liveActivities: MobileLiveActivityController? = nil + var makeForwardings: (HostLibraryModel) -> PortForwardingController = { PortForwardingController(hosts: $0) } static func live() -> MobileAppDependencies { MobileAppDependencies( diff --git a/TFI/App/MobileAppModel.swift b/TFI/App/MobileAppModel.swift index 15870851..a4d864f5 100644 --- a/TFI/App/MobileAppModel.swift +++ b/TFI/App/MobileAppModel.swift @@ -189,7 +189,7 @@ final class MobileAppModel { snippets = SnippetLibraryModel(library: library) knownHosts = KnownHostsLibraryModel(library: library) forwardingLibrary = ForwardingLibraryModel(library: library) - forwardings = PortForwardingController(hosts: library) + forwardings = dependencies.makeForwardings(library) transfers = SFTPTransferCoordinator( store: library, clientFactory: { SFTPClient() }, globalLimit: sftpSettings.preferences.globalLimit, perHostLimit: sftpSettings.preferences.perHostLimit) diff --git a/TFITests/MobileForwardingLifecycleTests.swift b/TFITests/MobileForwardingLifecycleTests.swift index 60c97614..f79d5f9a 100644 --- a/TFITests/MobileForwardingLifecycleTests.swift +++ b/TFITests/MobileForwardingLifecycleTests.swift @@ -1,12 +1,60 @@ import Foundation import TermindApplication import TermindDomain +import TermindRemoteInfrastructure import Testing @testable import TermindIOS @Suite @MainActor struct MobileForwardingLifecycleTests { + @Test func backgroundExpirationStopsEstablishedRecoveryAndForegroundCannotReviveIt() async throws { + let background = ForwardingBackgroundTask() + var dependencies = MobileAppDependencies.testing(background: background) + var sessions: [MobileRecoverySession] = [] + dependencies.makeForwardings = { library in + PortForwardingController( + hosts: library, + makeSession: { _, failure in + let session = MobileRecoverySession(failure: failure) + sessions.append(session) + return session + }, + makeReconnectController: { + SessionReconnectController(sleep: { _ in try await Task.sleep(for: .seconds(3600)) }) + }) + } + let app = try MobileAppModel(dependencies: dependencies) + let identity = Identity(name: "User", username: "user") + let host = TermindDomain.Host(name: "Server", host: "server.invalid", identityID: identity.id) + app.library.upsert(identity) + app.library.upsert(host) + let rule = PortForwardingRule(hostID: host.id) + try app.forwardingLibrary.save(rule) + let scene = UUID() + defer { + app.forwardings.stopAll() + app.removeScene(scene) + app.outbox.stop() + } + app.sceneChanged(scene, phase: .active) + app.forwardings.start(rule.id) + try await wait { app.forwardings.state(for: rule.id) == .running(activeConnections: 0) } + let context = try #require(app.forwardings.connection(for: rule.id)) + await sessions[0].fail() + try await wait { app.forwardings.state(for: rule.id) == .reconnecting } + #expect(!context.journey.isPresented) + app.sceneChanged(scene, phase: .background) + let expire = try #require(background.expiration) + expire() + app.sceneChanged(scene, phase: .active) + context.connection.reconnect.retryNow() + #expect(app.forwardings.state(for: rule.id) == .stopped) + #expect(app.forwardings.connection(for: rule.id) == nil) + #expect(context.connection.reconnectState == nil) + #expect(sessions.count == 1) + } + @Test func backgroundExpirationCancelsSharedAuthenticationAndForegroundNeverRestartsIt() async throws { let background = ForwardingBackgroundTask() let app = try MobileAppModel(inMemory: true, background: background) @@ -82,6 +130,16 @@ struct MobileForwardingLifecycleTests { } } +private actor MobileRecoverySession: SSHForwardingSessionProtocol { + let failure: SSHForwardingSession.FailureHandler + init(failure: @escaping SSHForwardingSession.FailureHandler) { self.failure = failure } + func connect(_ request: SSHConnectionRequest, progress: SSHRouteProgressHandler?) {} + func start(_ rule: PortForwardingRule, generation: UUID) {} + func stop(_ id: UUID) {} + func disconnect() {} + func fail() { failure("Connection lost") } +} + @MainActor private final class ForwardingBackgroundTask: MobileBackgroundTaskManaging { var expiration: (@MainActor () -> Void)? func begin(expiration: @escaping @MainActor () -> Void) { self.expiration = expiration } diff --git a/TFMTests/ArchitectureBoundaryTests.swift b/TFMTests/ArchitectureBoundaryTests.swift index 2474304f..df9554b6 100644 --- a/TFMTests/ArchitectureBoundaryTests.swift +++ b/TFMTests/ArchitectureBoundaryTests.swift @@ -9,9 +9,12 @@ struct ArchitectureBoundaryTests { contentsOf: sourceRoot.appending( path: "../Frameworks/TermindRemoteInfrastructure/SSH/PortForwardingController.swift"), encoding: .utf8) - for symbol in ["UserDefaults", "SettingsKey", "SessionReconnectController", "ExternalAuthentication"] { + for symbol in ["UserDefaults", "SettingsKey", "ExternalAuthentication"] { #expect(!containsToken(symbol, in: controller)) } + #expect(containsToken("SessionReconnectController", in: controller)) + #expect(controller.contains("interactionPolicy: .forbidden")) + #expect(containsToken("enforcingInteractionPolicy", in: controller)) for path in ["../TFI/App/MobileAppModel.swift", "../TFI/App/MobileRootView.swift"] { let source = try String(contentsOf: sourceRoot.appending(path: path), encoding: .utf8) #expect(!containsToken("SSHKeyboardInteractiveQueue", in: source)) diff --git a/TFMTests/PortForwardingJourneyTests.swift b/TFMTests/PortForwardingJourneyTests.swift index ab551a53..11754393 100644 --- a/TFMTests/PortForwardingJourneyTests.swift +++ b/TFMTests/PortForwardingJourneyTests.swift @@ -42,7 +42,7 @@ struct PortForwardingJourneyTests { #expect(app.forwarding.connection(for: rule.id) == nil) } - @Test func savedRulesNeverStartUntilRequestedAndFailureNeverRetries() async throws { + @Test func savedRulesNeverStartUntilRequestedAndInitialFailureNeverRetries() async throws { let fixture = Fixture(mode: .failure) let rule = fixture.rules[0] #expect(fixture.sessions.isEmpty) @@ -226,7 +226,7 @@ struct PortForwardingJourneyTests { #expect(fixture.sessions.count == 1) } - @Test func listenerFailureDoesNotDisconnectAnExistingRuleAndDisconnectNeverRestarts() async throws { + @Test func listenerFailureDoesNotDisconnectAnExistingRuleAndOnlyItsPeerRecovers() async throws { let fixture = Fixture(mode: .listenerFailure) defer { fixture.controller.stopAll() } let first = fixture.rules[0] @@ -243,12 +243,14 @@ struct PortForwardingJourneyTests { let session = try #require(fixture.sessions.first) #expect(await session.disconnects == 0) await session.fail() - try await wait { - if case .failed = fixture.controller.state(for: first.id) { return true } - return false - } + try await wait { fixture.controller.state(for: first.id) == .reconnecting } fixture.controller.reconcileConfiguration() #expect(fixture.sessions.count == 1) + fixture.controller.connection(for: first.id)?.connection.reconnect.retryNow() + try await wait { fixture.controller.state(for: first.id) == .running(activeConnections: 0) } + #expect(fixture.sessions.count == 2) + #expect(await fixture.sessions[1].starts == [first.id]) + #expect(fixture.controller.state(for: second.id).isFailed) } @Test func rapidStopAndStartCannotLeaveTheNewListenerStopped() async throws { @@ -362,10 +364,10 @@ private actor ForwardingTestSession: SSHForwardingSessionProtocol { try Task.checkCancellation() progress?(.connected(configuration.hostID)) } - func start(_ rule: PortForwardingRule) async throws { + func start(_ rule: PortForwardingRule, generation: UUID) async throws { if mode == .listenerFailure && rule.bindPort == 9000 { throw TermindSSHError.route("Port in use") } if mode == .countDuringStart { - count(rule.id, 3) + count(rule.id, generation, 3) try await Task.sleep(for: .milliseconds(20)) } starts.append(rule.id) diff --git a/TFMTests/PortForwardingReconnectTests.swift b/TFMTests/PortForwardingReconnectTests.swift new file mode 100644 index 00000000..d496304e --- /dev/null +++ b/TFMTests/PortForwardingReconnectTests.swift @@ -0,0 +1,582 @@ +import Foundation +import TermindApplication +import TermindDomain +import TermindRemoteInfrastructure +import Testing + +import struct TermindDomain.Host + +@Suite @MainActor +struct PortForwardingReconnectTests { + @Test func sharedHostRestoresEveryKindAndJoinsNewRulesWithoutPresentingJourney() async throws { + let fixture = ReconnectFixture() + defer { fixture.stop() } + try await fixture.start([0, 1, 2]) + let context = try #require(fixture.controller.connection(for: fixture.rules[0].id)) + let old = fixture.sessions[0] + await old.fail() + try await fixture.waitForRetry() + fixture.controller.start(fixture.rules[3].id) + #expect(fixture.sessions.count == 1) + #expect(fixture.rules.allSatisfy { fixture.controller.state(for: $0.id) == .reconnecting }) + #expect(PortForwardingRuntimeState.reconnecting.isActive) + #expect(!PortForwardingRuntimeState.reconnecting.isFailed) + #expect(PortForwardingRuntimeState.reconnecting.label == String(localized: "Reconnecting")) + #expect(!context.journey.isPresented) + await fixture.gate.release() + try await fixture.waitForRunning([0, 1, 2, 3]) + let replacement = fixture.sessions[1] + #expect(await replacement.starts == Set(fixture.rules.map(\.id))) + #expect(await replacement.requests.first?.interactionPolicy == .forbidden) + #expect(await replacement.hasProgress == false) + #expect(await replacement.requests.first?.configuration.credentials.isEmpty == true) + #expect(context.connection.reconnectState == nil) + #expect(context.authenticationPrompt == nil) + #expect(context.connection.pendingHostKey == nil) + #expect(context.connection.keyboardAuthentication.pending.isEmpty) + #expect(!context.journey.isPresented) + await old.report(fixture.rules[0].id, count: 17) + await old.fail() + await fixture.settle() + #expect(fixture.controller.state(for: fixture.rules[0].id) == .running(activeConnections: 0)) + #expect(fixture.sessions.count == 2) + } + + @Test func networkFailuresUseCappedBackoffUntilTheConnectionRecovers() async throws { + let fixture = ReconnectFixture(modes: [.success] + Array(repeating: .offline, count: 5) + [.success]) + defer { fixture.stop() } + try await fixture.start([0]) + await fixture.sessions[0].fail() + for (index, delay) in [1, 2, 5, 10, 30, 30].enumerated() { + try await fixture.waitForRetry() + let state = try #require(fixture.context?.connection.reconnectState) + #expect(state.attempt == index + 1) + #expect(await fixture.gate.nextDelay == .seconds(delay)) + #expect(!fixture.context!.journey.isPresented) + await fixture.gate.release() + try await eventually { fixture.sessions.count == index + 2 } + } + try await fixture.waitForRunning([0]) + #expect(fixture.context?.connection.reconnectState == nil) + } + + @Test func restartingOneRuleOnTheSameTransportRejectsItsOldCounts() async throws { + let fixture = ReconnectFixture() + defer { fixture.stop() } + try await fixture.start([0, 1]) + let session = fixture.sessions[0] + let id = fixture.rules[0].id + let oldGeneration = try #require(await session.generations[id]) + fixture.controller.stop(id) + fixture.controller.start(id) + try await fixture.waitForRunning([0, 1]) + let newGeneration = try #require(await session.generations[id]) + #expect(newGeneration != oldGeneration) + #expect(fixture.sessions.count == 1) + await session.report(id, generation: newGeneration, count: 2) + try await eventually { fixture.controller.state(for: id) == .running(activeConnections: 2) } + await session.report(id, generation: oldGeneration, count: 17) + await fixture.settle() + #expect(fixture.controller.state(for: id) == .running(activeConnections: 2)) + } + + @Test(arguments: [ReconnectSession.Mode.keyboard, .trust], [false, true]) + fileprivate func interactionOnDestinationOrJumpEndsRecoveryWithoutOpeningForms( + _ mode: ReconnectSession.Mode, _ jump: Bool + ) async throws { + let fixture = ReconnectFixture(modes: [.success, mode, mode], usesJump: jump, trusted: mode != .trust) + defer { fixture.stop() } + try await fixture.start([0]) + let context = try #require(fixture.context) + await fixture.sessions[0].fail() + try await fixture.retry() + try await eventually { fixture.controller.state(for: fixture.rules[0].id).isFailed } + #expect( + fixture.controller.state(for: fixture.rules[0].id) == .failed(SSHInteractionRequired().localizedDescription) + ) + #expect(context.connection.reconnectState == nil) + #expect(context.connection.keyboardAuthentication.pending.isEmpty) + #expect(context.connection.pendingHostKey == nil) + #expect(context.authenticationPrompt == nil) + #expect(!context.journey.isPresented) + fixture.controller.retry(fixture.rules[0].id) + try await eventually { + fixture.context?.connection.keyboardAuthentication.current != nil + || fixture.context?.connection.pendingHostKey != nil + } + #expect(fixture.sessions.count == 3) + #expect(fixture.context !== context) + #expect(fixture.context?.journey.isPresented == true) + #expect(await fixture.sessions[2].requests.first?.interactionPolicy == .allowed) + } + + @Test(arguments: [ + TermindSSHError.authenticationFailed("Rejected"), .missingCredential, .keyPassphraseRequired, + .missingUsername, .untrustedHost, .mismatchedHostKey, .hostKeyChanged(expected: ["old"], received: "new"), + .routeHost(hostID: UUID(), authentication: true, serverChallenge: true, hostKey: false, message: "Jump auth"), + .routeHost(hostID: UUID(), authentication: false, serverChallenge: false, hostKey: true, message: "Jump trust"), + ]) + func authenticationAndTrustErrorsRequireManualRetry(_ error: TermindSSHError) async throws { + let fixture = ReconnectFixture() + defer { fixture.stop() } + try await fixture.start([0]) + fixture.nextError = error + await fixture.sessions[0].fail() + try await fixture.retry() + try await eventually { fixture.controller.state(for: fixture.rules[0].id).isFailed } + #expect(fixture.controller.state(for: fixture.rules[0].id) == .failed(error.localizedDescription)) + #expect(fixture.context?.journey.isPresented == false) + #expect(fixture.context?.authenticationPrompt == nil) + #expect(fixture.context?.connection.reconnectState == nil) + } + + @Test(arguments: [false, true]) + func onlyRememberedHostTrustCanAuthorizeRecovery(_ remember: Bool) async throws { + let fixture = ReconnectFixture(modes: [.trust, .trust], trusted: false) + defer { fixture.stop() } + fixture.controller.start(fixture.rules[0].id) + try await eventually { fixture.context?.connection.pendingHostKey != nil } + let context = try #require(fixture.context) + fixture.controller.acceptHostKey( + for: fixture.rules[0].id, requestID: try #require(context.hostKeyRequestID), + observation: try #require(context.connection.pendingHostKey?.observation), remember: remember) + try await fixture.waitForRunning([0]) + await fixture.sessions[0].fail() + try await fixture.retry() + if remember { + try await fixture.waitForRunning([0]) + } else { + try await eventually { fixture.controller.state(for: fixture.rules[0].id).isFailed } + } + let request = try #require(await fixture.sessions[1].requests.first) + #expect(request.configuration.expectedHostKeys.isEmpty == !remember) + #expect(fixture.library.knownHosts.isEmpty == !remember) + #expect(!context.journey.isPresented) + } + + @Test func temporaryPasswordAndKeyPassphraseSurviveOnlyInTheConnectionSnapshot() async throws { + let fixture = ReconnectFixture() + defer { fixture.stop() } + let material = try SSHKeyMaterial.generate(.ed25519, comment: "recovery", passphrase: "temporary-passphrase") + let key = SSHKeyRef( + label: "Encrypted", algorithm: .ed25519, publicKey: material.publicKey, fingerprint: material.fingerprint, + credentialRef: "recovery-key", storage: .thisDevice, hasPassphrase: true) + try fixture.library.upsert(key, credential: .software(SSHPrivateKeySecret(privateKey: material.privateKey))) + var identity = fixture.identity + identity.keyID = key.id + fixture.library.upsert(identity) + fixture.controller.start(fixture.rules[0].id) + try await eventually { fixture.context?.authenticationPrompt != nil } + let prompt = try #require(fixture.context?.authenticationPrompt) + fixture.controller.submitAuthentication( + for: fixture.rules[0].id, promptID: prompt.id, identityID: identity.id, username: identity.username, + password: "temporary-password", passphrase: "temporary-passphrase", keyID: key.id, save: false) + try await fixture.waitForRunning([0]) + await fixture.sessions.last!.fail() + try await fixture.retry() + try await fixture.waitForRunning([0]) + let request = try #require(await fixture.sessions.last!.requests.first) + #expect(request.interactionPolicy == .forbidden) + #expect(request.configuration.credentials.count == 2) + guard case .privateKey(_, _, let passphrase) = request.configuration.credentials[0], + case .password(let password) = request.configuration.credentials[1] + else { + Issue.record("Expected the authenticated credential snapshot") + return + } + #expect(passphrase == "temporary-passphrase") + #expect(password == "temporary-password") + #expect(fixture.library.identity(identity.id)?.usesPassword == false) + #expect(fixture.context?.journey.plainTextLog.contains("temporary-") == false) + } + + @Test func stoppingOneRuleDuringBackoffPreservesOnlyItsPeer() async throws { + let fixture = ReconnectFixture() + defer { fixture.stop() } + try await fixture.start([0, 1]) + await fixture.sessions[0].fail() + try await fixture.waitForRetry() + fixture.controller.stop(fixture.rules[0].id) + await fixture.gate.release() + try await fixture.waitForRunning([1]) + #expect(await fixture.sessions[1].starts == [fixture.rules[1].id]) + #expect(fixture.controller.state(for: fixture.rules[0].id) == .stopped) + } + + @Test(arguments: ["stop", "remove", "host", "jump", "rule", "all"]) + func retiringIntentRejectsEvenAnUncooperativeRetryTimer(_ action: String) async throws { + let fixture = ReconnectFixture(usesJump: true) + defer { fixture.stop() } + try await fixture.start([0]) + let context = try #require(fixture.context) + await fixture.sessions[0].fail() + try await fixture.waitForRetry() + fixture.retire(action, index: 0) + await fixture.gate.release() + await fixture.settle() + #expect(fixture.sessions.count == 1) + #expect(fixture.controller.state(for: fixture.rules[0].id) == .stopped) + #expect(fixture.context == nil) + #expect(context.connection.reconnectState == nil) + } + + @Test(arguments: ["stop", "remove", "host", "jump", "rule", "all"], [false, true]) + func retiringDuringRecoveryRejectsLateOpeningAndListenerSuccess(_ action: String, _ listening: Bool) async throws { + let fixture = ReconnectFixture(modes: [.success, listening ? .holdLateStart : .holdConnect], usesJump: true) + defer { fixture.stop() } + try await fixture.start([1]) + let context = try #require(fixture.controller.connection(for: fixture.rules[1].id)) + await fixture.sessions[0].fail() + try await fixture.retry() + let session = fixture.sessions[1] + try await eventually { await (listening ? session.hasPendingStart : session.waitingToConnect) } + fixture.retire(action, index: 1) + if listening { await session.releaseStart() } else { await session.releaseConnect() } + try await eventually { await session.disconnects > 0 } + await fixture.settle() + #expect(fixture.controller.state(for: fixture.rules[1].id) == .stopped) + #expect(fixture.controller.connection(for: fixture.rules[1].id) == nil) + #expect(context.connection.reconnectState == nil) + #expect(!context.journey.isPresented) + #expect(await session.listening.isEmpty) + #expect(fixture.sessions.count == 2) + } + + @Test(arguments: [false, true]) + func stopAndStartOnASharedTransportRejectsLateListenerResults(_ fails: Bool) async throws { + let fixture = ReconnectFixture(modes: [.success, .holdLateStart]) + defer { fixture.stop() } + try await fixture.start([0, 1]) + await fixture.sessions[0].fail() + try await fixture.retry() + try await fixture.waitForRunning([0]) + let session = fixture.sessions[1] + try await eventually { await session.hasPendingStart } + let id = fixture.rules[1].id + let oldGeneration = try #require(await session.generations[id]) + fixture.controller.stop(id) + fixture.controller.start(id) + await session.releaseStart(failing: fails) + try await eventually { + guard await session.generations[id] != oldGeneration else { return false } + return await session.hasPendingStart + } + await session.releaseStart() + try await fixture.waitForRunning([0, 1]) + #expect(fixture.sessions.count == 2) + #expect(!fixture.context!.journey.isPresented) + #expect(await session.listening == [fixture.rules[0].id, id]) + } + + @Test func stopAndStartDuringOpeningRejectsItsLateSuccessAndWaitsForCleanup() async throws { + let fixture = ReconnectFixture(modes: [.success, .holdConnect, .success]) + defer { fixture.stop() } + try await fixture.start([0]) + await fixture.sessions[0].fail() + try await fixture.retry() + try await eventually { await fixture.sessions[1].waitingToConnect } + let old = fixture.sessions[1] + fixture.controller.stop(fixture.rules[0].id) + fixture.controller.start(fixture.rules[0].id) + try await eventually { fixture.sessions.count == 3 } + #expect(await fixture.sessions[2].requests.isEmpty) + await old.releaseConnect() + try await fixture.waitForRunning([0]) + #expect(await old.starts.isEmpty) + #expect(await old.disconnects > 0) + #expect(await fixture.sessions[2].requests.first?.interactionPolicy == .allowed) + await old.fail() + await fixture.settle() + #expect(fixture.controller.state(for: fixture.rules[0].id) == .running(activeConnections: 0)) + } + + @Test func recoveryWaitsForTheOldListenerToFinishClosingBeforeOpeningItsReplacement() async throws { + let fixture = ReconnectFixture(modes: [.holdDisconnect, .success]) + defer { fixture.stop() } + try await fixture.start([0]) + let old = fixture.sessions[0] + await old.fail() + try await fixture.retry() + try await eventually { + guard fixture.sessions.count == 2 else { return false } + return await old.waitingToDisconnect + } + #expect(await fixture.sessions[1].requests.isEmpty) + #expect(fixture.controller.state(for: fixture.rules[0].id) == .reconnecting) + await old.releaseDisconnect() + try await fixture.waitForRunning([0]) + #expect(await fixture.sessions[1].starts == [fixture.rules[0].id]) + } + + @Test func pendingListenerCancellationCannotEraseRecoveryAndStopDoesNotWaitForItsStart() async throws { + let fixture = ReconnectFixture(modes: [.holdRemoteStart, .holdRemoteStart]) + defer { fixture.stop() } + try await fixture.start([0]) + fixture.controller.start(fixture.rules[1].id) + try await eventually { await fixture.sessions[0].hasPendingStart } + await fixture.sessions[0].fail() + try await fixture.retry() + try await fixture.waitForRunning([0]) + try await eventually { await fixture.sessions[1].hasPendingStart } + #expect(fixture.controller.state(for: fixture.rules[1].id) == .reconnecting) + fixture.controller.stop(fixture.rules[1].id) + try await eventually { await !fixture.sessions[1].hasPendingStart } + #expect(fixture.controller.state(for: fixture.rules[0].id) == .running(activeConnections: 0)) + #expect(fixture.controller.state(for: fixture.rules[1].id) == .stopped) + } + + @Test func listenerFailureDuringRecoveryIsLocalAndStaysOutOfJourney() async throws { + let fixture = ReconnectFixture(modes: [.success, .remoteListenerFailure]) + defer { fixture.stop() } + try await fixture.start([0, 1]) + await fixture.sessions[0].fail() + try await fixture.retry() + try await fixture.waitForRunning([0]) + try await eventually { fixture.controller.state(for: fixture.rules[1].id).isFailed } + #expect(!fixture.context!.journey.isPresented) + await fixture.sessions[1].fail() + try await fixture.retry() + try await fixture.waitForRunning([0]) + #expect(await fixture.sessions[2].starts == [fixture.rules[0].id]) + } + + @Test func transportLossWhileStartingAListenerKeepsTheRuleInRecovery() async throws { + let fixture = ReconnectFixture(modes: [.success, .disconnectedListener, .success]) + defer { fixture.stop() } + try await fixture.start([1]) + await fixture.sessions[0].fail() + try await fixture.retry() + try await fixture.waitForRetry() + #expect(fixture.controller.state(for: fixture.rules[1].id) == .reconnecting) + #expect(await fixture.gate.nextDelay == .seconds(2)) + await fixture.gate.release() + try await fixture.waitForRunning([1]) + #expect(await fixture.sessions[2].starts == [fixture.rules[1].id]) + #expect(fixture.controller.connection(for: fixture.rules[1].id)?.journey.isPresented == false) + } +} + +@MainActor +private final class ReconnectFixture { + let identity = Identity(name: "User", username: "user") + let jump: Host + let host: Host + let rules: [PortForwardingRule] + let library: HostLibraryModel + let gate = ForwardingRetryGate() + var sessions: [ReconnectSession] = [] + var nextError: TermindSSHError? + let modes: [ReconnectSession.Mode] + lazy var controller = makeController() + + private func makeController() -> PortForwardingController { + let makeSession: PortForwardingController.SessionFactory = { [unowned self] count, failure in + let mode = sessions.count < modes.count ? modes[sessions.count] : .success + let session = ReconnectSession(mode: mode, error: nextError, count: count, failure: failure) + nextError = nil + sessions.append(session) + return session + } + return PortForwardingController( + hosts: library, makeSession: makeSession, + makeReconnectController: { [gate] in + SessionReconnectController(policy: .sessionDefault, sleep: { await gate.sleep($0) }, now: Date.init) + }) + } + var context: PortForwardingConnection? { controller.connection(for: rules[0].id) } + + init(modes: [ReconnectSession.Mode] = [], usesJump: Bool = false, trusted: Bool = true) { + self.modes = modes + jump = Host(name: "Jump", host: "jump.invalid", identityID: identity.id) + host = Host(name: "Host", host: "host.invalid", identityID: identity.id, jumpHostID: usesJump ? jump.id : nil) + rules = [ + PortForwardingRule(hostID: host.id, kind: .local, bindPort: 8000), + PortForwardingRule(hostID: host.id, kind: .remote, bindPort: 9000), + PortForwardingRule(hostID: host.id, kind: .dynamic, bindPort: 1080), + PortForwardingRule(hostID: host.id, kind: .local, bindPort: 8001), + ] + library = HostLibraryModel( + hosts: usesJump ? [jump, host] : [host], identities: [identity], + portForwardings: rules, + knownHosts: trusted + ? [jump, host].map { + KnownHostEntry(hostname: $0.host, port: 22, algorithm: .ed25519, fingerprint: "trusted") + } : []) + library.willDeletePortForwardings = { [weak self] in self?.controller.stop($0) } + } + + func start(_ indices: [Int]) async throws { + for index in indices { controller.start(rules[index].id) } + try await waitForRunning(indices) + } + func waitForRunning(_ indices: [Int]) async throws { + try await eventually { + indices.allSatisfy { self.controller.state(for: self.rules[$0].id) == .running(activeConnections: 0) } + } + } + func waitForRetry() async throws { + try await eventually { await self.gate.nextDelay != nil } + } + func retry() async throws { + try await waitForRetry() + let count = sessions.count + await gate.release() + try await eventually { self.sessions.count == count + 1 } + } + func settle() async { for _ in 0..<30 { await Task.yield() } } + func retire(_ action: String, index: Int) { + switch action { + case "stop": controller.stop(rules[index].id) + case "remove": controller.remove([rules[index].id]) + case "host": controller.hostConfigurationChanged(host.id) + case "jump": controller.hostConfigurationChanged(jump.id) + case "rule": + var rule = rules[index] + rule.bindPort += 10 + library.upsert(rule) + default: controller.stopAll() + } + } + func stop() { + controller.stopAll() + Task { await gate.finish() } + } +} + +private actor ForwardingRetryGate { + private var pending: [(Duration, CheckedContinuation)] = [] + private var finished = false + var nextDelay: Duration? { pending.first?.0 } + func sleep(_ duration: Duration) async { + guard !finished else { return } + await withCheckedContinuation { pending.append((duration, $0)) } + } + func release() { if !pending.isEmpty { pending.removeFirst().1.resume() } } + func finish() { + finished = true + while !pending.isEmpty { release() } + } +} + +private actor ReconnectSession: SSHForwardingSessionProtocol { + enum Mode: Sendable { + case success, offline, keyboard, trust, holdConnect, holdDisconnect, holdRemoteStart, holdLateStart + case remoteListenerFailure, disconnectedListener + } + let mode: Mode + let error: TermindSSHError? + let count: SSHForwardingSession.ConnectionHandler + let failure: SSHForwardingSession.FailureHandler + var requests: [SSHConnectionRequest] = [] + var starts: Set = [] + var listening: Set = [] + var generations: [UUID: UUID] = [:] + var hasProgress = false + var disconnects = 0 + private var connectGate: CheckedContinuation? + private var disconnectGate: CheckedContinuation? + private var pendingStarts: [UUID: CheckedContinuation] = [:] + var waitingToConnect: Bool { connectGate != nil } + var waitingToDisconnect: Bool { disconnectGate != nil } + var hasPendingStart: Bool { !pendingStarts.isEmpty } + + init( + mode: Mode, error: TermindSSHError?, count: @escaping SSHForwardingSession.ConnectionHandler, + failure: @escaping SSHForwardingSession.FailureHandler + ) { + self.mode = mode + self.error = error + self.count = count + self.failure = failure + } + func connect(_ request: SSHConnectionRequest, progress: SSHRouteProgressHandler?) async throws { + requests.append(request) + hasProgress = progress != nil + if let error { throw error } + if mode == .offline { throw TermindSSHError.socket("Offline") } + if mode == .holdConnect { await withCheckedContinuation { connectGate = $0 } } + let configuration = request.configuration + let hop = configuration.jumps.first + if mode == .keyboard { + _ = try await request.keyboardInteractive?( + SSHKeyboardInteractiveChallenge( + connectionID: UUID(), hostID: hop?.id ?? configuration.hostID, + hostname: hop?.host ?? configuration.host, port: 22, username: "user", round: 1, + name: "OTP", instruction: "", prompts: [.init(text: "Code", echo: false)])) + } + if mode == .trust { + let observation = SSHHostKeyObservation( + hostname: hop?.host ?? configuration.host, port: 22, algorithm: .ed25519, fingerprint: "trusted") + let keys = hop?.expectedHostKeys ?? configuration.expectedHostKeys + if !keys.contains(observation.identity) { + _ = try await request.authorizeHostKey?( + SSHHostKeyChallenge( + hostID: hop?.id ?? configuration.hostID, observation: observation, expectedHostKeys: keys)) + } + } + progress?(.connected(configuration.hostID)) + } + func start(_ rule: PortForwardingRule, generation: UUID) async throws { + generations[rule.id] = generation + if rule.kind == .remote { + if mode == .remoteListenerFailure { throw TermindSSHError.route("Port in use") } + if mode == .disconnectedListener { throw TermindSSHError.disconnected } + if mode == .holdRemoteStart || mode == .holdLateStart { + try await withCheckedThrowingContinuation { pendingStarts[rule.id] = $0 } + } + } + starts.insert(rule.id) + listening.insert(rule.id) + } + func stop(_ id: UUID) { + listening.remove(id) + if mode != .holdLateStart { pendingStarts.removeValue(forKey: id)?.resume(throwing: CancellationError()) } + } + func disconnect() async { + disconnects += 1 + if mode == .holdDisconnect { await withCheckedContinuation { disconnectGate = $0 } } + listening.removeAll() + if mode != .holdLateStart { cancelStarts() } + } + func releaseDisconnect() { + disconnectGate?.resume() + disconnectGate = nil + } + func releaseConnect() { + connectGate?.resume() + connectGate = nil + } + func releaseStart(failing: Bool = false) { + let pending = pendingStarts.values + pendingStarts.removeAll() + for continuation in pending { + if failing { + continuation.resume(throwing: TermindSSHError.route("Late listen failure")) + } else { + continuation.resume() + } + } + } + func fail() { + cancelStarts() + failure("Connection lost") + } + func report(_ id: UUID, generation: UUID? = nil, count value: Int) { + guard let generation = generation ?? generations[id] else { return } + count(id, generation, value) + } + private func cancelStarts() { + let pending = pendingStarts.values + pendingStarts.removeAll() + for continuation in pending { continuation.resume(throwing: CancellationError()) } + } +} + +@MainActor +private func eventually(_ predicate: () async -> Bool) async throws { + for _ in 0..<2000 { + if await predicate() { return } + try await Task.sleep(for: .milliseconds(1)) + } + throw CocoaError(.coderInvalidValue) +} diff --git a/TFMTests/SSHForwardingWireTests.swift b/TFMTests/SSHForwardingWireTests.swift index 3398aea1..083d58bf 100644 --- a/TFMTests/SSHForwardingWireTests.swift +++ b/TFMTests/SSHForwardingWireTests.swift @@ -9,6 +9,55 @@ import Testing /// Uses the loopback Paramiko or forwarding-only OpenSSH fixture in Scripts/. @Suite(.enabled(if: ProcessInfo.processInfo.environment["TERMIND_SSH_FIXTURE"] != nil), .serialized) nonisolated struct SSHForwardingWireTests { + @Test(arguments: [PortForwardingRule.Kind.local, .dynamic]) + func disconnectReleasesThePortBeforeReturningEvenWithEstablishedStreams(_ kind: PortForwardingRule.Kind) + async throws + { + let fixture = try fixture() + let rule = try rule(kind, fixture: fixture) + for _ in 0..<5 { + let count = ForwardingWireConnectionCount() + let session = SSHForwardingSession( + connectionHandler: { _, _, value in count.set(value) }, failureHandler: { _ in }) + try await session.connect(request(fixture, jumps: 0)) + do { + try await session.start(rule) + var duplicate = rule + duplicate.id = UUID() + await #expect(throws: TermindSSHError.self) { try await session.start(duplicate) } + let client = try SSHSocketConnector.open(host: "127.0.0.1", port: rule.bindPort, timeout: 5) + defer { Darwin.close(client) } + for _ in 0..<1000 where count.value == 0 { try await Task.sleep(for: .milliseconds(1)) } + try #require(count.value == 1) + // Close the server side first so its accepted stream can enter TIME_WAIT. + await session.disconnect() + _ = Darwin.shutdown(client, SHUT_RDWR) + + let candidate = Darwin.socket(AF_INET, SOCK_STREAM, 0) + try #require(candidate >= 0) + defer { Darwin.close(candidate) } + var yes: Int32 = 1 + try #require( + setsockopt(candidate, SOL_SOCKET, SO_REUSEADDR, &yes, socklen_t(MemoryLayout.size)) == 0) + var address = sockaddr_in() + address.sin_len = UInt8(MemoryLayout.size) + address.sin_family = sa_family_t(AF_INET) + address.sin_addr.s_addr = inet_addr("127.0.0.1") + address.sin_port = UInt16(rule.bindPort).bigEndian + let bound = withUnsafePointer(to: &address) { + $0.withMemoryRebound(to: sockaddr.self, capacity: 1) { + Darwin.bind(candidate, $0, socklen_t(MemoryLayout.size)) + } + } + try #require(bound == 0, "The previous forwarding listener must have released its port") + try #require(Darwin.listen(candidate, 1) == 0) + } catch { + await session.disconnect() + throw error + } + } + } + @Test(arguments: [PortForwardingRule.Kind.local, .remote, .dynamic], [1, 16, 64]) func concurrentStreamsKeepTheirPayloadsAndHalfClose(_ kind: PortForwardingRule.Kind, _ concurrency: Int) async throws @@ -24,7 +73,7 @@ nonisolated struct SSHForwardingWireTests { @Test func stoppingOneRulePreservesItsPeerAndReleasesItsListener() async throws { let fixture = try fixture() let session = SSHForwardingSession( - connectionHandler: { _, _ in }, failureHandler: { Issue.record(Comment(rawValue: $0)) }) + connectionHandler: { _, _, _ in }, failureHandler: { Issue.record(Comment(rawValue: $0)) }) try await session.connect(request(fixture, jumps: 0)) do { let first = try rule(.local, fixture: fixture) @@ -58,7 +107,7 @@ nonisolated struct SSHForwardingWireTests { @Test func slowConsumerPreservesPayloadWhileItsPeerProgresses() async throws { let fixture = try fixture() - let session = SSHForwardingSession(connectionHandler: { _, _ in }, failureHandler: { _ in }) + let session = SSHForwardingSession(connectionHandler: { _, _, _ in }, failureHandler: { _ in }) try await session.connect(request(fixture, jumps: 2)) do { let rule = try rule(.local, fixture: fixture) @@ -78,7 +127,7 @@ nonisolated struct SSHForwardingWireTests { @Test func disconnectUnderBackpressureReleasesTheListenerAcrossRepeatedStarts() async throws { let fixture = try fixture() - let session = SSHForwardingSession(connectionHandler: { _, _ in }, failureHandler: { _ in }) + let session = SSHForwardingSession(connectionHandler: { _, _, _ in }, failureHandler: { _ in }) let rule = try rule(.local, fixture: fixture) for _ in 0..<5 { try await session.connect(request(fixture, jumps: 1)) @@ -113,7 +162,7 @@ nonisolated struct SSHForwardingWireTests { let fixture = try fixture() // Only Paramiko provides the deliberately delayed listener reply. guard fixture.server != "openssh" else { return } - let session = SSHForwardingSession(connectionHandler: { _, _ in }, failureHandler: { _ in }) + let session = SSHForwardingSession(connectionHandler: { _, _, _ in }, failureHandler: { _ in }) try await session.connect(request(fixture, jumps: 0, username: "none-slow-forward")) do { let rule = try rule(.remote, fixture: fixture) @@ -135,7 +184,7 @@ nonisolated struct SSHForwardingWireTests { @Test func multipleRemoteListenersCanStopAndRestartIndependently() async throws { let fixture = try fixture() - let session = SSHForwardingSession(connectionHandler: { _, _ in }, failureHandler: { _ in }) + let session = SSHForwardingSession(connectionHandler: { _, _, _ in }, failureHandler: { _ in }) try await session.connect(request(fixture, jumps: 1)) do { let first = try rule(.remote, fixture: fixture) @@ -158,7 +207,7 @@ nonisolated struct SSHForwardingWireTests { @Test(.enabled(if: ProcessInfo.processInfo.environment["TERMIND_FORWARDING_BENCHMARK"] == "1")) func idleForwardingCPU() async throws { let fixture = try fixture() - let session = SSHForwardingSession(connectionHandler: { _, _ in }, failureHandler: { _ in }) + let session = SSHForwardingSession(connectionHandler: { _, _, _ in }, failureHandler: { _ in }) try await session.connect(request(fixture, jumps: 2)) let rule = try rule(.local, fixture: fixture) try await session.start(rule) @@ -182,7 +231,7 @@ nonisolated struct SSHForwardingWireTests { ) async throws { let fixture = try fixture() let session = SSHForwardingSession( - connectionHandler: { _, _ in }, failureHandler: { Issue.record(Comment(rawValue: $0)) }) + connectionHandler: { _, _, _ in }, failureHandler: { Issue.record(Comment(rawValue: $0)) }) try await session.connect(request(fixture, jumps: jumps)) do { let rule = try rule(kind, fixture: fixture) @@ -421,3 +470,10 @@ nonisolated struct SSHForwardingWireTests { return data } } + +private nonisolated final class ForwardingWireConnectionCount: @unchecked Sendable { + private let lock = NSLock() + private var count = 0 + var value: Int { lock.withLock { count } } + func set(_ value: Int) { lock.withLock { count = value } } +} diff --git a/TFMTests/SSHKeyboardInteractiveTests.swift b/TFMTests/SSHKeyboardInteractiveTests.swift index 4dc012c8..2a587ccb 100644 --- a/TFMTests/SSHKeyboardInteractiveTests.swift +++ b/TFMTests/SSHKeyboardInteractiveTests.swift @@ -69,7 +69,7 @@ struct SSHKeyboardInteractiveWireTests { @Test func forwardingAuthenticatesAndReconnectionRequestsFreshAnswers() async throws { let fixture = try fixture() let recorder = WireChallengeRecorder() - let session = SSHForwardingSession(connectionHandler: { _, _ in }, failureHandler: { _ in }) + let session = SSHForwardingSession(connectionHandler: { _, _, _ in }, failureHandler: { _ in }) let request = request(mode: "kbd", fixture: fixture, handler: recorder.handler) try await session.connect(request) await session.disconnect()