From 8794bd6387d60130cfe5e233a5bebb69b9fff81a Mon Sep 17 00:00:00 2001 From: Kaede Akino Date: Sat, 19 Sep 2026 23:13:22 +0800 Subject: [PATCH] Fall back terminal uploads to SSH session --- .../SSH/SSHSessionActor.swift | 726 ++++++++++++++++-- .../Transfers/SFTPTransferSession.swift | 49 +- Memory/ARCHITECTURE.md | 10 +- TFI/Features/Workspaces/MobileWorkspace.swift | 3 +- TFM/Features/SFTP/SFTPSessionModel.swift | 4 +- TFM/Features/Terminal/SSHSessionModel.swift | 12 + .../Terminal/TerminalPlaceholderView.swift | 50 +- TFMTests/SFTPTransferSessionTests.swift | 147 +++- TFMTests/SSHKeyboardInteractiveTests.swift | 18 + 9 files changed, 935 insertions(+), 84 deletions(-) diff --git a/Frameworks/TermindRemoteInfrastructure/SSH/SSHSessionActor.swift b/Frameworks/TermindRemoteInfrastructure/SSH/SSHSessionActor.swift index af2bacf1..fe6ca9e8 100644 --- a/Frameworks/TermindRemoteInfrastructure/SSH/SSHSessionActor.swift +++ b/Frameworks/TermindRemoteInfrastructure/SSH/SSHSessionActor.swift @@ -1,4 +1,5 @@ import CSSH2 +import CryptoKit import Darwin import Foundation import Logging @@ -560,8 +561,7 @@ public actor SSHSessionActor { public typealias ExitHandler = @Sendable (Result) -> Void public typealias ShellIntegrationHandler = @Sendable (SSHShellIntegrationEvent) -> Void - private let adapter: SSHSingleChannelAdapter - private var worker: SSHSessionWorker { adapter.channel } + private nonisolated let worker: SSHSessionWorker private let keyboardInteractive: SSHKeyboardInteractiveHandler? public init( @@ -571,9 +571,8 @@ public actor SSHSessionActor { logger: Logging.Logger = ApplicationLogger.disabled("termind.ssh-session") ) { self.keyboardInteractive = keyboardInteractive - let worker = SSHSessionWorker( + worker = SSHSessionWorker( output: output, exit: exit, shellIntegrationHandler: shellIntegration, logger: logger) - adapter = SSHSingleChannelAdapter(channel: worker) { await $0.disconnect() } } /// Appends a named public key to a remote account's `authorized_keys`, unless @@ -638,6 +637,13 @@ public actor SSHSessionActor { worker.write(data) } + /// Opens an SFTP subsystem on the already authenticated terminal transport. + /// This is a fallback only: callers must first attempt their ordinary + /// independent SFTP connection. + public func openSFTPChannel() async throws -> any SFTPClientProtocol { + try await worker.openSFTPChannel() + } + public func resize(columns: UInt16, rows: UInt16, widthPixels: UInt32, heightPixels: UInt32) { worker.resize( columns: columns, rows: rows, @@ -645,7 +651,7 @@ public actor SSHSessionActor { ) } - public func disconnect() async { await adapter.disconnect() } + public func disconnect() async { await worker.disconnect() } } public nonisolated final class SSHSessionWorker: @unchecked Sendable { @@ -672,6 +678,9 @@ public nonisolated final class SSHSessionWorker: @unchecked Sendable { private var shellIntegration: RemoteShellIntegrationBootstrap? private var commandEventFilter: RemoteCommandEventFilter? private var didReportExit = false + private var sftpChannels: [UUID: OpaquePointer] = [:] + private var sftpOpening = false + private var sftpOpenWaiters: [CheckedContinuation] = [] public init( output: @escaping SSHSessionActor.OutputHandler, @@ -982,7 +991,7 @@ public nonisolated final class SSHSessionWorker: @unchecked Sendable { } } - public func disconnect() async { await queue.perform { self.cleanup() } } + public func disconnect() async { await queue.perform { self.closeTerminalChannel() } } public static func authenticate( username: String, credentials: [SSHCredentials], session: OpaquePointer, @@ -1192,7 +1201,7 @@ public nonisolated final class SSHSessionWorker: @unchecked Sendable { } private func pump() { - guard let session, let channel, let bufferedTransport else { return } + guard let session, let bufferedTransport else { return } let deadline = DispatchTime.now().uptimeNanoseconds + SSHForwardingTuning.current.sliceNanoseconds bufferedTransport.beginPass(deadline: deadline) do { try bufferedTransport.flush(socket: socket) } catch { @@ -1204,57 +1213,57 @@ public nonisolated final class SSHSessionWorker: @unchecked Sendable { socketEvents?.update(read: false, write: true) return } - let resizeResult = ptyResize.flush { size in - libssh2_channel_request_pty_size_ex( - channel, Int32(size.columns), Int32(size.rows), - Int32(clamping: size.widthPixels), Int32(clamping: size.heightPixels) - ) - } - if resizeResult != 0, resizeResult != LIBSSH2_ERROR_EAGAIN { - var metadata: Logger.Metadata = [ - "error_code": "library", "protocol": "ssh", "stage": "pty_resize", - ] - if let hostID { metadata["host_id"] = RemoteLogging.hostID(hostID) } - logger.error( - "ssh.pty_resize.failed", - metadata: metadata) - ptyResize.discardFailedRequest() - } - // Continue other I/O on EAGAIN: window-change may itself be waiting - // for a previously blocked channel write or window adjustment. - var buffer = [CChar](repeating: 0, count: 32 * 1024) - while DispatchTime.now().uptimeNanoseconds < deadline { - let count = libssh2_channel_read_ex(channel, 0, &buffer, buffer.count) - if count > 0 { - deliverOutput(Data(bytes: buffer, count: count)) - } else if count == LIBSSH2_ERROR_EAGAIN || count == 0 { - break - } else { - reportExit(.failure(Self.sessionError(session, prefix: "SSH read failed"))) - cleanup() - return + if let channel { + let resizeResult = ptyResize.flush { size in + libssh2_channel_request_pty_size_ex( + channel, Int32(size.columns), Int32(size.rows), + Int32(clamping: size.widthPixels), Int32(clamping: size.heightPixels) + ) } - } - - if var integration = shellIntegration { - let update = integration.poll() - shellIntegration = update.isComplete ? nil : integration - applyShellIntegrationUpdate(update) - } - - while !pendingWrites.isEmpty && DispatchTime.now().uptimeNanoseconds < deadline { - let written = pendingWrites.withFirstBytes { bytes -> Int in - guard let address = bytes.baseAddress?.assumingMemoryBound(to: CChar.self) else { return 0 } - return libssh2_channel_write_ex(channel, 0, address, bytes.count) + if resizeResult != 0, resizeResult != LIBSSH2_ERROR_EAGAIN { + var metadata: Logger.Metadata = [ + "error_code": "library", "protocol": "ssh", "stage": "pty_resize", + ] + if let hostID { metadata["host_id"] = RemoteLogging.hostID(hostID) } + logger.error("ssh.pty_resize.failed", metadata: metadata) + ptyResize.discardFailedRequest() } - if written > 0 { - pendingWrites.consume(written) - } else if written == LIBSSH2_ERROR_EAGAIN { - break - } else { - reportExit(.failure(Self.sessionError(session, prefix: "SSH write failed"))) - cleanup() - return + // Continue other I/O on EAGAIN: window-change may itself be waiting + // for a previously blocked channel write or window adjustment. + var buffer = [CChar](repeating: 0, count: 32 * 1024) + while DispatchTime.now().uptimeNanoseconds < deadline { + let count = libssh2_channel_read_ex(channel, 0, &buffer, buffer.count) + if count > 0 { + deliverOutput(Data(bytes: buffer, count: count)) + } else if count == LIBSSH2_ERROR_EAGAIN || count == 0 { + break + } else { + reportExit(.failure(Self.sessionError(session, prefix: "SSH read failed"))) + cleanup() + return + } + } + + if var integration = shellIntegration { + let update = integration.poll() + shellIntegration = update.isComplete ? nil : integration + applyShellIntegrationUpdate(update) + } + + while !pendingWrites.isEmpty && DispatchTime.now().uptimeNanoseconds < deadline { + let written = pendingWrites.withFirstBytes { bytes -> Int in + guard let address = bytes.baseAddress?.assumingMemoryBound(to: CChar.self) else { return 0 } + return libssh2_channel_write_ex(channel, 0, address, bytes.count) + } + if written > 0 { + pendingWrites.consume(written) + } else if written == LIBSSH2_ERROR_EAGAIN { + break + } else { + reportExit(.failure(Self.sessionError(session, prefix: "SSH write failed"))) + cleanup() + return + } } } @@ -1265,10 +1274,10 @@ public nonisolated final class SSHSessionWorker: @unchecked Sendable { cleanup() return } - if libssh2_channel_eof(channel) == 1 { + if let channel, libssh2_channel_eof(channel) == 1 { let status = libssh2_channel_get_exit_status(channel) reportExit(.success(status)) - cleanup() + closeTerminalChannel() return } do { try bufferedTransport.flush(socket: socket) } catch { @@ -1346,6 +1355,21 @@ public nonisolated final class SSHSessionWorker: @unchecked Sendable { } } + private func closeTerminalChannel() { + pendingWrites.removeAll() + ptyResize.reset() + shellIntegration = nil + commandEventFilter = nil + hostID = nil + if let channel { + _ = libssh2_channel_send_eof(channel) + _ = libssh2_channel_close(channel) + _ = libssh2_channel_free(channel) + self.channel = nil + } + if sftpChannels.isEmpty { cleanup() } + } + private func cleanup() { pumpTimer?.cancel() pumpTimer = nil @@ -1362,6 +1386,8 @@ public nonisolated final class SSHSessionWorker: @unchecked Sendable { _ = libssh2_channel_free(channel) self.channel = nil } + for sftp in sftpChannels.values { _ = libssh2_sftp_shutdown(sftp) } + sftpChannels.removeAll() if let session { if let bufferedTransport { bufferedTransport.dispose(session: session) @@ -1508,6 +1534,590 @@ public nonisolated final class SSHSessionWorker: @unchecked Sendable { } +private nonisolated enum TerminalSFTPAttempt: Sendable { + case value(Value) + case again +} + +private nonisolated final class TerminalSFTPHandle: @unchecked Sendable { + let pointer: OpaquePointer + init(_ pointer: OpaquePointer) { self.pointer = pointer } +} + +private nonisolated final class TerminalSFTPWriteContext: @unchecked Sendable { + let sftp: OpaquePointer + let handle: OpaquePointer + let temporary: String + let destination: String + let overwrite: Bool + var handleOpen = true + + init( + sftp: OpaquePointer, handle: OpaquePointer, temporary: String, + destination: String, overwrite: Bool + ) { + self.sftp = sftp + self.handle = handle + self.temporary = temporary + self.destination = destination + self.overwrite = overwrite + } +} + +nonisolated extension SSHSessionWorker { + fileprivate func openSFTPChannel() async throws -> any SFTPClientProtocol { + await reserveSFTPOpen() + defer { releaseSFTPOpen() } + let id = UUID() + while true { + try Task.checkCancellation() + let attempt: TerminalSFTPAttempt = try await queue.perform { [self] in + guard let session, let bufferedTransport else { throw SFTPError.notConnected } + let deadline = DispatchTime.now().uptimeNanoseconds + SSHForwardingTuning.current.sliceNanoseconds + bufferedTransport.beginPass(deadline: deadline) + try bufferedTransport.flush(socket: socket) + if bufferedTransport.isBackpressured { return .again } + guard let sftp = libssh2_sftp_init(session) else { + if libssh2_session_last_errno(session) == LIBSSH2_ERROR_EAGAIN { return .again } + throw Self.sessionError(session, prefix: "Unable to start SFTP") + } + sftpChannels[id] = sftp + try bufferedTransport.flush(socket: socket) + updateSharedTransportEvents(session: session, transport: bufferedTransport) + return .value(()) + } + if case .value = attempt { + do { + try Task.checkCancellation() + return TerminalSessionSFTPClient(worker: self, id: id) + } catch { + await closeSFTPChannel(id) + throw error + } + } + try await Task.sleep(for: .milliseconds(2)) + } + } + + private func reserveSFTPOpen() async { + await withCheckedContinuation { continuation in + queue.async { [self] in + if sftpOpening { + sftpOpenWaiters.append(continuation) + } else { + sftpOpening = true + continuation.resume() + } + } + } + } + + private func releaseSFTPOpen() { + queue.async { [self] in + if sftpOpenWaiters.isEmpty { + sftpOpening = false + } else { + sftpOpenWaiters.removeFirst().resume() + } + } + } + + fileprivate func closeSFTPChannel(_ id: UUID) async { + while true { + let attempt: TerminalSFTPAttempt = + (try? await queue.perform { [self] in + guard let sftp = sftpChannels[id] else { return .value(()) } + guard let session, let bufferedTransport else { + sftpChannels[id] = nil + return .value(()) + } + let deadline = DispatchTime.now().uptimeNanoseconds + SSHForwardingTuning.current.sliceNanoseconds + bufferedTransport.beginPass(deadline: deadline) + try bufferedTransport.flush(socket: socket) + let result = libssh2_sftp_shutdown(sftp) + if result == LIBSSH2_ERROR_EAGAIN { return .again } + sftpChannels[id] = nil + try bufferedTransport.flush(socket: socket) + updateSharedTransportEvents(session: session, transport: bufferedTransport) + if channel == nil && sftpChannels.isEmpty { cleanup() } + return .value(()) + }) ?? .value(()) + if case .value = attempt { return } + try? await Task.sleep(for: .milliseconds(2)) + } + } + + private func performSFTP( + _ id: UUID, + _ operation: @escaping @Sendable (OpaquePointer, OpaquePointer) throws -> TerminalSFTPAttempt + ) async throws -> Value { + while true { + try Task.checkCancellation() + let attempt = try await queue.perform { [self] in + guard let session, let sftp = sftpChannels[id], let bufferedTransport else { + throw SFTPError.notConnected + } + let deadline = DispatchTime.now().uptimeNanoseconds + SSHForwardingTuning.current.sliceNanoseconds + bufferedTransport.beginPass(deadline: deadline) + try bufferedTransport.flush(socket: socket) + if bufferedTransport.isBackpressured { return TerminalSFTPAttempt.again } + let result = try operation(session, sftp) + try bufferedTransport.flush(socket: socket) + updateSharedTransportEvents(session: session, transport: bufferedTransport) + return result + } + switch attempt { + case .value(let value): return value + case .again: try await Task.sleep(for: .milliseconds(2)) + } + } + } + + private func updateSharedTransportEvents(session: OpaquePointer, transport: SSHBufferedTransport) { + let directions = libssh2_session_block_directions(session) + socketEvents?.update( + read: !transport.isBackpressured, + write: transport.wantsWrite || directions & LIBSSH2_SESSION_BLOCK_OUTBOUND != 0) + } + + fileprivate func terminalSFTPHomeDirectory(_ id: UUID) async throws -> String { + try await performSFTP(id) { session, sftp in + let source = "." + var target = [CChar](repeating: 0, count: 4 * 1024) + let count = source.withCString { + libssh2_sftp_symlink_ex( + sftp, $0, UInt32(source.utf8.count), &target, UInt32(target.count), + Int32(LIBSSH2_SFTP_REALPATH)) + } + if count == LIBSSH2_ERROR_EAGAIN { return .again } + guard count > 0 else { + throw Self.sessionError(session, prefix: "Unable to resolve the remote home directory") + } + let bytes = target.prefix(Int(count)).map { UInt8(bitPattern: $0) } + return .value(RemotePath.normalized(String(decoding: bytes, as: UTF8.self))) + } + } + + fileprivate func terminalSFTPAttributes(_ id: UUID, path: String) async throws -> RemoteFile? { + try await performSFTP(id) { session, sftp in + var attributes = LIBSSH2_SFTP_ATTRIBUTES() + let result = path.withCString { + libssh2_sftp_stat_ex(sftp, $0, UInt32(path.utf8.count), Int32(LIBSSH2_SFTP_LSTAT), &attributes) + } + if result == LIBSSH2_ERROR_EAGAIN { return .again } + if result == 0 { return .value(Self.terminalRemoteFile(path: path, attributes: attributes)) } + if libssh2_sftp_last_error(sftp) == LIBSSH2_FX_NO_SUCH_FILE { return .value(nil) } + throw Self.sessionError(session, prefix: "Unable to inspect \(path)") + } + } + + fileprivate func terminalSFTPListDirectory(_ id: UUID, path: String) async throws -> [RemoteFile] { + let handle: TerminalSFTPHandle = try await performSFTP(id) { session, sftp in + let opened = path.withCString { + libssh2_sftp_open_ex( + sftp, $0, UInt32(path.utf8.count), 0, 0, Int32(LIBSSH2_SFTP_OPENDIR)) + } + if let opened { return .value(TerminalSFTPHandle(opened)) } + if libssh2_session_last_errno(session) == LIBSSH2_ERROR_EAGAIN { return .again } + throw Self.sessionError(session, prefix: "Unable to open \(path)") + } + var files: [RemoteFile] = [] + do { + while true { + let next: RemoteFile? = try await performSFTP(id) { session, _ in + var name = [CChar](repeating: 0, count: 4 * 1024) + var attributes = LIBSSH2_SFTP_ATTRIBUTES() + let count = libssh2_sftp_readdir_ex( + handle.pointer, &name, name.count, nil, 0, &attributes) + if count == LIBSSH2_ERROR_EAGAIN { return .again } + guard count >= 0 else { throw Self.sessionError(session, prefix: "Unable to read \(path)") } + guard count > 0 else { return .value(nil) } + let filename = String( + decoding: name.prefix(Int(count)).map { UInt8(bitPattern: $0) }, as: UTF8.self) + return .value( + Self.terminalRemoteFile( + path: RemotePath.join(path, filename), name: filename, attributes: attributes)) + } + guard let next else { + break + } + if next.name == "." || next.name == ".." { continue } + files.append(next) + } + } catch { + try? await terminalSFTPCloseHandle(id, handle: handle.pointer) + throw error + } + try await terminalSFTPCloseHandle(id, handle: handle.pointer) + return files.sorted { + if $0.isDirectory != $1.isDirectory { return $0.isDirectory } + return $0.name.localizedStandardCompare($1.name) == .orderedAscending + } + } + + fileprivate func terminalSFTPCreateDirectory(_ id: UUID, path: String) async throws { + try await performStatus(id, prefix: "Unable to create \(path)") { sftp in + path.withCString { libssh2_sftp_mkdir_ex(sftp, $0, UInt32(path.utf8.count), 0o755) } + } + } + + fileprivate func terminalSFTPRemoveFile(_ id: UUID, path: String) async throws { + try await performStatus(id, prefix: "Unable to delete \(path)") { sftp in + path.withCString { libssh2_sftp_unlink_ex(sftp, $0, UInt32(path.utf8.count)) } + } + } + + fileprivate func terminalSFTPRemoveDirectory(_ id: UUID, path: String) async throws { + try await performStatus(id, prefix: "Unable to delete \(path)") { sftp in + path.withCString { libssh2_sftp_rmdir_ex(sftp, $0, UInt32(path.utf8.count)) } + } + } + + fileprivate func terminalSFTPSetPermissions(_ id: UUID, path: String, permissions: UInt32) async throws { + try await performStatus(id, prefix: "Unable to change permissions for \(path)") { sftp in + var attributes = LIBSSH2_SFTP_ATTRIBUTES() + attributes.flags = UInt(LIBSSH2_SFTP_ATTR_PERMISSIONS) + attributes.permissions = UInt(permissions & 0o7777) + return path.withCString { + libssh2_sftp_stat_ex(sftp, $0, UInt32(path.utf8.count), Int32(LIBSSH2_SFTP_SETSTAT), &attributes) + } + } + } + + private func performStatus( + _ id: UUID, prefix: String, + _ operation: @escaping @Sendable (OpaquePointer) -> Int32 + ) async throws { + let _: Void = try await performSFTP(id) { session, sftp in + let result = operation(sftp) + if result == LIBSSH2_ERROR_EAGAIN { return .again } + guard result == 0 else { throw Self.sessionError(session, prefix: prefix) } + return .value(()) + } + } + + fileprivate func terminalSFTPRename( + _ id: UUID, source: String, destination: String, overwrite: Bool + ) async throws { + let _: Void = try await performSFTP(id) { session, sftp in + let result = source.withCString { sourceBuffer in + destination.withCString { destinationBuffer in + SFTPRenameOperation.perform( + overwrite: overwrite, + posixRename: { + libssh2_sftp_posix_rename_ex( + sftp, sourceBuffer, source.utf8.count, + destinationBuffer, destination.utf8.count) + }, + standardRename: { + libssh2_sftp_rename_ex( + sftp, sourceBuffer, UInt32(source.utf8.count), + destinationBuffer, UInt32(destination.utf8.count), 0) + }, + lastStatus: { libssh2_sftp_last_error(sftp) }) + } + } + if result == LIBSSH2_ERROR_EAGAIN { return .again } + guard result == 0 else { throw Self.sessionError(session, prefix: "Unable to rename \(source)") } + return .value(()) + } + } + + private func terminalSFTPCloseHandle(_ id: UUID, handle: OpaquePointer) async throws { + let handle = TerminalSFTPHandle(handle) + let _: Void = try await performSFTP(id) { session, _ in + let result = libssh2_sftp_close_handle(handle.pointer) + if result == LIBSSH2_ERROR_EAGAIN { return .again } + guard result == 0 else { throw Self.sessionError(session, prefix: "Unable to close SFTP handle") } + return .value(()) + } + } + + fileprivate func terminalSFTPUpload( + _ id: UUID, localURL: URL, remotePath: String, overwrite: Bool, + cancellation: TransferCancellationToken, resume: SFTPResumeState, + progress: @escaping @Sendable (SFTPProgress) -> Void + ) async throws { + guard resume.snapshot.editPolicy == nil else { throw SFTPEditFailure.unsupported } + if resume.snapshot.committed { return } + let values = try localURL.resourceValues( + forKeys: [.fileSizeKey, .contentModificationDateKey, .isRegularFileKey]) + guard values.isRegularFile == true, let size = values.fileSize else { throw SFTPResumeFailure.sourceChanged } + let identity = SFTPSourceIdentity(byteCount: Int64(size), modifiedAt: values.contentModificationDate) + try resume.validateSource(identity) + let snapshot = resume.snapshot + let file = try FileHandle(forReadingFrom: localURL) + defer { try? file.close() } + if snapshot.offset > 0 { + var remaining = snapshot.offset + var digest = SHA256() + while remaining > 0 { + guard !cancellation.isCancelled else { throw SFTPError.cancelled } + let data = try file.read(upToCount: Int(min(remaining, 64 * 1024))) ?? Data() + guard !data.isEmpty else { throw SFTPResumeFailure.partialChanged } + digest.update(data: data) + remaining -= Int64(data.count) + } + guard digest.finalize() == snapshot.digest.finalize() else { throw SFTPResumeFailure.partialChanged } + } + let temporary = resume.temporaryPath(for: remotePath) + let context = try await terminalSFTPBeginWrite( + id, temporary: temporary, destination: remotePath, overwrite: overwrite, + continuing: snapshot.hasTemporaryFile, snapshot: snapshot) + if !snapshot.hasTemporaryFile { resume.createdTemporaryFile() } + do { + var completed = snapshot.offset + var sampledBytes: Int64 = 0 + var sampledAt = ContinuousClock.now + while true { + guard !cancellation.isCancelled else { throw SFTPError.cancelled } + let data = try file.read(upToCount: 32_700) ?? Data() + if data.isEmpty { break } + guard Int64(data.count) <= identity.byteCount - completed else { throw SFTPResumeFailure.sourceChanged } + var position = 0 + while position < data.count { + guard !cancellation.isCancelled else { throw SFTPError.cancelled } + let start = position + let written: Int = try await performSFTP(id) { session, _ in + let count = data.withUnsafeBytes { + libssh2_sftp_write( + context.handle, + $0.baseAddress!.advanced(by: start).assumingMemoryBound(to: CChar.self), + data.count - start) + } + if count == LIBSSH2_ERROR_EAGAIN { return .again } + guard count > 0 else { + throw Self.sessionError(session, prefix: "Unable to upload \(remotePath)") + } + return .value(count) + } + resume.acknowledge(data.subdata(in: position..<(position + written))) + position += written + completed += Int64(written) + sampledBytes += Int64(written) + let elapsed = sampledAt.duration(to: .now) + if elapsed >= .milliseconds(250) { + let parts = elapsed.components + let seconds = max(0.001, Double(parts.seconds) + Double(parts.attoseconds) / 1e18) + progress( + SFTPProgress( + completedBytes: completed, totalBytes: identity.byteCount, + bytesPerSecond: Int64(Double(sampledBytes) / seconds))) + sampledBytes = 0 + sampledAt = .now + } + } + } + guard completed == identity.byteCount else { throw SFTPResumeFailure.sourceChanged } + let finalValues = try localURL.resourceValues( + forKeys: [.fileSizeKey, .contentModificationDateKey, .isRegularFileKey]) + guard finalValues.isRegularFile == true, finalValues.fileSize == size, + finalValues.contentModificationDate == values.contentModificationDate + else { throw SFTPResumeFailure.sourceChanged } + try await terminalSFTPFinishWrite(id, context: context, resume: resume) + progress(SFTPProgress(completedBytes: completed, totalBytes: identity.byteCount, bytesPerSecond: 0)) + } catch { + if context.handleOpen { try? await terminalSFTPCloseHandle(id, handle: context.handle) } + throw error + } + } + + private func terminalSFTPBeginWrite( + _ id: UUID, temporary: String, destination: String, overwrite: Bool, + continuing: Bool, snapshot: SFTPResumeState.Snapshot + ) async throws -> TerminalSFTPWriteContext { + let flags = UInt( + LIBSSH2_FXF_READ | LIBSSH2_FXF_WRITE + | (continuing ? 0 : (LIBSSH2_FXF_CREAT | LIBSSH2_FXF_EXCL))) + let context: TerminalSFTPWriteContext = try await performSFTP(id) { session, sftp in + let handle = temporary.withCString { + libssh2_sftp_open_ex( + sftp, $0, UInt32(temporary.utf8.count), flags, 0o600, + Int32(LIBSSH2_SFTP_OPENFILE)) + } + if let handle { + return .value( + TerminalSFTPWriteContext( + sftp: sftp, handle: handle, temporary: temporary, + destination: destination, overwrite: overwrite)) + } + if libssh2_session_last_errno(session) == LIBSSH2_ERROR_EAGAIN { return .again } + throw Self.sessionError(session, prefix: "Unable to create \(destination)") + } + if continuing { + let remoteSize: Int64 = try await performSFTP(id) { session, _ in + var attrs = LIBSSH2_SFTP_ATTRIBUTES() + let result = libssh2_sftp_fstat_ex(context.handle, &attrs, 0) + if result == LIBSSH2_ERROR_EAGAIN { return .again } + guard result == 0, attrs.flags & UInt(LIBSSH2_SFTP_ATTR_SIZE) != 0, + attrs.filesize <= UInt64(Int64.max) + else { throw Self.sessionError(session, prefix: "Unable to inspect incomplete file") } + return .value(Int64(attrs.filesize)) + } + guard remoteSize == snapshot.offset else { + try? await terminalSFTPCloseHandle(id, handle: context.handle) + throw SFTPResumeFailure.partialChanged + } + try await terminalSFTPVerifyRemotePrefix(id, handle: context.handle, snapshot: snapshot) + } + return context + } + + private func terminalSFTPVerifyRemotePrefix( + _ id: UUID, handle: OpaquePointer, snapshot: SFTPResumeState.Snapshot + ) async throws { + let handle = TerminalSFTPHandle(handle) + libssh2_sftp_seek64(handle.pointer, 0) + var remaining = snapshot.offset + var digest = SHA256() + while remaining > 0 { + let wanted = Int(min(remaining, 64 * 1024)) + let data: Data = try await performSFTP(id) { session, _ in + var bytes = [UInt8](repeating: 0, count: wanted) + let count = libssh2_sftp_read(handle.pointer, &bytes, bytes.count) + if count == LIBSSH2_ERROR_EAGAIN { return .again } + guard count > 0 else { throw Self.sessionError(session, prefix: "Unable to verify incomplete file") } + return .value(Data(bytes.prefix(count))) + } + digest.update(data: data) + remaining -= Int64(data.count) + } + guard digest.finalize() == snapshot.digest.finalize() else { throw SFTPResumeFailure.partialChanged } + libssh2_sftp_seek64(handle.pointer, UInt64(snapshot.offset)) + } + + private func terminalSFTPFinishWrite( + _ id: UUID, context: TerminalSFTPWriteContext, resume: SFTPResumeState + ) async throws { + let remoteSize: Int64 = try await performSFTP(id) { session, _ in + var attrs = LIBSSH2_SFTP_ATTRIBUTES() + let result = libssh2_sftp_fstat_ex(context.handle, &attrs, 0) + if result == LIBSSH2_ERROR_EAGAIN { return .again } + guard result == 0, attrs.flags & UInt(LIBSSH2_SFTP_ATTR_SIZE) != 0, + attrs.filesize <= UInt64(Int64.max) + else { throw Self.sessionError(session, prefix: "Unable to inspect uploaded file") } + return .value(Int64(attrs.filesize)) + } + guard remoteSize == resume.snapshot.offset else { throw SFTPResumeFailure.partialChanged } + try await terminalSFTPCloseHandle(id, handle: context.handle) + context.handleOpen = false + if let existing = try await terminalSFTPAttributes(id, path: context.destination), + !context.overwrite || existing.isDirectory + { + throw SFTPStreamUploadOperation.Failure.targetConflict + } + resume.willCommit() + try await terminalSFTPRename( + id, source: context.temporary, destination: context.destination, + overwrite: context.overwrite) + resume.didCommit() + } + + private static func terminalRemoteFile( + path: String, name: String? = nil, attributes: LIBSSH2_SFTP_ATTRIBUTES + ) -> RemoteFile { + let permissions = UInt32(attributes.permissions) + let kind: RemoteFile.Kind + switch permissions & 0o170000 { + case 0o040000: kind = .directory + case 0o100000: kind = .file + case 0o120000: kind = .symbolicLink + case 0: kind = .unknown + default: kind = .special + } + return RemoteFile( + path: path, + name: name ?? URL(fileURLWithPath: path).lastPathComponent, + kind: kind, + size: attributes.filesize <= UInt64(Int64.max) ? Int64(attributes.filesize) : 0, + modifiedAt: attributes.flags & UInt(LIBSSH2_SFTP_ATTR_ACMODTIME) != 0 + ? Date(timeIntervalSince1970: Double(attributes.mtime)) : nil, + permissions: permissions, + ownerUID: UInt32(attributes.uid), + ownerGID: UInt32(attributes.gid)) + } +} + +private actor TerminalSessionSFTPClient: SFTPClientProtocol { + private nonisolated let worker: SSHSessionWorker + private let id: UUID + private var disconnected = false + + init(worker: SSHSessionWorker, id: UUID) { + self.worker = worker + self.id = id + } + + deinit { + let worker = worker + let id = id + Task { await worker.closeSFTPChannel(id) } + } + + func openSiblingChannel() async throws -> any SFTPClientProtocol { try await worker.openSFTPChannel() } + func connect(using configuration: SSHConnectionConfiguration, progress: SSHRouteProgressHandler?) async throws {} + func connect( + using request: SSHConnectionRequest, progress: SSHRouteProgressHandler?, + connectionFailure: (@Sendable (String) -> Void)? + ) async throws {} + func disconnect() async { + guard !disconnected else { return } + disconnected = true + await worker.closeSFTPChannel(id) + } + func homeDirectory() async throws -> String { try await worker.terminalSFTPHomeDirectory(id) } + func listDirectory(at path: String) async throws -> [RemoteFile] { + try await worker.terminalSFTPListDirectory(id, path: path) + } + func attributes(at path: String) async throws -> RemoteFile? { + try await worker.terminalSFTPAttributes(id, path: path) + } + func readFile(at path: String, maximumBytes: Int) async throws -> Data { + throw SFTPResumeFailure.unsupported + } + func writeFile(_ data: Data, at path: String, overwrite: Bool) async throws { + throw SFTPResumeFailure.unsupported + } + func setPermissions(at path: String, to permissions: UInt32) async throws { + try await worker.terminalSFTPSetPermissions(id, path: path, permissions: permissions) + } + func createDirectory(at path: String) async throws { + try await worker.terminalSFTPCreateDirectory(id, path: path) + } + func rename(from source: String, to destination: String, overwrite: Bool) async throws { + try await worker.terminalSFTPRename(id, source: source, destination: destination, overwrite: overwrite) + } + func removeFile(at path: String) async throws { try await worker.terminalSFTPRemoveFile(id, path: path) } + func removeDirectory(at path: String) async throws { + try await worker.terminalSFTPRemoveDirectory(id, path: path) + } + func uploadFile( + from localURL: URL, to remotePath: String, overwrite: Bool, + cancellation: TransferCancellationToken, resume: SFTPResumeState, + progress: @escaping @Sendable (SFTPProgress) -> Void + ) async throws { + try await worker.terminalSFTPUpload( + id, localURL: localURL, remotePath: remotePath, overwrite: overwrite, + cancellation: cancellation, resume: resume, progress: progress) + } + func uploadFile( + from localURL: URL, to remotePath: String, overwrite: Bool, + cancellation: TransferCancellationToken, + progress: @escaping @Sendable (SFTPProgress) -> Void + ) async throws { + try await uploadFile( + from: localURL, to: remotePath, overwrite: overwrite, + cancellation: cancellation, resume: SFTPResumeState(), progress: progress) + } + func downloadFile( + from remotePath: String, to localURL: URL, overwrite: Bool, + cancellation: TransferCancellationToken, + progress: @escaping @Sendable (SFTPProgress) -> Void + ) async throws { + throw SFTPResumeFailure.unsupported + } +} + public nonisolated struct SSHOpenedSession: @unchecked Sendable { public var transport: SSHRoutedSocket public var session: OpaquePointer diff --git a/Frameworks/TermindRemoteInfrastructure/Transfers/SFTPTransferSession.swift b/Frameworks/TermindRemoteInfrastructure/Transfers/SFTPTransferSession.swift index 059c7d21..5f416e64 100644 --- a/Frameworks/TermindRemoteInfrastructure/Transfers/SFTPTransferSession.swift +++ b/Frameworks/TermindRemoteInfrastructure/Transfers/SFTPTransferSession.swift @@ -1,6 +1,15 @@ import Foundation import TermindApplication +public nonisolated enum SFTPTransferPrimaryRole: Hashable, Sendable { + /// A browser control Channel that data work should avoid while alternatives exist. + case control + /// An independent preflight connection that can be donated to the first transfer. + case donatedConnection + /// An authenticated terminal transport used only after independent preflight failed. + case terminalFallback +} + /// A browser and its queued transfers share ownership of one authenticated connection. /// The admitted topology decides between sibling Channels and additional connections; /// additional connections are always opened without interaction from this generation's snapshot. @@ -13,7 +22,7 @@ public final class SFTPTransferSession { private var ownerAttached = true private var transfers: Set = [] private var activeSlots = 0 - private let prefersPrimary: Bool + private let primaryRole: SFTPTransferPrimaryRole private var primaryBusy = false private var probing = false private enum Capability { case unknown, connections, channels } @@ -23,14 +32,14 @@ public final class SFTPTransferSession { public init( configuration: SSHConnectionConfiguration, client: any SFTPClientProtocol, - wasOpenedSilently: Bool = false, + primaryRole: SFTPTransferPrimaryRole = .control, clientFactory: @escaping SFTPTransferCoordinator.ClientFactory = { SFTPClient() } ) { self.configuration = configuration self.primary = client self.clientFactory = clientFactory - self.prefersPrimary = wasOpenedSilently - self.capability = wasOpenedSilently ? .connections : .unknown + self.primaryRole = primaryRole + self.capability = primaryRole == .terminalFallback ? .channels : .unknown } deinit { @@ -86,7 +95,10 @@ public final class SFTPTransferSession { ) async throws -> SFTPTransferLease { try checkCancellation(cancellation) while probing { try await waitForPrimary(cancellation: cancellation) } - if mode == .channelsOnly, let primary, !(prefersPrimary && !primaryBusy) { + if shouldLeasePrimary(mode: mode), let lease = leasePrimaryIfAvailable() { + return lease + } + if mode == .channelsOnly, let primary { do { let client = try await primary.openSiblingChannel() try checkCancellation(cancellation) @@ -102,7 +114,7 @@ public final class SFTPTransferSession { } } if mode == .connectionsPreferred, capability != .channels, primary != nil, - !(prefersPrimary && !primaryBusy) + primaryRole != .donatedConnection || primaryBusy { let isProbe = capability == .unknown if isProbe { probing = true } @@ -126,6 +138,9 @@ public final class SFTPTransferSession { } } if mode == .connectionsPreferred, capability == .channels, let primary { + if primaryRole == .terminalFallback, let lease = leasePrimaryIfAvailable() { + return lease + } do { try checkCancellation(cancellation) let client = try await primary.openSiblingChannel() @@ -165,6 +180,28 @@ public final class SFTPTransferSession { } } + private func shouldLeasePrimary(mode: SSHChannelSchedulingMode) -> Bool { + switch primaryRole { + case .control: + false + case .donatedConnection: + !primaryBusy + case .terminalFallback: + mode == .channelsOnly && !primaryBusy + } + } + + private func leasePrimaryIfAvailable() -> SFTPTransferLease? { + guard !primaryBusy, let primary else { return nil } + primaryBusy = true + return SFTPTransferLease(client: primary) { [self] in + primaryBusy = false + wakeWaiters() + await closeIfUnused() + availabilityChanged?() + } + } + private func openSilently(cancellation: TransferCancellationToken?) async throws -> any SFTPClientProtocol { let client = clientFactory() let request = SSHConnectionRequest(configuration: configuration, interactionPolicy: .forbidden) diff --git a/Memory/ARCHITECTURE.md b/Memory/ARCHITECTURE.md index 9014837e..b3f09e46 100644 --- a/Memory/ARCHITECTURE.md +++ b/Memory/ARCHITECTURE.md @@ -105,7 +105,9 @@ AgentKit is a separately versioned remote package; its sibling working tree is n external-window fallback. - Terminal drops/additional transfer connections forbid app password/passphrase/MFA/trust forms throughout the route, but permit system key authorization. Username-only/SSH `none` remain eligible: actual connection success, not saved - credential presence, decides. Cancellation interrupts opening; interaction-required cleanup fails silently. + credential presence, decides. Cancellation interrupts opening. A terminal drop always tries its independent SFTP + preflight first; any non-cancellation connection failure, including interaction-required authentication, falls back to + SFTP Channels on the already authenticated terminal transport without presenting a form. - Username-only Identities try `none`, then advertised methods. Device-local `lastUsedAt` tracks successful saved-secret reads, not authentication. Batch route reads; missing/availability/edit reads do not count. First access per Identity/ process writes directly without Vault journal/refresh; sync strips this field and preserves local values on apply. @@ -167,8 +169,10 @@ AgentKit is a separately versioned remote package; its sibling working tree is n - Generation-scoped `SFTPTransferSession` outlives panes. SFTP data work resolves one topology per admitted file: Channels Only opens sibling SFTP Channels, Connections Preferred silently probes the full route and falls back to siblings, and Smart uses Channels through 64 MB/64,000,000 bytes (including unknown sizes) before preferring connections. A failed silent probe degrades later work - without replaying active operations; cancellation caches no failure. Browser control Channels stay independent, terminal - preflight may donate its successful connection, and durable trust refresh never promotes once-only trust. + without replaying active operations; cancellation caches no failure. Browser control Channels stay independent. Terminal + preflight donates a successful independent connection; only failed preflight uses the terminal's multiplexed PTY/SFTP + libssh2 owner and fixes that generation to Channels. Closing the PTY does not retire transfers holding that transport. + Durable trust refresh never promotes once-only trust. - One process scheduler covers Workspace/terminal/edit transfers and Agent data operations. It defaults to 30 active files/ten per Host.ID (ranges 1–100 and 1–min(50, global)), rotating Hosts then Workspace/Agent-run/batch sources while retaining FIFO within a source. A Channel and an independent connection consume the same logical-file capacity. Cross-Host copy consumes one diff --git a/TFI/Features/Workspaces/MobileWorkspace.swift b/TFI/Features/Workspaces/MobileWorkspace.swift index 2c01b8e4..eaa8cf52 100644 --- a/TFI/Features/Workspaces/MobileWorkspace.swift +++ b/TFI/Features/Workspaces/MobileWorkspace.swift @@ -1068,7 +1068,8 @@ final class MobileWorkspace: Identifiable { } let session: SFTPTransferSession if ownsClient { - session = SFTPTransferSession(configuration: configuration, client: client, wasOpenedSilently: true) + session = SFTPTransferSession( + configuration: configuration, client: client, primaryRole: .donatedConnection) } else { guard let existing = transferSession else { throw SFTPError.notConnected } session = existing diff --git a/TFM/Features/SFTP/SFTPSessionModel.swift b/TFM/Features/SFTP/SFTPSessionModel.swift index 3c24d3b3..3b855114 100644 --- a/TFM/Features/SFTP/SFTPSessionModel.swift +++ b/TFM/Features/SFTP/SFTPSessionModel.swift @@ -178,7 +178,7 @@ final class SFTPSessionModel: Identifiable { initialPath: String = "", client: (any SFTPClientProtocol)? = nil, configuration: SSHConnectionConfiguration? = nil, - wasOpenedSilently: Bool = false, + primaryRole: SFTPTransferPrimaryRole = .control, homeResolution: RemoteHomeResolutionState = .pending, automaticRefreshInterval: Duration = .seconds(15) ) { @@ -193,7 +193,7 @@ final class SFTPSessionModel: Identifiable { self.configuration = configuration if let client, let configuration { transferSession = SFTPTransferSession( - configuration: configuration, client: client, wasOpenedSilently: wasOpenedSilently) + configuration: configuration, client: client, primaryRole: primaryRole) } self.homeResolution = homeResolution self.automaticRefreshInterval = automaticRefreshInterval diff --git a/TFM/Features/Terminal/SSHSessionModel.swift b/TFM/Features/Terminal/SSHSessionModel.swift index 796ec00f..d2c4049c 100644 --- a/TFM/Features/Terminal/SSHSessionModel.swift +++ b/TFM/Features/Terminal/SSHSessionModel.swift @@ -261,6 +261,18 @@ final class SSHSessionModel: Identifiable { var reconnectState: ReconnectState? { connection.reconnectState } + func openTerminalSFTPChannel(for attemptID: UUID?) async throws -> any SFTPClientProtocol { + guard status == .connected, connectionAttemptID == attemptID, let ssh else { + throw SFTPError.notConnected + } + let client = try await ssh.openSFTPChannel() + guard status == .connected, connectionAttemptID == attemptID, self.ssh === ssh else { + await client.disconnect() + throw CancellationError() + } + return client + } + convenience init(hostID: Host.ID, title: String, initialCommand: String? = nil) { self.init( hostID: hostID, title: title, initialCommand: initialCommand, diff --git a/TFM/Features/Terminal/TerminalPlaceholderView.swift b/TFM/Features/Terminal/TerminalPlaceholderView.swift index 055a950a..1d4cd8aa 100644 --- a/TFM/Features/Terminal/TerminalPlaceholderView.swift +++ b/TFM/Features/Terminal/TerminalPlaceholderView.swift @@ -4,6 +4,38 @@ import TermindApplication import TermindDomain import TermindRemoteInfrastructure +struct TerminalUploadConnection { + let client: any SFTPClientProtocol + let primaryRole: SFTPTransferPrimaryRole +} + +enum TerminalUploadConnectionResolver { + @MainActor + static func resolve( + isCurrent: () -> Bool, + connectIndependent: () async throws -> any SFTPClientProtocol, + openTerminalChannel: () async throws -> any SFTPClientProtocol + ) async throws -> TerminalUploadConnection { + do { + return TerminalUploadConnection( + client: try await connectIndependent(), primaryRole: .donatedConnection) + } catch is CancellationError { + throw CancellationError() + } catch let error as SFTPError { + if case .cancelled = error { throw error } + try Task.checkCancellation() + guard isCurrent() else { throw CancellationError() } + return TerminalUploadConnection( + client: try await openTerminalChannel(), primaryRole: .terminalFallback) + } catch { + try Task.checkCancellation() + guard isCurrent() else { throw CancellationError() } + return TerminalUploadConnection( + client: try await openTerminalChannel(), primaryRole: .terminalFallback) + } + } +} + /// A libghostty terminal surface with Termind's connection and suggestion chrome. struct TerminalPlaceholderView: View { let sshSession: SSHSessionModel @@ -197,10 +229,18 @@ struct TerminalPlaceholderView: View { } var connectedClient: (any SFTPClientProtocol)? do { - let client = try await app.connections.connectSFTP( - using: SSHConnectionRequest(configuration: configuration, interactionPolicy: .forbidden), - progress: nil - ) + let resolved = try await TerminalUploadConnectionResolver.resolve( + isCurrent: { sshSession.connectionAttemptID == generation }, + connectIndependent: { + try await app.connections.connectSFTP( + using: SSHConnectionRequest(configuration: configuration, interactionPolicy: .forbidden), + progress: nil + ) + }, + openTerminalChannel: { + try await sshSession.openTerminalSFTPChannel(for: generation) + }) + let client = resolved.client connectedClient = client try Task.checkCancellation() guard sshSession.connectionAttemptID == generation else { throw CancellationError() } @@ -209,7 +249,7 @@ struct TerminalPlaceholderView: View { initialPath: destination, client: client, configuration: configuration, - wasOpenedSilently: true, + primaryRole: resolved.primaryRole, homeResolution: sshSession.homeResolution ) await uploadSession.prepareUpload(files) diff --git a/TFMTests/SFTPTransferSessionTests.swift b/TFMTests/SFTPTransferSessionTests.swift index 84b8208e..4fa68707 100644 --- a/TFMTests/SFTPTransferSessionTests.swift +++ b/TFMTests/SFTPTransferSessionTests.swift @@ -8,16 +8,59 @@ import Testing @Suite @MainActor struct SFTPTransferSessionTests { + @Test func terminalUploadKeepsASuccessfulIndependentConnection() async throws { + let resolution = UploadConnectionResolutionHarness(independent: .success) + let result = try await TerminalUploadConnectionResolver.resolve( + isCurrent: { true }, + connectIndependent: { try await resolution.connectIndependent() }, + openTerminalChannel: { await resolution.openTerminalChannel() }) + #expect(result.primaryRole == .donatedConnection) + #expect(await resolution.fallbacks == 0) + } + + @Test func terminalUploadFallsBackWhenTheIndependentConnectionNeedsInteraction() async throws { + let resolution = UploadConnectionResolutionHarness(independent: .interactionRequired) + let result = try await TerminalUploadConnectionResolver.resolve( + isCurrent: { true }, + connectIndependent: { try await resolution.connectIndependent() }, + openTerminalChannel: { await resolution.openTerminalChannel() }) + #expect(result.primaryRole == .terminalFallback) + #expect(await resolution.fallbacks == 1) + } + + @Test func terminalUploadDoesNotFallBackAfterCancellationOrRetirement() async throws { + let cancelled = UploadConnectionResolutionHarness(independent: .cancelled) + await #expect(throws: CancellationError.self) { + try await TerminalUploadConnectionResolver.resolve( + isCurrent: { true }, + connectIndependent: { try await cancelled.connectIndependent() }, + openTerminalChannel: { await cancelled.openTerminalChannel() }) + } + #expect(await cancelled.fallbacks == 0) + + let retired = UploadConnectionResolutionHarness(independent: .interactionRequired) + await #expect(throws: CancellationError.self) { + try await TerminalUploadConnectionResolver.resolve( + isCurrent: { false }, + connectIndependent: { try await retired.connectIndependent() }, + openTerminalChannel: { await retired.openTerminalChannel() }) + } + #expect(await retired.fallbacks == 0) + } + @Test func closingTheDesktopPaneKeepsItsSerialUploadAlive() async throws { let harness = TransferSessionHarness(behaviors: [.interactive]) let primary = TransferSessionClient(harness: harness, primary: true) let configuration = SSHConnectionConfiguration( hostID: UUID(), host: "fixture", port: 22, username: "user", credentials: [], expectedHostKeys: []) + let store = HostLibraryModel() let coordinator = SFTPTransferCoordinator( - store: HostLibraryModel(), clientFactory: { TransferSessionClient(harness: harness) }, + store: store, clientFactory: { TransferSessionClient(harness: harness) }, strategy: .connectionsPreferred) let browser = SFTPSessionModel( hostID: configuration.hostID, initialPath: "/", client: primary, configuration: configuration) + browser.bindStatus(to: store) + #expect(store.status(of: configuration.hostID) == .connected) let workspace = Workspace(pane: Pane(.sftpSession(browser))) let window = WindowModel(workspaces: [workspace]) let file = FileManager.default.temporaryDirectory.appending(path: UUID().uuidString) @@ -29,10 +72,12 @@ struct SFTPTransferSessionTests { window.closePane(workspace.focusedPane.id) #expect(window.workspaces.isEmpty) + #expect(store.status(of: configuration.hostID) == .connected) #expect(await primary.disconnects == 0) await harness.releaseOne() try await eventually { coordinator.activeCount == 0 } #expect(coordinator.sftpTransfers.first?.state == .completed) + #expect(store.status(of: configuration.hostID) == .disconnected) #expect(await primary.disconnects == 1) } @@ -104,6 +149,55 @@ struct SFTPTransferSessionTests { await session.releaseOwner() } + @Test func smartUsesChannelsOnTheDonatedConnectionThroughTheDecimalThreshold() async throws { + let harness = TransferSessionHarness(behaviors: []) + let primary = TransferSessionClient(harness: harness, primary: true) + let (session, coordinator) = fixture( + primary: primary, harness: harness, limit: 2, + primaryRole: .donatedConnection, strategy: .smart) + enqueue( + [SFTPConcurrencyStrategy.smartConnectionThreshold, SFTPConcurrencyStrategy.smartConnectionThreshold], + session: session, coordinator: coordinator) + try await eventually { await harness.started == 2 } + #expect(await harness.opens == 0) + #expect(await harness.siblingChannels == 1) + await harness.releaseAll() + try await eventually { coordinator.activeCount == 0 } + await session.releaseOwner() + } + + @Test func smartUsesAnotherConnectionAboveTheDecimalThreshold() async throws { + let harness = TransferSessionHarness(behaviors: []) + let primary = TransferSessionClient(harness: harness, primary: true) + let (session, coordinator) = fixture( + primary: primary, harness: harness, limit: 2, + primaryRole: .donatedConnection, strategy: .smart) + let large = SFTPConcurrencyStrategy.smartConnectionThreshold + 1 + enqueue([large, large], session: session, coordinator: coordinator) + try await eventually { await harness.started == 2 } + #expect(await harness.opens == 1) + #expect(await harness.siblingChannels == 0) + await harness.releaseAll() + try await eventually { coordinator.activeCount == 0 } + await session.releaseOwner() + } + + @Test func terminalFallbackNeverRetriesConnectionsForLargeFiles() async throws { + let harness = TransferSessionHarness(behaviors: []) + let primary = TransferSessionClient(harness: harness, primary: true) + let (session, coordinator) = fixture( + primary: primary, harness: harness, limit: 3, + primaryRole: .terminalFallback, strategy: .smart) + let large = SFTPConcurrencyStrategy.smartConnectionThreshold + 1 + enqueue([large, large, large], session: session, coordinator: coordinator) + try await eventually { await harness.started == 3 } + #expect(await harness.opens == 0) + #expect(await harness.siblingChannels == 2) + await harness.releaseAll() + try await eventually { coordinator.activeCount == 0 } + await session.releaseOwner() + } + @Test func emptyCredentialsCanOpenParallelConnectionsWithinTheLimits() async throws { let harness = TransferSessionHarness(behaviors: []) let primary = TransferSessionClient(harness: harness, primary: true) @@ -159,7 +253,8 @@ struct SFTPTransferSessionTests { @Test func preflightConnectionIsUsedWithoutReauthentication() async throws { let harness = TransferSessionHarness(behaviors: [.interactive]) let primary = TransferSessionClient(harness: harness, primary: true) - let (session, coordinator) = fixture(primary: primary, harness: harness, wasOpenedSilently: true) + let (session, coordinator) = fixture( + primary: primary, harness: harness, primaryRole: .donatedConnection) enqueue(1, session: session, coordinator: coordinator) try await eventually { await harness.started == 1 } #expect(await harness.opens == 0) @@ -173,7 +268,8 @@ struct SFTPTransferSessionTests { @Test func pausedTransfersReleaseTheDetachedPrimaryAndResumeSilently() async throws { let harness = TransferSessionHarness(behaviors: []) let primary = TransferSessionClient(harness: harness, primary: true) - let (session, coordinator) = fixture(primary: primary, harness: harness, wasOpenedSilently: true) + let (session, coordinator) = fixture( + primary: primary, harness: harness, primaryRole: .donatedConnection) let id = enqueue(1, session: session, coordinator: coordinator)[0] try await eventually { await harness.started == 1 } await session.releaseOwner() @@ -199,7 +295,7 @@ struct SFTPTransferSessionTests { globalLimit: 15, perHostLimit: 5, strategy: .connectionsPreferred) let upload = SFTPSessionModel( hostID: configuration.hostID, initialPath: "/destination", client: primary, - configuration: configuration, wasOpenedSilently: true) + configuration: configuration, primaryRole: .donatedConnection) let directory = FileManager.default.temporaryDirectory.appending(path: UUID().uuidString) try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) defer { try? FileManager.default.removeItem(at: directory) } @@ -237,7 +333,7 @@ struct SFTPTransferSessionTests { let sessions = (0..<2).map { _ in SFTPTransferSession( configuration: configuration, client: TransferSessionClient(harness: harness), - wasOpenedSilently: true) + primaryRole: .donatedConnection) } for session in sessions { enqueue(4, session: session, coordinator: coordinator) } try await eventually { await harness.started == 5 } @@ -253,14 +349,14 @@ struct SFTPTransferSessionTests { private func fixture( primary: TransferSessionClient, harness: TransferSessionHarness, limit: Int = 4, - wasOpenedSilently: Bool = false, + primaryRole: SFTPTransferPrimaryRole = .control, strategy: SFTPConcurrencyStrategy = .connectionsPreferred ) -> (SFTPTransferSession, SFTPTransferCoordinator) { let session = SFTPTransferSession( configuration: SSHConnectionConfiguration( hostID: UUID(), host: "fixture", port: 22, username: "user", credentials: [], expectedHostKeys: []), - client: primary, wasOpenedSilently: wasOpenedSilently) + client: primary, primaryRole: primaryRole) let coordinator = SFTPTransferCoordinator( store: HostLibraryModel(), clientFactory: { TransferSessionClient(harness: harness) }, globalLimit: limit, perHostLimit: limit, strategy: strategy) @@ -269,11 +365,18 @@ struct SFTPTransferSessionTests { @discardableResult private func enqueue(_ count: Int, session: SFTPTransferSession, coordinator: SFTPTransferCoordinator) -> [UUID] { - (0.. [UUID] { + byteCounts.enumerated().map { index, byteCount in coordinator.enqueue( direction: .upload, batchID: UUID(), localURL: URL(fileURLWithPath: "/tmp/termind-test-\(index)"), remotePath: "/file-\(index)", - byteCount: 1, hostID: session.configuration.hostID, connection: .session(session), + byteCount: byteCount, hostID: session.configuration.hostID, connection: .session(session), scopedRoot: URL(fileURLWithPath: "/tmp"), overwrite: true) } } @@ -287,6 +390,32 @@ struct SFTPTransferSessionTests { } } +private actor UploadConnectionResolutionHarness { + enum Independent { case success, interactionRequired, cancelled } + + let independent: Independent + private(set) var fallbacks = 0 + private let client: TransferSessionClient + + init(independent: Independent) { + self.independent = independent + client = TransferSessionClient(harness: TransferSessionHarness(behaviors: [])) + } + + func connectIndependent() async throws -> any SFTPClientProtocol { + switch independent { + case .success: client + case .interactionRequired: throw SSHInteractionRequired() + case .cancelled: throw CancellationError() + } + } + + func openTerminalChannel() -> any SFTPClientProtocol { + fallbacks += 1 + return client + } +} + private actor TransferSessionHarness { enum Behavior: Sendable { case success, interactive, waitForCancellation } private var behaviors: [Behavior] diff --git a/TFMTests/SSHKeyboardInteractiveTests.swift b/TFMTests/SSHKeyboardInteractiveTests.swift index 2a587ccb..3ca6e088 100644 --- a/TFMTests/SSHKeyboardInteractiveTests.swift +++ b/TFMTests/SSHKeyboardInteractiveTests.swift @@ -66,6 +66,24 @@ struct SSHKeyboardInteractiveWireTests { await session.disconnect() } + @Test func terminalSFTPChannelReusesAuthenticationAndOutlivesThePTY() async throws { + let fixture = try fixture() + let recorder = WireChallengeRecorder() + let session = SSHSessionActor(output: { _ in }, exit: { _ in }) + try await session.connect( + using: request(mode: "kbd", fixture: fixture, handler: recorder.handler), + shellIntegrationEnabled: false, columns: 80, rows: 24 + ) + let client = try await session.openSFTPChannel() + #expect(try await client.listDirectory(at: "/").map(\.name) == ["fixture.txt"]) + #expect(await recorder.challenges.count == 1) + + await session.disconnect() + #expect(try await client.attributes(at: "/fixture.txt")?.size == 8) + #expect(await recorder.challenges.count == 1) + await client.disconnect() + } + @Test func forwardingAuthenticatesAndReconnectionRequestsFreshAnswers() async throws { let fixture = try fixture() let recorder = WireChallengeRecorder()