From 8fdf2eb0d3966b0829748ae58123ddb1972764eb Mon Sep 17 00:00:00 2001 From: Kaede Akino Date: Sun, 20 Sep 2026 00:33:36 +0800 Subject: [PATCH] Unify SFTP channel implementation --- .../LaunchFileLogServiceTests.swift | 5 +- .../Agent/AgentBuiltInSSHTools.swift | 8 +- .../Agent/AgentConnections.swift | 44 +- .../Agent/AgentHostServices.swift | 2 +- .../SSH/RemoteConnectionCoordinator.swift | 6 +- ...aredSFTPClient.swift => SFTPChannel.swift} | 245 ++- .../SSH/SFTPClient.swift | 1558 ++--------------- .../SSH/SSHMultiplexedSFTPChannelOwner.swift | 176 ++ .../SSH/SSHMultiplexedSession.swift | 20 +- .../SSH/SSHMultiplexedSessionWorker.swift | 171 +- .../SSH/SSHSessionActor.swift | 160 +- Memory/ARCHITECTURE.md | 4 + Memory/REFACTOR.md | 16 + TFMTests/AgentConnectionPoolTests.swift | 54 +- TFMTests/AgentHostAuthenticationTests.swift | 22 +- TFMTests/SSHKeyboardInteractiveTests.swift | 9 + 16 files changed, 679 insertions(+), 1821 deletions(-) rename Frameworks/TermindRemoteInfrastructure/SSH/{SSHSharedSFTPClient.swift => SFTPChannel.swift} (80%) create mode 100644 Frameworks/TermindRemoteInfrastructure/SSH/SSHMultiplexedSFTPChannelOwner.swift diff --git a/FrameworkTests/LaunchFileLogServiceTests.swift b/FrameworkTests/LaunchFileLogServiceTests.swift index 730ae00e..76a748c7 100644 --- a/FrameworkTests/LaunchFileLogServiceTests.swift +++ b/FrameworkTests/LaunchFileLogServiceTests.swift @@ -154,7 +154,7 @@ struct LaunchFileLogServiceTests { Tcp: RtoAlgorithm """ let sampler = SSHMetricsSampler( - session: MetricsLogExecSession( + session: MetricsLogSession( result: SSHExecResult( stdout: output, stderr: "", exitCode: 0, stdoutTruncated: false, stderrTruncated: false)), logger: service.makeLogger(label: "termind.metrics") @@ -199,7 +199,7 @@ private final class LogFixture: @unchecked Sendable { } } -private actor MetricsLogExecSession: SSHMultiplexedSessionProtocol { +private actor MetricsLogSession: SSHMultiplexedSessionProtocol { let result: SSHExecResult init(result: SSHExecResult) { self.result = result } func connect(using _: SSHConnectionConfiguration) async throws {} @@ -207,6 +207,7 @@ private actor MetricsLogExecSession: SSHMultiplexedSessionProtocol { _: String, workingDirectory _: String?, stdin _: Data?, timeout _: TimeInterval, outputLimit _: Int, onOutput _: SSHExecOutputHandler? ) async throws -> SSHExecResult { result } + func openSFTPChannel() async throws -> any SFTPClientProtocol { throw SFTPError.notConnected } func disconnect() async {} } diff --git a/Frameworks/TermindRemoteInfrastructure/Agent/AgentBuiltInSSHTools.swift b/Frameworks/TermindRemoteInfrastructure/Agent/AgentBuiltInSSHTools.swift index 316708fe..cf9af11a 100644 --- a/Frameworks/TermindRemoteInfrastructure/Agent/AgentBuiltInSSHTools.swift +++ b/Frameworks/TermindRemoteInfrastructure/Agent/AgentBuiltInSSHTools.swift @@ -404,13 +404,13 @@ nonisolated public struct SSHExecTool: AgentToolDefinition, AgentToolSchemaBuild // ends on time is what lets the user be heard at all. let handover = AgentCommandHandover() let running = Task { () async throws -> SSHExecResult in - var connection = try await authentication.execConnection( + var connection = try await authentication.connection( hostID: hostID, invocationID: invocation.id, interaction: context.sshInteraction, pool: pool ) var retriedDeadSession = false while true { - let session = connection.client + let session = connection.session // The call may already have given up while this was // connecting. Starting anyway would leave a command // running, printing and finishing with nobody holding @@ -431,7 +431,7 @@ nonisolated public struct SSHExecTool: AgentToolDefinition, AgentToolSchemaBuild outputLimit: 256 * 1_024, onOutput: stream.handler) } - } catch SSHExecSessionError.transportUnavailable + } catch SSHCommandExecutionError.transportUnavailable where !retriedDeadSession { // Safe because no command channel opened. The @@ -445,7 +445,7 @@ nonisolated public struct SSHExecTool: AgentToolDefinition, AgentToolSchemaBuild metadata: ["presentation_phase": .string("running")] )) await pool.disconnect(hostID: hostID) - connection.client = try await authentication.openSession( + connection.session = try await authentication.openSession( hostID: hostID, authentication: connection.authentication, pool: pool ) diff --git a/Frameworks/TermindRemoteInfrastructure/Agent/AgentConnections.swift b/Frameworks/TermindRemoteInfrastructure/Agent/AgentConnections.swift index 528f020b..f3b1c736 100644 --- a/Frameworks/TermindRemoteInfrastructure/Agent/AgentConnections.swift +++ b/Frameworks/TermindRemoteInfrastructure/Agent/AgentConnections.swift @@ -698,11 +698,11 @@ public actor AgentSSHConfigurationSession { nonisolated public struct AgentSSHConnection: Sendable { public var authentication: AgentSSHConfigurationSession - public var client: any SSHMultiplexedSessionProtocol + public var session: any SSHMultiplexedSessionProtocol - public init(authentication: AgentSSHConfigurationSession, client: any SSHMultiplexedSessionProtocol) { + public init(authentication: AgentSSHConfigurationSession, session: any SSHMultiplexedSessionProtocol) { self.authentication = authentication - self.client = client + self.session = session } } @@ -722,7 +722,7 @@ public actor AgentSSHAuthenticationRegistry { public init(resolver: AgentHostResolver) { self.resolver = resolver } - public func exec( + public func session( hostID: Host.ID, invocationID: AgentToolInvocation.ID, interaction: AgentSSHInteraction, pool: AgentConnectionPool ) async throws -> any SSHMultiplexedSessionProtocol { @@ -735,18 +735,18 @@ public actor AgentSSHAuthenticationRegistry { } } - public func execConnection( + public func connection( hostID: Host.ID, invocationID: AgentToolInvocation.ID, interaction: AgentSSHInteraction, pool: AgentConnectionPool ) async throws -> AgentSSHConnection { try await withSession( hostID: hostID, invocationID: invocationID, interaction: interaction ) { authentication in - let client = try await self.openSession( + let session = try await self.openSession( hostID: hostID, authentication: authentication, pool: pool ) return AgentSSHConnection( - authentication: authentication, client: client + authentication: authentication, session: session ) } } @@ -759,13 +759,13 @@ public actor AgentSSHAuthenticationRegistry { hostID: hostID, authentication: authentication, pool: pool ) { attempt in if let attempt { - return try await pool.exec(hostID: hostID) { _ in attempt.request } + return try await pool.session(hostID: hostID) { _ in attempt.request } } - return try await pool.exec(hostID: hostID) + return try await pool.session(hostID: hostID) } } - public func sftp( + public func sftpClient( hostID: Host.ID, invocationID: AgentToolInvocation.ID, interaction: AgentSSHInteraction, pool: AgentConnectionPool ) async throws -> any SFTPClientProtocol { @@ -776,9 +776,9 @@ public actor AgentSSHAuthenticationRegistry { hostID: hostID, authentication: authentication, pool: pool ) { attempt in if let attempt { - return try await pool.sftp(hostID: hostID) { _ in attempt.request } + return try await pool.sftpClient(hostID: hostID) { _ in attempt.request } } - return try await pool.sftp(hostID: hostID) + return try await pool.sftpClient(hostID: hostID) } } } @@ -1030,7 +1030,7 @@ public actor AgentConnectionPool { _ = await disconnectSFTPState(hostID: hostID) } - let primary = try await sftp(hostID: hostID, configuration: override) + let primary = try await sftpClient(hostID: hostID, configuration: override) let committedFingerprint = try await routeFingerprint(hostID) if sftpStates[hostID] == nil { let record = SFTPRecord(id: UUID(), client: primary, busy: false, lastUsedAt: .now) @@ -1200,15 +1200,15 @@ public actor AgentConnectionPool { onOutput: SSHExecOutputHandler? = nil ) async throws -> SSHExecResult { do { - let session = try await exec(hostID: hostID, configuration: override) + let session = try await session(hostID: hostID, configuration: override) return try await withExecLease(hostID: hostID, session: session) { try await session.execute( command, workingDirectory: workingDirectory, stdin: stdin, timeout: timeout, outputLimit: outputLimit, onOutput: onOutput) } - } catch SSHExecSessionError.transportUnavailable { + } catch SSHCommandExecutionError.transportUnavailable { await disconnect(hostID: hostID) - let session = try await exec(hostID: hostID, configuration: override) + let session = try await session(hostID: hostID, configuration: override) return try await withExecLease(hostID: hostID, session: session) { try await session.execute( command, workingDirectory: workingDirectory, stdin: stdin, @@ -1222,7 +1222,7 @@ public actor AgentConnectionPool { operation: @escaping @Sendable () async throws -> Value ) async throws -> Value { guard let pooled = entries[hostID]?.session, pooled === session else { - throw SSHExecSessionError.transportUnavailable(TermindSSHError.disconnected) + throw SSHCommandExecutionError.transportUnavailable(TermindSSHError.disconnected) } activeExecLeases[hostID, default: 0] += 1 defer { @@ -1236,23 +1236,25 @@ public actor AgentConnectionPool { return try await operation() } - public func exec( + public func session( hostID: UUID, configuration override: RequestResolving? = nil ) async throws -> any SSHMultiplexedSessionProtocol { try await resolve(hostID: hostID, configuration: override) } - public func sftp( + public func sftpClient( hostID: UUID, configuration override: RequestResolving? = nil ) async throws -> any SFTPClientProtocol { let fingerprint = try await routeFingerprint(hostID) if let state = sftpStates[hostID], state.fingerprint == fingerprint, let primary = state.records[state.primaryID] { + sftpStates[hostID]?.records[state.primaryID]?.lastUsedAt = .now + entries[hostID]?.lastUsedAt = .now return primary.client } if sftpStates[hostID] != nil { await disconnect(hostID: hostID) } - let session = try await exec(hostID: hostID, configuration: override) + let session = try await session(hostID: hostID, configuration: override) let opening: SFTPPrimaryOpen let ownsOpening: Bool if let existing = sftpPrimaryOpens[hostID], existing.fingerprint == fingerprint, @@ -1286,7 +1288,7 @@ public actor AgentConnectionPool { guard committedFingerprint == fingerprint, entries[hostID]?.session === session else { await client.disconnect() await disconnect(hostID: hostID) - return try await sftp(hostID: hostID, configuration: override) + return try await sftpClient(hostID: hostID, configuration: override) } let record = SFTPRecord(id: UUID(), client: client, busy: false, lastUsedAt: .now) sftpStates[hostID] = SFTPState( diff --git a/Frameworks/TermindRemoteInfrastructure/Agent/AgentHostServices.swift b/Frameworks/TermindRemoteInfrastructure/Agent/AgentHostServices.swift index e9dec1a1..bf9bacb9 100644 --- a/Frameworks/TermindRemoteInfrastructure/Agent/AgentHostServices.swift +++ b/Frameworks/TermindRemoteInfrastructure/Agent/AgentHostServices.swift @@ -33,7 +33,7 @@ public struct AgentHostServices { connections: AgentConnectionPool( configuration: { try await resolver.configuration(for: $0) }, routeFingerprint: { try await resolver.routeFingerprint(for: $0) }, - openSession: { try await coordinator.connectMultiplexed(using: $0) }, + openSession: { try await coordinator.connectMultiplexedSession(using: $0) }, openSFTP: { try await coordinator.connectSFTP(using: $0) }, sftpConcurrencyScheduler: sftpConcurrencyScheduler ), diff --git a/Frameworks/TermindRemoteInfrastructure/SSH/RemoteConnectionCoordinator.swift b/Frameworks/TermindRemoteInfrastructure/SSH/RemoteConnectionCoordinator.swift index 8f50671f..2816876a 100644 --- a/Frameworks/TermindRemoteInfrastructure/SSH/RemoteConnectionCoordinator.swift +++ b/Frameworks/TermindRemoteInfrastructure/SSH/RemoteConnectionCoordinator.swift @@ -44,15 +44,15 @@ public nonisolated final class RemoteConnectionCoordinator: Sendable { /// Mirrors `connectSFTP`, including its guarantee that a session which fails /// to come up is torn down here rather than handed back half-open. - public func connectMultiplexed( + public func connectMultiplexedSession( using configuration: SSHConnectionConfiguration ) async throws -> any SSHMultiplexedSessionProtocol { - try await connectMultiplexed( + try await connectMultiplexedSession( using: SSHConnectionRequest(configuration: configuration) ) } - public func connectMultiplexed( + public func connectMultiplexedSession( using request: SSHConnectionRequest ) async throws -> any SSHMultiplexedSessionProtocol { let session = makeMultiplexedSession() diff --git a/Frameworks/TermindRemoteInfrastructure/SSH/SSHSharedSFTPClient.swift b/Frameworks/TermindRemoteInfrastructure/SSH/SFTPChannel.swift similarity index 80% rename from Frameworks/TermindRemoteInfrastructure/SSH/SSHSharedSFTPClient.swift rename to Frameworks/TermindRemoteInfrastructure/SSH/SFTPChannel.swift index 822a7fa9..ea9c49ba 100644 --- a/Frameworks/TermindRemoteInfrastructure/SSH/SSHSharedSFTPClient.swift +++ b/Frameworks/TermindRemoteInfrastructure/SSH/SFTPChannel.swift @@ -6,37 +6,37 @@ import Foundation import TermindApplication import TermindDomain -nonisolated enum SSHSharedSFTPAttempt: Sendable { +nonisolated enum SFTPChannelAttempt: Sendable { case value(Value) case again } -nonisolated final class SSHSharedSFTPValue: @unchecked Sendable { +nonisolated final class SFTPChannelValue: @unchecked Sendable { let value: Any init(_ value: Any) { self.value = value } } -nonisolated enum SSHSharedSFTPAnyAttempt: @unchecked Sendable { - case value(SSHSharedSFTPValue) +nonisolated enum SFTPChannelAnyAttempt: @unchecked Sendable { + case value(SFTPChannelValue) case again } -/// Type-erased access to a non-blocking libssh2 owner. The PTY and Agent -/// workers keep their own pumps and lifetimes; this driver lets one complete -/// SFTP implementation take bounded turns on either owner. -nonisolated final class SSHSharedSFTPDriver: @unchecked Sendable { - typealias RawOperation = @Sendable (OpaquePointer, OpaquePointer) throws -> SSHSharedSFTPAnyAttempt +/// Type-erased access to a non-blocking libssh2 owner. Standalone, PTY and +/// Agent sessions keep their own lifetimes; this driver lets one SFTP +/// implementation take bounded turns on any of those owners. +nonisolated final class SFTPChannelDriver: @unchecked Sendable { + typealias RawOperation = @Sendable (OpaquePointer, OpaquePointer) throws -> SFTPChannelAnyAttempt let identity: UUID private let openAction: @Sendable () async throws -> UUID private let closeAction: @Sendable (UUID) async -> Void - private let performAction: @Sendable (UUID, @escaping RawOperation) async throws -> SSHSharedSFTPValue + private let performAction: @Sendable (UUID, @escaping RawOperation) async throws -> SFTPChannelValue init( identity: UUID, open: @escaping @Sendable () async throws -> UUID, close: @escaping @Sendable (UUID) async -> Void, - perform: @escaping @Sendable (UUID, @escaping RawOperation) async throws -> SSHSharedSFTPValue + perform: @escaping @Sendable (UUID, @escaping RawOperation) async throws -> SFTPChannelValue ) { self.identity = identity openAction = open @@ -49,27 +49,27 @@ nonisolated final class SSHSharedSFTPDriver: @unchecked Sendable { func perform( _ id: UUID, - _ operation: @escaping @Sendable (OpaquePointer, OpaquePointer) throws -> SSHSharedSFTPAttempt + _ operation: @escaping @Sendable (OpaquePointer, OpaquePointer) throws -> SFTPChannelAttempt ) async throws -> Value { let boxed = try await performAction(id) { session, sftp in switch try operation(session, sftp) { - case .value(let value): .value(SSHSharedSFTPValue(value)) + case .value(let value): .value(SFTPChannelValue(value)) case .again: .again } } guard let value = boxed.value as? Value else { - preconditionFailure("Shared SFTP driver returned an unexpected value type") + preconditionFailure("SFTP Channel driver returned an unexpected value type") } return value } } -private nonisolated final class SSHSharedSFTPHandle: @unchecked Sendable { +private nonisolated final class SFTPChannelHandle: @unchecked Sendable { let pointer: OpaquePointer init(_ pointer: OpaquePointer) { self.pointer = pointer } } -private nonisolated final class SSHSharedSFTPWriteContext: @unchecked Sendable { +private nonisolated final class SFTPChannelWriteContext: @unchecked Sendable { let handle: OpaquePointer let temporary: String let destination: String @@ -84,7 +84,7 @@ private nonisolated final class SSHSharedSFTPWriteContext: @unchecked Sendable { } } -private nonisolated struct SSHSharedSFTPProgressMeter { +private nonisolated struct SFTPProgressMeter { let total: Int64 let report: @Sendable (SFTPProgress) -> Void private(set) var completed: Int64 @@ -117,24 +117,24 @@ private nonisolated struct SSHSharedSFTPProgressMeter { } } -/// An SFTP subsystem Channel on a shared non-blocking SSH transport. -actor SSHSharedSFTPClient: SFTPClientProtocol { - private nonisolated let driver: SSHSharedSFTPDriver +/// An opened SFTP subsystem Channel on a multiplexed non-blocking SSH transport. +actor SFTPChannel: SFTPClientProtocol { + private nonisolated let driver: SFTPChannelDriver private let id: UUID private var disconnected = false - static func open(using driver: SSHSharedSFTPDriver) async throws -> any SFTPClientProtocol { + static func open(using driver: SFTPChannelDriver) async throws -> any SFTPClientProtocol { let id = try await driver.open() do { try Task.checkCancellation() - return SSHSharedSFTPClient(driver: driver, id: id) + return SFTPChannel(driver: driver, id: id) } catch { await driver.close(id) throw error } } - private init(driver: SSHSharedSFTPDriver, id: UUID) { + private init(driver: SFTPChannelDriver, id: UUID) { self.driver = driver self.id = id } @@ -259,14 +259,14 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { continuing: false, snapshot: nil) do { try await write(data, to: context.handle, cancellation: nil, acknowledge: nil) - try await closeHandle(SSHSharedSFTPHandle(context.handle)) + try await closeHandle(SFTPChannelHandle(context.handle)) context.handleOpen = false if let existing = try await attributes(at: path), !overwrite || existing.isDirectory { throw SFTPStreamUploadOperation.Failure.targetConflict } try await rename(from: temporary, to: path, overwrite: overwrite) } catch { - if context.handleOpen { try? await closeHandle(SSHSharedSFTPHandle(context.handle)) } + if context.handleOpen { try? await closeHandle(SFTPChannelHandle(context.handle)) } try? await removeFile(at: temporary) throw error } @@ -333,7 +333,6 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { 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]) @@ -343,15 +342,21 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { let identity = SFTPSourceIdentity(byteCount: Int64(size), modifiedAt: values.contentModificationDate) try resume.validateSource(identity) let snapshot = resume.snapshot + if try await recoverCommittedWrite( + destination: remotePath, resume: resume, cancellation: cancellation + ) { + return + } let file = try FileHandle(forReadingFrom: localURL) defer { try? file.close() } try Self.verifyLocalPrefix(file, snapshot: snapshot, cancellation: cancellation) let context = try await beginWrite( temporary: resume.temporaryPath(for: remotePath), destination: remotePath, - overwrite: overwrite, continuing: snapshot.hasTemporaryFile, snapshot: snapshot) + overwrite: overwrite, continuing: snapshot.hasTemporaryFile, + snapshot: snapshot, cancellation: cancellation) if !snapshot.hasTemporaryFile { resume.createdTemporaryFile() } do { - var meter = SSHSharedSFTPProgressMeter( + var meter = SFTPProgressMeter( total: identity.byteCount, offset: snapshot.offset, report: progress) while true { guard !cancellation.isCancelled else { throw SFTPError.cancelled } @@ -370,7 +375,7 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { try await finishWrite(context, resume: resume) meter.finish() } catch { - if context.handleOpen { try? await closeHandle(SSHSharedSFTPHandle(context.handle)) } + if context.handleOpen { try? await closeHandle(SFTPChannelHandle(context.handle)) } throw error } } @@ -416,7 +421,7 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { try Self.verifyLocalPrefix(file, snapshot: snapshot, cancellation: cancellation) try await verifyRemotePrefix(handle, snapshot: snapshot, cancellation: cancellation) try file.seek(toOffset: UInt64(snapshot.offset)) - var meter = SSHSharedSFTPProgressMeter( + var meter = SFTPProgressMeter( total: identity.byteCount, offset: snapshot.offset, report: progress) while let data = try await read(handle, maximumBytes: 32_700) { guard !cancellation.isCancelled else { throw SFTPError.cancelled } @@ -430,6 +435,11 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { guard meter.completed == identity.byteCount, try await remoteIdentity(handle) == identity else { throw SFTPResumeFailure.sourceChanged } + if resume.snapshot.capturesFileVersion { + resume.recordCommittedVersion( + SFTPFileVersion(try await remoteFile(at: remotePath, from: handle))) + } + guard !cancellation.isCancelled else { throw SFTPError.cancelled } try await closeHandle(handle) handleOpen = false try file.synchronize() @@ -466,15 +476,20 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { progress: @escaping @Sendable (SFTPProgress) -> Void, sourceSucceeded: @escaping @Sendable () -> Bool ) async throws { - guard resume.snapshot.editPolicy == nil else { throw SFTPEditFailure.unsupported } if resume.snapshot.committed { return } let snapshot = resume.snapshot + if try await recoverCommittedWrite( + destination: remotePath, resume: resume, cancellation: cancellation + ) { + return + } let context = try await beginWrite( temporary: resume.temporaryPath(for: remotePath), destination: remotePath, - overwrite: overwrite, continuing: snapshot.hasTemporaryFile, snapshot: snapshot) + overwrite: overwrite, continuing: snapshot.hasTemporaryFile, + snapshot: snapshot, cancellation: cancellation) if !snapshot.hasTemporaryFile { resume.createdTemporaryFile() } do { - var meter = SSHSharedSFTPProgressMeter(total: byteCount, offset: snapshot.offset, report: progress) + var meter = SFTPProgressMeter(total: byteCount, offset: snapshot.offset, report: progress) while true { guard !cancellation.isCancelled else { throw SFTPError.cancelled } var buffer = [UInt8](repeating: 0, count: 32_700) @@ -501,7 +516,7 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { try await finishWrite(context, resume: resume) meter.finish() } catch { - if context.handleOpen { try? await closeHandle(SSHSharedSFTPHandle(context.handle)) } + if context.handleOpen { try? await closeHandle(SFTPChannelHandle(context.handle)) } throw error } } @@ -526,7 +541,7 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { do { let identity = try await remoteIdentity(handle) guard identity.byteCount <= maximumBytes else { throw SFTPDownloadFailure.tooLarge } - return SSHSharedSFTPDownloadStream( + return SFTPChannelDownloadStream( client: self, handle: handle, identity: identity, cancellation: cancellation) } catch { try? await closeHandle(handle) @@ -540,7 +555,7 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { cancellation: TransferCancellationToken, progress: @escaping @Sendable (SFTPProgress) -> Void ) async throws { - if let destination = destination as? SSHSharedSFTPClient, + if let destination = destination as? SFTPChannel, destination.driver.identity == driver.identity { try await copyOnSharedTransport( @@ -585,7 +600,7 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { private func copyOnSharedTransport( from sourcePath: String, to destinationPath: String, - destination: SSHSharedSFTPClient, byteCount: Int64, overwrite: Bool, + destination: SFTPChannel, byteCount: Int64, overwrite: Bool, cancellation: TransferCancellationToken, progress: @escaping @Sendable (SFTPProgress) -> Void ) async throws { @@ -603,7 +618,7 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { temporary: temporary, destination: destinationPath, overwrite: overwrite, continuing: false, snapshot: nil) do { - var meter = SSHSharedSFTPProgressMeter(total: byteCount, report: progress) + var meter = SFTPProgressMeter(total: byteCount, report: progress) while let data = try await read(source, maximumBytes: 64 * 1024) { guard !cancellation.isCancelled else { throw SFTPError.cancelled } guard Int64(data.count) <= byteCount - meter.completed else { @@ -618,7 +633,7 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { } try await closeHandle(source) sourceOpen = false - try await destination.closeHandle(SSHSharedSFTPHandle(output.handle)) + try await destination.closeHandle(SFTPChannelHandle(output.handle)) output.handleOpen = false if let existing = try await destination.attributes(at: destinationPath), !overwrite || existing.isDirectory @@ -630,7 +645,7 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { } catch { if sourceOpen { try? await closeHandle(source) } if output.handleOpen { - try? await destination.closeHandle(SSHSharedSFTPHandle(output.handle)) + try? await destination.closeHandle(SFTPChannelHandle(output.handle)) } try? await destination.removeFile(at: temporary) throw error @@ -668,39 +683,39 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { } fileprivate func streamRead( - _ handle: SSHSharedSFTPHandle, cancellation: TransferCancellationToken + _ handle: SFTPChannelHandle, cancellation: TransferCancellationToken ) async throws -> Data? { guard !cancellation.isCancelled else { throw SFTPError.cancelled } return try await read(handle, maximumBytes: 64 * 1024) } - fileprivate func streamSeek(_ handle: SSHSharedSFTPHandle, offset: Int64) async throws { + fileprivate func streamSeek(_ handle: SFTPChannelHandle, offset: Int64) async throws { guard offset >= 0 else { throw SFTPResumeFailure.partialChanged } try await seek(handle, offset: offset) } - fileprivate func streamClose(_ handle: SSHSharedSFTPHandle) async throws { + fileprivate func streamClose(_ handle: SFTPChannelHandle) async throws { try await closeHandle(handle) } - fileprivate func streamIdentity(_ handle: SSHSharedSFTPHandle) async throws -> SFTPSourceIdentity { + fileprivate func streamIdentity(_ handle: SFTPChannelHandle) async throws -> SFTPSourceIdentity { try await remoteIdentity(handle) } private func openHandle( path: String, flags: UInt, mode: Int, type: Int32 - ) async throws -> SSHSharedSFTPHandle { + ) async throws -> SFTPChannelHandle { try await driver.perform(id) { session, sftp in let opened = path.withCString { libssh2_sftp_open_ex(sftp, $0, UInt32(path.utf8.count), flags, mode, type) } - if let opened { return .value(SSHSharedSFTPHandle(opened)) } + if let opened { return .value(SFTPChannelHandle(opened)) } if libssh2_session_last_errno(session) == LIBSSH2_ERROR_EAGAIN { return .again } throw Self.error(session, sftp, "Unable to open \(path)") } } - private func closeHandle(_ handle: SSHSharedSFTPHandle) async throws { + private func closeHandle(_ handle: SFTPChannelHandle) async throws { let _: Void = try await driver.perform(id) { session, sftp in let result = libssh2_sftp_close_handle(handle.pointer) if result == LIBSSH2_ERROR_EAGAIN { return .again } @@ -709,7 +724,7 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { } } - private func read(_ handle: SSHSharedSFTPHandle, maximumBytes: Int) async throws -> Data? { + private func read(_ handle: SFTPChannelHandle, maximumBytes: Int) async throws -> Data? { try await driver.perform(id) { session, sftp in var buffer = [UInt8](repeating: 0, count: maximumBytes) let count = libssh2_sftp_read(handle.pointer, &buffer, buffer.count) @@ -719,7 +734,7 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { } } - private func seek(_ handle: SSHSharedSFTPHandle, offset: Int64) async throws { + private func seek(_ handle: SFTPChannelHandle, offset: Int64) async throws { let _: Void = try await driver.perform(id) { _, _ in libssh2_sftp_seek64(handle.pointer, UInt64(offset)) return .value(()) @@ -730,7 +745,7 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { _ data: Data, to handle: OpaquePointer, cancellation: TransferCancellationToken?, acknowledge: ((Data) -> Void)? ) async throws { - let stableHandle = SSHSharedSFTPHandle(handle) + let stableHandle = SFTPChannelHandle(handle) var position = 0 while position < data.count { if cancellation?.isCancelled == true { throw SFTPError.cancelled } @@ -763,14 +778,50 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { } } + private func recoverCommittedWrite( + destination: String, resume: SFTPResumeState, + cancellation: TransferCancellationToken + ) async throws -> Bool { + let snapshot = resume.snapshot + guard snapshot.hasTemporaryFile, snapshot.committing, + let temporary = snapshot.temporaryPath, + try await attributes(at: temporary) == nil + else { return false } + let handle = try await openHandle( + path: destination, flags: UInt(LIBSSH2_FXF_READ), mode: 0, + type: Int32(LIBSSH2_SFTP_OPENFILE)) + do { + guard try await remoteSize(handle) == snapshot.offset else { + throw SFTPResumeFailure.partialChanged + } + try await verifyRemotePrefix( + handle, snapshot: snapshot, cancellation: cancellation) + resume.recordCommittedVersion( + SFTPFileVersion(try await remoteFile(at: destination, from: handle))) + try await closeHandle(handle) + resume.didCommit() + return true + } catch { + try? await closeHandle(handle) + throw error + } + } + private func beginWrite( temporary: String, destination: String, overwrite: Bool, - continuing: Bool, snapshot: SFTPResumeState.Snapshot? - ) async throws -> SSHSharedSFTPWriteContext { + continuing: Bool, snapshot: SFTPResumeState.Snapshot?, + cancellation: TransferCancellationToken = TransferCancellationToken() + ) async throws -> SFTPChannelWriteContext { + if let editPolicy = snapshot?.editPolicy { + try editPolicy.validate(try await attributes(at: destination)) + } + if continuing, try await attributes(at: temporary) == nil { + throw SFTPResumeFailure.partialChanged + } let flags = UInt( LIBSSH2_FXF_READ | LIBSSH2_FXF_WRITE | (continuing ? 0 : (LIBSSH2_FXF_CREAT | LIBSSH2_FXF_EXCL))) - let context: SSHSharedSFTPWriteContext = try await driver.perform(id) { session, sftp in + let context: SFTPChannelWriteContext = try await driver.perform(id) { session, sftp in let handle = temporary.withCString { libssh2_sftp_open_ex( sftp, $0, UInt32(temporary.utf8.count), flags, 0o600, @@ -778,7 +829,7 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { } if let handle { return .value( - SSHSharedSFTPWriteContext( + SFTPChannelWriteContext( handle: handle, temporary: temporary, destination: destination, overwrite: overwrite)) } @@ -786,23 +837,37 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { throw Self.error(session, sftp, "Unable to create \(destination)") } if continuing, let snapshot { - guard try await remoteSize(SSHSharedSFTPHandle(context.handle)) == snapshot.offset else { - try? await closeHandle(SSHSharedSFTPHandle(context.handle)) + let handle = SFTPChannelHandle(context.handle) + let size = try await remoteIdentity(handle).byteCount + guard size >= snapshot.offset else { + try? await closeHandle(SFTPChannelHandle(context.handle)) throw SFTPResumeFailure.partialChanged } try await verifyRemotePrefix( - SSHSharedSFTPHandle(context.handle), snapshot: snapshot, - cancellation: TransferCancellationToken()) + handle, snapshot: snapshot, cancellation: cancellation) + if size != snapshot.offset { + try await truncate(handle, to: snapshot.offset) + } } return context } - private func finishWrite(_ context: SSHSharedSFTPWriteContext, resume: SFTPResumeState) async throws { - guard try await remoteSize(SSHSharedSFTPHandle(context.handle)) == resume.snapshot.offset else { + private func finishWrite(_ context: SFTPChannelWriteContext, resume: SFTPResumeState) async throws { + let handle = SFTPChannelHandle(context.handle) + let snapshot = resume.snapshot + guard try await remoteSize(handle) == snapshot.offset else { throw SFTPResumeFailure.partialChanged } - try await closeHandle(SSHSharedSFTPHandle(context.handle)) + if let editPolicy = snapshot.editPolicy { + try await setPermissions(on: handle, to: editPolicy.permissions) + let committed = try await remoteFile(at: context.destination, from: handle) + resume.recordCommittedVersion(SFTPFileVersion(committed)) + } + try await closeHandle(handle) context.handleOpen = false + if let editPolicy = snapshot.editPolicy { + try editPolicy.validate(try await attributes(at: context.destination)) + } if let existing = try await attributes(at: context.destination), !context.overwrite || existing.isDirectory { @@ -815,7 +880,7 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { resume.didCommit() } - private func remoteSize(_ handle: SSHSharedSFTPHandle) async throws -> Int64 { + private func remoteSize(_ handle: SFTPChannelHandle) async throws -> Int64 { try await driver.perform(id) { session, sftp in var attributes = LIBSSH2_SFTP_ATTRIBUTES() let result = libssh2_sftp_fstat_ex(handle.pointer, &attributes, 0) @@ -827,7 +892,51 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { } } - private func remoteIdentity(_ handle: SSHSharedSFTPHandle) async throws -> SFTPSourceIdentity { + private func truncate(_ handle: SFTPChannelHandle, to size: Int64) async throws { + let _: Void = try await driver.perform(id) { session, sftp in + var attributes = LIBSSH2_SFTP_ATTRIBUTES() + attributes.flags = UInt(LIBSSH2_SFTP_ATTR_SIZE) + attributes.filesize = UInt64(size) + let result = libssh2_sftp_fstat_ex(handle.pointer, &attributes, 1) + if result == LIBSSH2_ERROR_EAGAIN { return .again } + guard result == 0 else { + throw Self.error(session, sftp, "Unable to trim incomplete file") + } + return .value(()) + } + } + + private func setPermissions(on handle: SFTPChannelHandle, to permissions: UInt32) async throws { + let _: Void = try await driver.perform(id) { session, sftp in + var attributes = LIBSSH2_SFTP_ATTRIBUTES() + attributes.flags = UInt(LIBSSH2_SFTP_ATTR_PERMISSIONS) + attributes.permissions = UInt(permissions & 0o7777) + let result = libssh2_sftp_fstat_ex(handle.pointer, &attributes, 1) + if result == LIBSSH2_ERROR_EAGAIN { return .again } + guard result == 0 else { + throw Self.error( + session, sftp, + String(localized: "Unable to preserve remote file permissions")) + } + return .value(()) + } + } + + private func remoteFile( + at path: String, from handle: SFTPChannelHandle + ) async throws -> RemoteFile { + try await driver.perform(id) { session, sftp in + var attributes = LIBSSH2_SFTP_ATTRIBUTES() + let result = libssh2_sftp_fstat_ex(handle.pointer, &attributes, 0) + if result == LIBSSH2_ERROR_EAGAIN { return .again } + guard result == 0 else { + throw Self.error(session, sftp, "Unable to inspect remote file") + } + return .value(Self.remoteFile(path: path, attributes: attributes)) + } + } + + private func remoteIdentity(_ handle: SFTPChannelHandle) async throws -> SFTPSourceIdentity { try await driver.perform(id) { session, sftp in var attributes = LIBSSH2_SFTP_ATTRIBUTES() let result = libssh2_sftp_fstat_ex(handle.pointer, &attributes, 0) @@ -846,7 +955,7 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { } private func verifyRemotePrefix( - _ handle: SSHSharedSFTPHandle, snapshot: SFTPResumeState.Snapshot, + _ handle: SFTPChannelHandle, snapshot: SFTPResumeState.Snapshot, cancellation: TransferCancellationToken ) async throws { try await seek(handle, offset: 0) @@ -932,9 +1041,9 @@ actor SSHSharedSFTPClient: SFTPClientProtocol { } } -private actor SSHSharedSFTPDownloadStream: SFTPDownloadStream { - private let client: SSHSharedSFTPClient - private let handle: SSHSharedSFTPHandle +private actor SFTPChannelDownloadStream: SFTPDownloadStream { + private let client: SFTPChannel + private let handle: SFTPChannelHandle let sourceIdentity: SFTPSourceIdentity? let byteCount: Int64 private let cancellation: TransferCancellationToken @@ -943,7 +1052,7 @@ private actor SSHSharedSFTPDownloadStream: SFTPDownloadStream { private var eof = false init( - client: SSHSharedSFTPClient, handle: SSHSharedSFTPHandle, + client: SFTPChannel, handle: SFTPChannelHandle, identity: SFTPSourceIdentity, cancellation: TransferCancellationToken ) { self.client = client diff --git a/Frameworks/TermindRemoteInfrastructure/SSH/SFTPClient.swift b/Frameworks/TermindRemoteInfrastructure/SSH/SFTPClient.swift index 4a8f4c8c..5ece1d97 100644 --- a/Frameworks/TermindRemoteInfrastructure/SSH/SFTPClient.swift +++ b/Frameworks/TermindRemoteInfrastructure/SSH/SFTPClient.swift @@ -1,6 +1,4 @@ import AgentKit -import CSSH2 -import CryptoKit import Darwin import Foundation import Logging @@ -285,70 +283,38 @@ public nonisolated struct SFTPTemporaryFileCleanupError: LocalizedError, Sendabl } public actor SFTPClient: SFTPClientProtocol { - public func openSiblingChannel() async throws -> any SFTPClientProtocol { - SFTPClient(worker: try await worker.openSibling()) - } - - public func uploadFile( - from localURL: URL, to remotePath: String, overwrite: Bool, - cancellation: TransferCancellationToken, resume: SFTPResumeState, - progress: @escaping @Sendable (SFTPProgress) -> Void - ) async throws { - try await worker.uploadFile( - from: localURL, to: remotePath, overwrite: overwrite, - cancellation: cancellation, resume: resume, progress: progress) - } - public func downloadFile( - from remotePath: String, to localURL: URL, overwrite: Bool, - cancellation: TransferCancellationToken, resume: SFTPResumeState, - progress: @escaping @Sendable (SFTPProgress) -> Void - ) async throws { - try await worker.downloadFile( - from: remotePath, to: localURL, overwrite: overwrite, - cancellation: cancellation, resume: resume, progress: progress) - } - public func uploadStream( - fromSocket socket: Int32, to remotePath: String, byteCount: Int64, overwrite: Bool, - cancellation: TransferCancellationToken, resume: SFTPResumeState, - progress: @escaping @Sendable (SFTPProgress) -> Void, - sourceSucceeded: @escaping @Sendable () -> Bool - ) async throws { - try await worker.uploadFile( - fromSocket: socket, to: remotePath, byteCount: byteCount, overwrite: overwrite, - cancellation: cancellation, resume: resume, progress: progress, sourceSucceeded: sourceSucceeded) - } - - public func openDownloadStream( - at path: String, maximumBytes: Int64, cancellation: TransferCancellationToken - ) async throws -> any SFTPDownloadStream { - let (id, identity) = try await worker.openDownloadStream( - at: path, maximumBytes: maximumBytes, cancellation: cancellation) - return NativeSFTPDownloadStream(worker: worker, id: id, identity: identity) - } private let keyboardInteractive: SSHKeyboardInteractiveHandler? - private nonisolated let worker: SFTPWorker + private let logger: Logger + private var generation: UInt64 = 0 + private var session: SSHMultiplexedSession? + private var channel: (any SFTPClientProtocol)? public init( keyboardInteractive: SSHKeyboardInteractiveHandler? = nil, logger: Logger = ApplicationLogger.disabled("termind.sftp") ) { self.keyboardInteractive = keyboardInteractive - worker = SFTPWorker(logger: logger) + self.logger = logger } - private init(worker: SFTPWorker) { - keyboardInteractive = nil - self.worker = worker + deinit { + let session = session + let channel = channel + Task { + await channel?.disconnect() + await session?.disconnect() + } } public func connect( using configuration: SSHConnectionConfiguration, progress: SSHRouteProgressHandler? = nil ) async throws { - try await worker.connect( - using: SSHConnectionRequest(configuration: configuration, keyboardInteractive: keyboardInteractive), - progress: progress - ) + try await connect( + using: SSHConnectionRequest( + configuration: configuration, + keyboardInteractive: keyboardInteractive), + progress: progress, connectionFailure: nil) } public func connect( @@ -358,8 +324,7 @@ public actor SFTPClient: SFTPClientProtocol { ) async throws { try await connect( using: SSHConnectionRequest(configuration: configuration), - progress: progress, connectionFailure: connectionFailure - ) + progress: progress, connectionFailure: connectionFailure) } public func connect( @@ -367,73 +332,150 @@ public actor SFTPClient: SFTPClientProtocol { progress: SSHRouteProgressHandler?, connectionFailure: (@Sendable (String) -> Void)? ) async throws { - try await worker.connect( - using: request.withKeyboardInteractiveFallback(keyboardInteractive), progress: progress, - connectionFailure: connectionFailure - ) + generation &+= 1 + let token = generation + let previousChannel = channel + let previousSession = session + channel = nil + session = nil + await previousChannel?.disconnect() + await previousSession?.disconnect() + + let replacement = SSHMultiplexedSession(logger: logger) + do { + try await replacement.connect( + using: request.withKeyboardInteractiveFallback(keyboardInteractive), + progress: progress, connectionFailure: connectionFailure) + progress?( + .initializing( + request.configuration.hostID, + String(localized: "Starting SFTP subsystem"))) + let channel = try await replacement.openSFTPChannel() + guard generation == token else { + await channel.disconnect() + await replacement.disconnect() + throw CancellationError() + } + session = replacement + self.channel = channel + } catch { + await replacement.disconnect() + throw error + } } - public func disconnect() async { await worker.disconnect() } + public func disconnect() async { + generation &+= 1 + let channel = channel + let session = session + self.channel = nil + self.session = nil + await channel?.disconnect() + await session?.disconnect() + } + + public func openSiblingChannel() async throws -> any SFTPClientProtocol { + try await connectedChannel().openSiblingChannel() + } public func homeDirectory() async throws -> String { - try await worker.homeDirectory() + try await connectedChannel().homeDirectory() } public func listDirectory(at path: String) async throws -> [RemoteFile] { - try await worker.listDirectory(at: path) + try await connectedChannel().listDirectory(at: path) } public func attributes(at path: String) async throws -> RemoteFile? { - try await worker.attributes(at: path) + try await connectedChannel().attributes(at: path) } public func readFile(at path: String, maximumBytes: Int) async throws -> Data { - try await worker.readFile(at: path, maximumBytes: maximumBytes) + try await connectedChannel().readFile(at: path, maximumBytes: maximumBytes) } public func readTextPage(at path: String, offset: Int, limit: Int) async throws -> AgentTextPage { - try await worker.readTextPage(at: path, offset: offset, limit: limit) + try await connectedChannel().readTextPage(at: path, offset: offset, limit: limit) } public func writeFile(_ data: Data, at path: String, overwrite: Bool) async throws { - try await worker.writeFile(data, at: path, overwrite: overwrite) + try await connectedChannel().writeFile(data, at: path, overwrite: overwrite) } public func setPermissions(at path: String, to permissions: UInt32) async throws { - try await worker.setPermissions(at: path, to: permissions) + try await connectedChannel().setPermissions(at: path, to: permissions) } public func createDirectory(at path: String) async throws { - try await worker.createDirectory(at: path) + try await connectedChannel().createDirectory(at: path) } public func rename(from source: String, to destination: String, overwrite: Bool) async throws { - try await worker.rename(from: source, to: destination, overwrite: overwrite) + try await connectedChannel().rename( + from: source, to: destination, overwrite: overwrite) } - public func removeFile(at path: String) async throws { try await worker.removeFile(at: path) } - public func removeDirectory(at path: String) async throws { try await worker.removeDirectory(at: path) } + public func removeFile(at path: String) async throws { + try await connectedChannel().removeFile(at: path) + } + + public func removeDirectory(at path: String) async throws { + try await connectedChannel().removeDirectory(at: path) + } + + public func uploadFile( + from localURL: URL, to remotePath: String, overwrite: Bool, + cancellation: TransferCancellationToken, resume: SFTPResumeState, + progress: @escaping @Sendable (SFTPProgress) -> Void + ) async throws { + try await connectedChannel().uploadFile( + from: localURL, to: remotePath, overwrite: overwrite, + cancellation: cancellation, resume: resume, progress: progress) + } public func uploadFile( from localURL: URL, to remotePath: String, overwrite: Bool, cancellation: TransferCancellationToken, progress: @escaping @Sendable (SFTPProgress) -> Void ) async throws { - try await worker.uploadFile( + try await connectedChannel().uploadFile( from: localURL, to: remotePath, overwrite: overwrite, - cancellation: cancellation, progress: progress - ) + cancellation: cancellation, progress: progress) } public func uploadStream( - fromSocket socket: Int32, to remotePath: String, byteCount: Int64, overwrite: Bool, - cancellation: TransferCancellationToken, progress: @escaping @Sendable (SFTPProgress) -> Void, + fromSocket socket: Int32, to remotePath: String, byteCount: Int64, + overwrite: Bool, cancellation: TransferCancellationToken, + resume: SFTPResumeState, + progress: @escaping @Sendable (SFTPProgress) -> Void, sourceSucceeded: @escaping @Sendable () -> Bool ) async throws { - try await worker.uploadFile( - fromSocket: socket, to: remotePath, byteCount: byteCount, overwrite: overwrite, - cancellation: cancellation, progress: progress, sourceSucceeded: sourceSucceeded - ) + try await connectedChannel().uploadStream( + fromSocket: socket, to: remotePath, byteCount: byteCount, + overwrite: overwrite, cancellation: cancellation, resume: resume, + progress: progress, sourceSucceeded: sourceSucceeded) + } + + public func uploadStream( + fromSocket socket: Int32, to remotePath: String, byteCount: Int64, + overwrite: Bool, cancellation: TransferCancellationToken, + progress: @escaping @Sendable (SFTPProgress) -> Void, + sourceSucceeded: @escaping @Sendable () -> Bool + ) async throws { + try await connectedChannel().uploadStream( + fromSocket: socket, to: remotePath, byteCount: byteCount, + overwrite: overwrite, cancellation: cancellation, + progress: progress, sourceSucceeded: sourceSucceeded) + } + + public func downloadFile( + from remotePath: String, to localURL: URL, overwrite: Bool, + cancellation: TransferCancellationToken, resume: SFTPResumeState, + progress: @escaping @Sendable (SFTPProgress) -> Void + ) async throws { + try await connectedChannel().downloadFile( + from: remotePath, to: localURL, overwrite: overwrite, + cancellation: cancellation, resume: resume, progress: progress) } public func downloadFile( @@ -441,66 +483,40 @@ public actor SFTPClient: SFTPClientProtocol { cancellation: TransferCancellationToken, progress: @escaping @Sendable (SFTPProgress) -> Void ) async throws { - try await worker.downloadFile( + try await connectedChannel().downloadFile( from: remotePath, to: localURL, overwrite: overwrite, - cancellation: cancellation, progress: progress - ) + cancellation: cancellation, progress: progress) + } + + public func openDownloadStream( + at path: String, maximumBytes: Int64, + cancellation: TransferCancellationToken + ) async throws -> any SFTPDownloadStream { + try await connectedChannel().openDownloadStream( + at: path, maximumBytes: maximumBytes, cancellation: cancellation) } public func copyFile( - from sourcePath: String, - to destinationPath: String, - on destination: any SFTPClientProtocol, - byteCount: Int64, - overwrite: Bool, - cancellation: TransferCancellationToken, + from sourcePath: String, to destinationPath: String, + on destination: any SFTPClientProtocol, byteCount: Int64, + overwrite: Bool, cancellation: TransferCancellationToken, progress: @escaping @Sendable (SFTPProgress) -> Void ) async throws { - if let destination = destination as? SFTPClient, self === destination { - try await worker.copyFile( - from: sourcePath, to: destinationPath, byteCount: byteCount, - overwrite: overwrite, cancellation: cancellation, progress: progress) - return - } - let relay = try SFTPSocketRelay() - let observer = cancellation.observeCancellation { relay.stop() } - defer { - cancellation.removeObserver(observer) - relay.close() - } - try await withTaskCancellationHandler { - try Task.checkCancellation() - try await withThrowingTaskGroup(of: Void.self) { group in - group.addTask { - do { - try await self.worker.downloadFile( - from: sourcePath, toSocket: relay.writer, - cancellation: cancellation - ) - relay.finishWriting() - } catch { - cancellation.cancel() - throw error - } - } - group.addTask { - try await destination.uploadStream( - fromSocket: relay.reader, to: destinationPath, byteCount: byteCount, - overwrite: overwrite, cancellation: cancellation, progress: progress, - sourceSucceeded: { relay.sourceSucceeded }) - } - do { - while try await group.next() != nil {} - } catch { - cancellation.cancel() - relay.stop() - group.cancelAll() - throw error - } + let destinationChannel = + if let destination = destination as? SFTPClient { + try await destination.connectedChannel() + } else { + destination } - } onCancel: { - cancellation.cancel() - } + try await connectedChannel().copyFile( + from: sourcePath, to: destinationPath, on: destinationChannel, + byteCount: byteCount, overwrite: overwrite, + cancellation: cancellation, progress: progress) + } + + private func connectedChannel() throws -> any SFTPClientProtocol { + guard let channel else { throw SFTPError.notConnected } + return channel } } @@ -553,1283 +569,3 @@ nonisolated final class SFTPSocketRelay: @unchecked Sendable { } } } - -private nonisolated final class NativeSFTPDownloadStream: SFTPDownloadStream { - let worker: SFTPWorker - let id: UUID - let sourceIdentity: SFTPSourceIdentity? - var byteCount: Int64 { sourceIdentity!.byteCount } - init(worker: SFTPWorker, id: UUID, identity: SFTPSourceIdentity) { - self.worker = worker - self.id = id - self.sourceIdentity = identity - } - func seek(to offset: Int64) async throws { try await worker.seekDownloadStream(id: id, offset: offset) } - func readChunk() async throws -> Data? { try await worker.readDownloadChunk(id: id) } - func close() async throws { try await worker.closeDownloadStream(id: id) } -} - -private nonisolated final class SFTPConnectionOwner: @unchecked Sendable { - let queue = DispatchQueue(label: "com.akinokaede.termind.ssh.sftp") - private let queueKey = DispatchSpecificKey() - var routeTransport: SSHRoutedSocket? - var session: OpaquePointer? - var keepaliveTimer: (any DispatchSourceTimer)? - var connectionFailure: (@Sendable (String) -> Void)? - var didReportConnectionFailure = false - var channelCount = 0 - - init() { queue.setSpecific(key: queueKey, value: 1) } - - func performSync(_ operation: () -> Void) { - if DispatchQueue.getSpecific(key: queueKey) != nil { - operation() - } else { - queue.sync(execute: operation) - } - } - - func install(_ opened: SSHOpenedSession, connectionFailure: (@Sendable (String) -> Void)?) { - precondition(channelCount == 0 && session == nil) - routeTransport = opened.transport - session = opened.session - self.connectionFailure = connectionFailure - didReportConnectionFailure = false - libssh2_keepalive_config(opened.session, 1, 15) - let timer = DispatchSource.makeTimerSource(queue: queue) - timer.schedule(deadline: .now() + 15, repeating: 15) - timer.setEventHandler { [weak self] in - guard let self, self.session == opened.session else { return } - var secondsToNext: Int32 = 0 - let result = libssh2_keepalive_send(opened.session, &secondsToNext) - if result < 0, result != LIBSSH2_ERROR_EAGAIN { - let error = SSHSessionWorker.sessionError( - opened.session, prefix: "SFTP keepalive failed") - self.reportConnectionFailure(error.localizedDescription) - } - } - keepaliveTimer = timer - timer.resume() - } - - func retainChannel() { channelCount += 1 } - - func releaseChannel() { - precondition(channelCount > 0) - channelCount -= 1 - if channelCount == 0 { closeTransport() } - } - - func closeIfUnused() { - if channelCount == 0 { closeTransport() } - } - - func reportConnectionFailure(_ message: String) { - guard !didReportConnectionFailure else { return } - didReportConnectionFailure = true - connectionFailure?(message) - } - - private func closeTransport() { - keepaliveTimer?.cancel() - keepaliveTimer = nil - if let session { - _ = libssh2_session_disconnect_ex( - session, SSH_DISCONNECT_BY_APPLICATION, "Termind closed SFTP", "") - _ = libssh2_session_free(session) - self.session = nil - } - routeTransport?.close() - routeTransport = nil - connectionFailure = nil - didReportConnectionFailure = false - } -} - -private nonisolated final class SFTPWorker: @unchecked Sendable { - private struct DownloadHandle { - var handle: OpaquePointer - var byteCount: Int64 - var modificationTime: UInt? - var received: Int64 = 0 - var eof = false - var cancellation: TransferCancellationToken - } - private final class RemoteReadContext: @unchecked Sendable { - let handle: OpaquePointer - init(_ handle: OpaquePointer) { self.handle = handle } - } - - private final class RemoteWriteContext: @unchecked Sendable { - let sftp: OpaquePointer - let handle: OpaquePointer - let temporary: String - let destination: String - let overwrite: Bool - let resume: SFTPResumeState? - let editPolicy: SFTPEditUploadPolicy? - var handleOpen = true - - init( - sftp: OpaquePointer, handle: OpaquePointer, temporary: String, - destination: String, overwrite: Bool, resume: SFTPResumeState?, - editPolicy: SFTPEditUploadPolicy? - ) { - self.sftp = sftp - self.handle = handle - self.temporary = temporary - self.destination = destination - self.overwrite = overwrite - self.resume = resume - self.editPolicy = editPolicy - } - } - private var downloadHandles: [UUID: DownloadHandle] = [:] - private let owner: SFTPConnectionOwner - private var queue: DispatchQueue { owner.queue } - private var session: OpaquePointer? { owner.session } - private var sftp: OpaquePointer? - private let logger: Logger - - init(logger: Logger) { - owner = SFTPConnectionOwner() - self.logger = logger - } - - private init(owner: SFTPConnectionOwner, sftp: OpaquePointer, logger: Logger) { - self.owner = owner - self.sftp = sftp - self.logger = logger - } - - deinit { owner.performSync { cleanup() } } - - public func connect( - using request: SSHConnectionRequest, - progress: SSHRouteProgressHandler?, - connectionFailure: (@Sendable (String) -> Void)? = nil - ) async throws { - let operationID = UUID().uuidString.lowercased() - let hostID = request.configuration.hostID - logger.info( - "sftp.connect.started", - metadata: [ - "host_id": RemoteLogging.hostID(hostID), "operation_id": "\(operationID)", - "protocol": "sftp", "stage": "connecting", - ]) - await queue.perform { self.cleanup() } - do { - let opened = try await SSHSessionWorker.openAuthenticatedSession( - using: request, timeout: 20, progress: progress, logger: logger - ) - try await queue.perform { - let session = opened.session - self.owner.install(opened, connectionFailure: connectionFailure) - progress?(.initializing(request.configuration.hostID, String(localized: "Starting SFTP subsystem"))) - guard let sftp = libssh2_sftp_init(session) else { - throw SSHSessionWorker.sessionError(session, prefix: "Unable to start SFTP") - } - self.owner.retainChannel() - self.sftp = sftp - } - try Task.checkCancellation() - logger.info( - "sftp.connect.completed", - metadata: [ - "host_id": RemoteLogging.hostID(hostID), "operation_id": "\(operationID)", - "outcome": "success", "protocol": "sftp", "stage": "connected", - ]) - } catch { - await queue.perform { self.cleanup() } - logger.error( - "sftp.connect.failed", - metadata: [ - "error_code": RemoteLogging.errorCode(error), "host_id": RemoteLogging.hostID(hostID), - "operation_id": "\(operationID)", "outcome": "failure", "protocol": "sftp", - "stage": "connecting", - ]) - throw error - } - } - - public func disconnect() async { await queue.perform { self.cleanup() } } - - func openSibling() async throws -> SFTPWorker { - try await queue.perform { - guard let session = self.session else { throw SFTPError.notConnected } - guard let sibling = libssh2_sftp_init(session) else { - throw SSHSessionWorker.sessionError(session, prefix: "Unable to start SFTP") - } - self.owner.retainChannel() - return SFTPWorker(owner: self.owner, sftp: sibling, logger: self.logger) - } - } - - public func homeDirectory() async throws -> String { - try await queue.perform { - let sftp = try self.requireSFTP() - let source = "." - var target = [CChar](repeating: 0, count: 4 * 1024) - let count = source.withCString { sourceBuffer in - libssh2_sftp_symlink_ex( - sftp, sourceBuffer, UInt32(source.utf8.count), - &target, UInt32(target.count), Int32(LIBSSH2_SFTP_REALPATH) - ) - } - guard count > 0 else { throw self.error("Unable to resolve the remote home directory") } - let bytes = target.prefix(Int(count)).map { UInt8(bitPattern: $0) } - return RemotePath.normalized(String(decoding: bytes, as: UTF8.self)) - } - } - - public func listDirectory(at path: String) async throws -> [RemoteFile] { - try await queue.perform { - let sftp = try self.requireSFTP() - guard - let handle = path.withCString({ - libssh2_sftp_open_ex( - sftp, $0, UInt32(path.utf8.count), 0, 0, - Int32(LIBSSH2_SFTP_OPENDIR) - ) - }) - else { - throw self.error("Unable to open \(path)") - } - defer { _ = libssh2_sftp_close_handle(handle) } - var result: [RemoteFile] = [] - var name = [CChar](repeating: 0, count: 4 * 1024) - while true { - var attributes = LIBSSH2_SFTP_ATTRIBUTES() - let count = libssh2_sftp_readdir_ex( - handle, &name, name.count, nil, 0, &attributes - ) - if count > 0 { - let bytes = name.prefix(Int(count)).map { UInt8(bitPattern: $0) } - let filename = String(decoding: bytes, as: UTF8.self) - guard filename != ".", filename != ".." else { continue } - result.append( - Self.remoteFile(path: Self.join(path, filename), name: filename, attributes: attributes)) - } else if count == 0 { - break - } else { - throw self.error("Unable to read \(path)") - } - } - return result.sorted { - if $0.isDirectory != $1.isDirectory { return $0.isDirectory } - return $0.name.localizedStandardCompare($1.name) == .orderedAscending - } - } - } - - public func attributes(at path: String) async throws -> RemoteFile? { - try await queue.perform { - let sftp = try self.requireSFTP() - 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 == 0 { - return Self.remoteFile( - path: path, name: URL(fileURLWithPath: path).lastPathComponent, - attributes: attributes - ) - } - if libssh2_sftp_last_error(sftp) == LIBSSH2_FX_NO_SUCH_FILE { return nil } - throw self.error("Unable to inspect \(path)") - } - } - - func openDownloadStream( - at path: String, maximumBytes: Int64, cancellation: TransferCancellationToken - ) async throws -> (UUID, SFTPSourceIdentity) { - try await queue.perform { - guard maximumBytes >= 0 else { throw SFTPDownloadFailure.tooLarge } - guard !cancellation.isCancelled else { throw SFTPError.cancelled } - let sftp = try self.requireSFTP() - var attributes = LIBSSH2_SFTP_ATTRIBUTES() - let inspected = path.withCString { - libssh2_sftp_stat_ex(sftp, $0, UInt32(path.utf8.count), Int32(LIBSSH2_SFTP_LSTAT), &attributes) - } - if inspected != 0 { - if libssh2_sftp_last_error(sftp) == LIBSSH2_FX_NO_SUCH_FILE { throw SFTPDownloadFailure.missingFile } - throw self.error("Unable to inspect download source") - } - guard attributes.flags & UInt(LIBSSH2_SFTP_ATTR_PERMISSIONS) != 0, - attributes.permissions & 0o170000 == 0o100000 - else { throw SFTPDownloadFailure.unsupportedType } - guard - let handle = path.withCString({ - libssh2_sftp_open_ex( - sftp, $0, UInt32(path.utf8.count), UInt(LIBSSH2_FXF_READ), 0, Int32(LIBSSH2_SFTP_OPENFILE)) - }) - else { throw self.error("Unable to open download source") } - do { - guard libssh2_sftp_fstat_ex(handle, &attributes, 0) == 0 else { - throw self.error("Unable to inspect download handle") - } - guard attributes.flags & UInt(LIBSSH2_SFTP_ATTR_PERMISSIONS) != 0, - attributes.permissions & 0o170000 == 0o100000 - else { throw SFTPDownloadFailure.unsupportedType } - guard attributes.flags & UInt(LIBSSH2_SFTP_ATTR_SIZE) != 0 else { - throw self.error("Unable to determine download size") - } - guard attributes.filesize <= UInt64(maximumBytes), attributes.filesize <= UInt64(Int64.max) - else { throw SFTPDownloadFailure.tooLarge } - guard !cancellation.isCancelled else { throw SFTPError.cancelled } - let id = UUID() - let count = Int64(attributes.filesize) - self.downloadHandles[id] = DownloadHandle( - handle: handle, byteCount: count, - modificationTime: attributes.flags & UInt(LIBSSH2_SFTP_ATTR_ACMODTIME) != 0 - ? attributes.mtime : nil, - cancellation: cancellation - ) - return (id, try self.remoteIdentity(handle)) - } catch { - _ = libssh2_sftp_close_handle(handle) - throw error - } - } - } - - func readDownloadChunk(id: UUID) async throws -> Data? { - try await queue.perform { - guard var entry = self.downloadHandles[id] else { throw SFTPError.notConnected } - guard !entry.cancellation.isCancelled else { throw SFTPError.cancelled } - if entry.eof { return nil } - var buffer = [UInt8](repeating: 0, count: 64 * 1_024) - let count = libssh2_sftp_read(entry.handle, &buffer, buffer.count) - if count > 0 { - guard Int64(count) <= entry.byteCount - entry.received else { throw SFTPDownloadFailure.sourceChanged } - entry.received += Int64(count) - self.downloadHandles[id] = entry - return Data(buffer.prefix(count)) - } - guard count == 0 else { throw self.error("Unable to read download source") } - var final = LIBSSH2_SFTP_ATTRIBUTES() - guard libssh2_sftp_fstat_ex(entry.handle, &final, 0) == 0 else { - throw self.error("Unable to verify download source") - } - guard final.flags & UInt(LIBSSH2_SFTP_ATTR_SIZE) != 0, - final.filesize == UInt64(entry.byteCount), entry.received == entry.byteCount - else { throw SFTPDownloadFailure.sourceChanged } - if let initialTime = entry.modificationTime { - guard final.flags & UInt(LIBSSH2_SFTP_ATTR_ACMODTIME) != 0, final.mtime == initialTime else { - throw SFTPDownloadFailure.sourceChanged - } - } - entry.eof = true - self.downloadHandles[id] = entry - return nil - } - } - - func seekDownloadStream(id: UUID, offset: Int64) async throws { - try await queue.perform { - guard var entry = self.downloadHandles[id] else { throw SFTPError.notConnected } - guard offset >= 0, offset <= entry.byteCount else { throw SFTPResumeFailure.partialChanged } - guard !entry.cancellation.isCancelled else { throw SFTPError.cancelled } - libssh2_sftp_seek64(entry.handle, UInt64(offset)) - entry.received = offset - entry.eof = false - self.downloadHandles[id] = entry - } - } - - func closeDownloadStream(id: UUID) async throws { - try await queue.perform { - guard let entry = self.downloadHandles.removeValue(forKey: id) else { return } - guard libssh2_sftp_close_handle(entry.handle) == 0 else { throw SFTPDownloadFailure.closeFailed } - } - } - - public func readTextPage(at path: String, offset: Int, limit: Int) async throws -> AgentTextPage { - try Task.checkCancellation() - let cancellation = TransferCancellationToken() - return try await withTaskCancellationHandler { - try await queue.perform { - var reader = try AgentTextPageReader(offset: offset, limit: limit) - return try self.withRemoteReadHandle(at: path) { handle in - var buffer = [CChar](repeating: 0, count: 16 * 1_024) - while !reader.isComplete { - guard !cancellation.isCancelled else { throw CancellationError() } - let count = libssh2_sftp_read(handle, &buffer, buffer.count) - if count > 0 { - try reader.append(Data(bytes: buffer, count: count)) - } else if count == 0 { - break - } else { - throw self.error("Unable to read \(path)") - } - } - guard !cancellation.isCancelled else { throw CancellationError() } - return try reader.finish() - } - } - } onCancel: { - cancellation.cancel() - } - } - - public func readFile(at path: String, maximumBytes: Int) async throws -> Data { - try await queue.perform { - guard maximumBytes > 0 else { return Data() } - return try self.withRemoteReadHandle(at: path) { handle in - var data = Data() - var buffer = [CChar](repeating: 0, count: min(64 * 1024, maximumBytes + 1)) - while data.count <= maximumBytes { - let wanted = min(buffer.count, maximumBytes + 1 - data.count) - let count = libssh2_sftp_read(handle, &buffer, wanted) - if count > 0 { - data.append(Data(bytes: buffer, count: count)) - } else if count == 0 { - break - } else { - throw self.error("Unable to read \(path)") - } - } - guard data.count <= maximumBytes else { - throw SFTPError.operation("\(path) exceeds the inline read limit of \(maximumBytes) bytes.") - } - return data - } - } - } - - public func writeFile(_ data: Data, at path: String, overwrite: Bool) async throws { - try await queue.perform { - let sftp = try self.requireSFTP() - if !overwrite { - var attributes = LIBSSH2_SFTP_ATTRIBUTES() - let exists = path.withCString { - libssh2_sftp_stat_ex( - sftp, $0, UInt32(path.utf8.count), - Int32(LIBSSH2_SFTP_LSTAT), &attributes - ) == 0 - } - if exists { throw SFTPError.operation("\(path) already exists.") } - } - try self.withAtomicRemoteWrite( - sftp: sftp, destination: path, overwrite: overwrite, - temporaryTag: ".termind-agent-part-" - ) { handle in - try data.withUnsafeBytes { bytes in - guard let base = bytes.baseAddress?.assumingMemoryBound(to: CChar.self) else { - return - } - var offset = 0 - while offset < bytes.count { - let written = libssh2_sftp_write( - handle, base.advanced(by: offset), bytes.count - offset - ) - guard written > 0 else { throw self.error("Unable to write \(path)") } - offset += written - } - } - } - } - } - - public func setPermissions(at path: String, to permissions: UInt32) async throws { - try await queue.perform { - let sftp = try self.requireSFTP() - var attributes = LIBSSH2_SFTP_ATTRIBUTES() - attributes.flags = UInt(LIBSSH2_SFTP_ATTR_PERMISSIONS) - attributes.permissions = UInt(permissions & 0o7777) - let result = path.withCString { - libssh2_sftp_stat_ex( - sftp, $0, UInt32(path.utf8.count), - Int32(LIBSSH2_SFTP_SETSTAT), &attributes - ) - } - guard result == 0 else { - throw self.error("Unable to change permissions for \(path)") - } - } - } - - public func createDirectory(at path: String) async throws { - try await queue.perform { - let sftp = try self.requireSFTP() - let result = path.withCString { - libssh2_sftp_mkdir_ex(sftp, $0, UInt32(path.utf8.count), 0o755) - } - guard result == 0 else { throw self.error("Unable to create \(path)") } - } - } - - public func rename(from source: String, to destination: String, overwrite: Bool) async throws { - try await queue.perform { - let sftp = try self.requireSFTP() - let flags = overwrite ? Int(LIBSSH2_SFTP_RENAME_OVERWRITE) : 0 - let result = source.withCString { sourceBuffer in - destination.withCString { destinationBuffer in - libssh2_sftp_rename_ex( - sftp, sourceBuffer, UInt32(source.utf8.count), - destinationBuffer, UInt32(destination.utf8.count), flags - ) - } - } - guard result == 0 else { throw self.error("Unable to rename \(source)") } - } - } - - public func removeFile(at path: String) async throws { - try await queue.perform { - let sftp = try self.requireSFTP() - guard - path.withCString({ - libssh2_sftp_unlink_ex(sftp, $0, UInt32(path.utf8.count)) - }) == 0 - else { - throw self.error("Unable to delete \(path)") - } - } - } - - public func removeDirectory(at path: String) async throws { - try await queue.perform { - let sftp = try self.requireSFTP() - guard - path.withCString({ - libssh2_sftp_rmdir_ex(sftp, $0, UInt32(path.utf8.count)) - }) == 0 - else { - throw self.error("Unable to delete \(path)") - } - } - } - - /// Alternating reads and writes on one worker avoids a socket-relay deadlock - /// when a serial fallback uses the same connection for both endpoints. - public func copyFile( - from source: String, to destination: String, byteCount: Int64, overwrite: Bool, - cancellation: TransferCancellationToken, - progress: @escaping @Sendable (SFTPProgress) -> Void - ) async throws { - try await queue.perform { - let sftp = try self.requireSFTP() - try self.withRemoteReadHandle(at: source) { input in - let identity = try self.remoteIdentity(input) - guard identity.byteCount == byteCount else { throw SFTPResumeFailure.sourceChanged } - try self.withAtomicRemoteWrite( - sftp: sftp, destination: destination, overwrite: overwrite, cancellation: cancellation, - validateCommit: { - guard !cancellation.isCancelled else { throw SFTPError.cancelled } - guard try self.remoteIdentity(input) == identity else { - throw SFTPResumeFailure.sourceChanged - } - } - ) { output in - var meter = SFTPProgressMeter(total: byteCount, report: progress) - var buffer = [CChar](repeating: 0, count: 64 * 1024) - while true { - guard !cancellation.isCancelled else { throw SFTPError.cancelled } - let count = libssh2_sftp_read(input, &buffer, buffer.count) - guard count >= 0 else { throw self.error("Unable to read \(source)") } - if count == 0 { break } - guard Int64(count) <= byteCount - meter.completed else { - throw SFTPResumeFailure.sourceChanged - } - try buffer.withUnsafeBufferPointer { bytes in - var offset = 0 - while offset < count { - guard !cancellation.isCancelled else { throw SFTPError.cancelled } - let written = libssh2_sftp_write(output, bytes.baseAddress! + offset, count - offset) - guard written > 0 else { throw self.error("Unable to upload \(destination)") } - offset += written - meter.add(written) - } - } - } - guard meter.completed == byteCount else { throw SFTPResumeFailure.sourceChanged } - meter.finish() - } - } - } - } - - public func uploadFile( - from localURL: URL, to remotePath: String, overwrite: Bool, - cancellation: TransferCancellationToken, resume: SFTPResumeState? = nil, - progress: @escaping @Sendable (SFTPProgress) -> Void - ) async throws { - let file = try FileHandle(forReadingFrom: localURL) - defer { try? file.close() } - let identity = try Self.localIdentity(localURL) - try resume?.validateSource(identity) - let offset = resume?.snapshot.offset ?? 0 - if let resume { try Self.verifyLocalPrefix(file, snapshot: resume.snapshot, cancellation: cancellation) } - guard - let context = try await queue.perform({ - try self.beginRemoteWrite( - destination: remotePath, overwrite: overwrite, - resume: resume, cancellation: cancellation) - }) - else { return } - do { - var meter = SFTPProgressMeter(total: identity.byteCount, offset: offset, report: progress) - while true { - if cancellation.isCancelled { throw SFTPError.cancelled } - let data = try file.read(upToCount: 32_700) ?? Data() - if data.isEmpty { break } - guard Int64(data.count) <= identity.byteCount - meter.completed else { - throw SFTPResumeFailure.sourceChanged - } - var position = 0 - while position < data.count { - if cancellation.isCancelled { throw SFTPError.cancelled } - let currentPosition = position - let written = try await queue.perform { - data.withUnsafeBytes { bytes in - libssh2_sftp_write( - context.handle, - bytes.baseAddress!.advanced(by: currentPosition).assumingMemoryBound(to: CChar.self), - bytes.count - currentPosition) - } - } - guard written > 0 else { throw error("Unable to upload \(remotePath)") } - resume?.acknowledge(data.subdata(in: position..<(position + written))) - position += written - meter.add(written) - } - } - guard meter.completed == identity.byteCount else { throw SFTPResumeFailure.sourceChanged } - guard !cancellation.isCancelled else { throw SFTPError.cancelled } - let sourceStillValid = try Self.localIdentity(localURL) == identity - try await queue.perform { try self.finishRemoteWrite(context, sourceStillValid: sourceStillValid) } - meter.finish() - } catch { - do { try await queue.perform { try self.abortRemoteWrite(context, cause: error) } } catch { throw error } - } - } - - public func downloadFile( - from remotePath: String, to localURL: URL, overwrite: Bool, - cancellation: TransferCancellationToken, resume: SFTPResumeState? = nil, - progress: @escaping @Sendable (SFTPProgress) -> Void - ) async throws { - let context = try await queue.perform { try self.beginRemoteRead(at: remotePath) } - defer { queue.async { self.closeRemoteRead(context) } } - let fm = FileManager.default - let temporary = URL( - fileURLWithPath: resume?.temporaryPath(for: localURL.path) - ?? (localURL.path + ".termind-part-" + UUID().uuidString)) - let snapshot = resume?.snapshot - let offset = snapshot?.offset ?? 0 - if snapshot?.source == nil { - guard fm.createFile(atPath: temporary.path, contents: nil) else { - throw SFTPError.localFile("Unable to create \(temporary.path)") - } - } - var succeeded = false - defer { if !succeeded && resume == nil { try? fm.removeItem(at: temporary) } } - let file = try FileHandle(forUpdating: temporary) - defer { try? file.close() } - let identity = try await queue.perform { try self.remoteIdentity(context.handle) } - try resume?.validateSource(identity) - if let snapshot { - guard try file.seekToEnd() == UInt64(offset) else { throw SFTPResumeFailure.partialChanged } - try file.seek(toOffset: 0) - try Self.verifyLocalPrefix(file, snapshot: snapshot, cancellation: cancellation) - try await queue.perform { - try self.verifyRemotePrefix( - context.handle, snapshot: snapshot, cancellation: cancellation) - } - } - var meter = SFTPProgressMeter(total: identity.byteCount, offset: offset, report: progress) - while true { - if cancellation.isCancelled { throw SFTPError.cancelled } - let data: Data? = try await queue.perform { - var buffer = [CChar](repeating: 0, count: 32_700) - let count = libssh2_sftp_read(context.handle, &buffer, buffer.count) - if count > 0 { return Data(bytes: buffer, count: count) } - if count == 0 { return nil } - throw self.error("Unable to download \(remotePath)") - } - guard let data else { break } - guard Int64(data.count) <= identity.byteCount - meter.completed else { - throw SFTPResumeFailure.sourceChanged - } - try file.write(contentsOf: data) - resume?.acknowledge(data) - meter.add(data.count) - } - guard meter.completed == identity.byteCount, - try await queue.perform({ try self.remoteIdentity(context.handle) }) == identity - else { throw SFTPResumeFailure.sourceChanged } - if resume?.snapshot.capturesFileVersion == true { - let fileVersion = try await queue.perform { () -> SFTPFileVersion in - var downloaded = LIBSSH2_SFTP_ATTRIBUTES() - guard libssh2_sftp_fstat_ex(context.handle, &downloaded, 0) == 0 else { - throw self.error(String(localized: "Unable to inspect edited file")) - } - return SFTPFileVersion( - Self.remoteFile( - path: remotePath, name: (remotePath as NSString).lastPathComponent, - attributes: downloaded)) - } - resume?.recordCommittedVersion(fileVersion) - } - guard !cancellation.isCancelled else { throw SFTPError.cancelled } - try file.synchronize() - meter.finish() - if fm.fileExists(atPath: localURL.path) { - guard overwrite else { throw SFTPError.localFile("\(localURL.lastPathComponent) already exists.") } - _ = try fm.replaceItemAt(localURL, withItemAt: temporary) - } else { - try fm.moveItem(at: temporary, to: localURL) - } - resume?.didCommit() - succeeded = true - } - - private static func localIdentity(_ url: URL) throws -> SFTPSourceIdentity { - let values = try url.resourceValues(forKeys: [.fileSizeKey, .contentModificationDateKey, .isRegularFileKey]) - guard values.isRegularFile == true, let size = values.fileSize else { throw SFTPResumeFailure.sourceChanged } - return SFTPSourceIdentity(byteCount: Int64(size), modifiedAt: values.contentModificationDate) - } - - private func remoteIdentity(_ handle: OpaquePointer) throws -> SFTPSourceIdentity { - var attrs = LIBSSH2_SFTP_ATTRIBUTES() - guard libssh2_sftp_fstat_ex(handle, &attrs, 0) == 0 else { throw error("Unable to inspect transfer file") } - guard attrs.flags & UInt(LIBSSH2_SFTP_ATTR_SIZE) != 0, attrs.filesize <= UInt64(Int64.max), - attrs.flags & UInt(LIBSSH2_SFTP_ATTR_PERMISSIONS) != 0, attrs.permissions & 0o170000 == 0o100000 - else { throw SFTPResumeFailure.sourceChanged } - return SFTPSourceIdentity( - byteCount: Int64(attrs.filesize), - modifiedAt: - attrs.flags & UInt(LIBSSH2_SFTP_ATTR_ACMODTIME) != 0 - ? Date(timeIntervalSince1970: Double(attrs.mtime)) : nil) - } - - private static func verifyLocalPrefix( - _ file: FileHandle, snapshot: SFTPResumeState.Snapshot, cancellation: TransferCancellationToken - ) throws { - 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 } - } - - private func verifyRemotePrefix( - _ handle: OpaquePointer, snapshot: SFTPResumeState.Snapshot, cancellation: TransferCancellationToken - ) throws { - libssh2_sftp_seek64(handle, 0) - var remaining = snapshot.offset - var digest = SHA256() - var buffer = [UInt8](repeating: 0, count: 64 * 1024) - while remaining > 0 { - guard !cancellation.isCancelled else { throw SFTPError.cancelled } - let count = libssh2_sftp_read(handle, &buffer, Int(min(remaining, Int64(buffer.count)))) - guard count > 0 else { - if count < 0 { throw error("Unable to verify incomplete file") } - throw SFTPResumeFailure.partialChanged - } - digest.update(data: Data(buffer.prefix(count))) - remaining -= Int64(count) - } - guard digest.finalize() == snapshot.digest.finalize() else { throw SFTPResumeFailure.partialChanged } - libssh2_sftp_seek64(handle, UInt64(snapshot.offset)) - } - - public func downloadFile( - from remotePath: String, toSocket socket: Int32, - cancellation: TransferCancellationToken - ) async throws { - let context = try await queue.perform { try self.beginRemoteRead(at: remotePath) } - defer { queue.async { self.closeRemoteRead(context) } } - while true { - if cancellation.isCancelled { throw SFTPError.cancelled } - let data: Data? = try await queue.perform { - var buffer = [CChar](repeating: 0, count: 32_700) - let count = libssh2_sftp_read(context.handle, &buffer, buffer.count) - if count > 0 { return Data(bytes: buffer, count: count) } - if count == 0 { return nil } - throw self.error("Unable to read \(remotePath)") - } - guard let data else { break } - var offset = 0 - while offset < data.count { - if cancellation.isCancelled { throw SFTPError.cancelled } - let sent = data.withUnsafeBytes { - Darwin.send(socket, $0.baseAddress!.advanced(by: offset), data.count - offset, 0) - } - if sent > 0 { - offset += sent - } else if sent < 0, errno == EINTR { - continue - } else { - throw SFTPError.operation("The SFTP copy relay was interrupted.") - } - } - } - } - - public func uploadFile( - fromSocket socket: Int32, to remotePath: String, - byteCount: Int64, overwrite: Bool, - cancellation: TransferCancellationToken, resume: SFTPResumeState? = nil, - progress: @escaping @Sendable (SFTPProgress) -> Void, - sourceSucceeded: @escaping @Sendable () -> Bool - ) async throws { - guard - let context = try await queue.perform({ - try self.beginRemoteWrite( - destination: remotePath, overwrite: overwrite, - resume: resume, cancellation: cancellation) - }) - else { return } - do { - var meter = SFTPProgressMeter(total: byteCount, offset: resume?.snapshot.offset ?? 0, report: progress) - var buffer = [CChar](repeating: 0, count: 32_700) - while true { - if cancellation.isCancelled { throw SFTPError.cancelled } - let count = Darwin.recv(socket, &buffer, buffer.count, 0) - if count > 0 { - guard Int64(count) <= byteCount - meter.completed else { - throw SFTPError.operation(String(localized: "The SFTP copy source grew during transfer.")) - } - let data = Data(bytes: buffer, count: count) - var offset = 0 - while offset < data.count { - if cancellation.isCancelled { throw SFTPError.cancelled } - let currentOffset = offset - let written = try await queue.perform { - data.withUnsafeBytes { - libssh2_sftp_write( - context.handle, - $0.baseAddress!.advanced(by: currentOffset).assumingMemoryBound(to: CChar.self), - data.count - currentOffset) - } - } - guard written > 0 else { throw error("Unable to copy to \(remotePath)") } - resume?.acknowledge(data.subdata(in: offset..<(offset + written))) - offset += written - meter.add(written) - } - } else if count == 0 { - break - } else if errno != EINTR { - throw SFTPError.operation("The SFTP copy relay was interrupted.") - } - } - guard !cancellation.isCancelled, sourceSucceeded() else { throw SFTPError.cancelled } - guard meter.completed == byteCount else { - throw SFTPError.operation( - "The source changed while it was being copied " - + "(expected \(byteCount) bytes, received \(meter.completed)).") - } - try await queue.perform { try self.finishRemoteWrite(context, sourceStillValid: true) } - meter.finish() - } catch { - do { try await queue.perform { try self.abortRemoteWrite(context, cause: error) } } catch { throw error } - } - } - - private func beginRemoteRead(at path: String) throws -> RemoteReadContext { - let sftp = try requireSFTP() - guard - let handle = path.withCString({ - libssh2_sftp_open_ex( - sftp, $0, UInt32(path.utf8.count), UInt(LIBSSH2_FXF_READ), 0, - Int32(LIBSSH2_SFTP_OPENFILE)) - }) - else { throw error("Unable to open \(path)") } - return RemoteReadContext(handle) - } - - private func closeRemoteRead(_ context: RemoteReadContext) { - _ = libssh2_sftp_close_handle(context.handle) - } - - /// Prepares an atomic upload while holding the libssh2 owner. The returned - /// handle may then be advanced in bounded turns so sibling Channels share - /// the transport fairly. - private func beginRemoteWrite( - destination: String, overwrite: Bool, resume: SFTPResumeState?, - cancellation: TransferCancellationToken - ) throws -> RemoteWriteContext? { - if resume?.snapshot.committed == true { return nil } - let sftp = try requireSFTP() - let temporary = resume?.temporaryPath(for: destination) ?? (destination + ".termind-part-" + UUID().uuidString) - let continuing = resume?.snapshot.hasTemporaryFile == true - if continuing { - var attributes = LIBSSH2_SFTP_ATTRIBUTES() - let result = temporary.withCString { - libssh2_sftp_stat_ex(sftp, $0, UInt32(temporary.utf8.count), Int32(LIBSSH2_SFTP_LSTAT), &attributes) - } - if result != 0 { - guard libssh2_sftp_last_error(sftp) == LIBSSH2_FX_NO_SUCH_FILE else { - throw error("Unable to inspect incomplete file") - } - guard let resume, resume.snapshot.committing else { throw SFTPResumeFailure.partialChanged } - try withRemoteReadHandle(at: destination) { handle in - guard try remoteIdentity(handle).byteCount == resume.snapshot.offset else { - throw SFTPResumeFailure.partialChanged - } - try verifyRemotePrefix(handle, snapshot: resume.snapshot, cancellation: cancellation) - } - if let file = try editTarget(sftp: sftp, path: destination) { - resume.recordCommittedVersion(SFTPFileVersion(file)) - } - resume.didCommit() - return nil - } - guard attributes.flags & UInt(LIBSSH2_SFTP_ATTR_PERMISSIONS) != 0, - attributes.permissions & 0o170000 == 0o100000 - else { throw SFTPResumeFailure.partialChanged } - } - let editPolicy = resume?.snapshot.editPolicy - if let editPolicy { try editPolicy.validate(try editTarget(sftp: sftp, path: destination)) } - let flags = - resume == nil - ? UInt(LIBSSH2_FXF_WRITE | LIBSSH2_FXF_CREAT | LIBSSH2_FXF_TRUNC) - : UInt(LIBSSH2_FXF_READ | LIBSSH2_FXF_WRITE | (continuing ? 0 : (LIBSSH2_FXF_CREAT | LIBSSH2_FXF_EXCL))) - guard - let handle = temporary.withCString({ - libssh2_sftp_open_ex( - sftp, $0, UInt32(temporary.utf8.count), flags, 0o600, - Int32(LIBSSH2_SFTP_OPENFILE)) - }) - else { throw error("Unable to create \(destination)") } - resume?.createdTemporaryFile() - do { - if let resume { - let snapshot = resume.snapshot - let identity = try remoteIdentity(handle) - guard identity.byteCount >= snapshot.offset else { throw SFTPResumeFailure.partialChanged } - try verifyRemotePrefix(handle, snapshot: snapshot, cancellation: cancellation) - if identity.byteCount != snapshot.offset { - var size = LIBSSH2_SFTP_ATTRIBUTES() - size.flags = UInt(LIBSSH2_SFTP_ATTR_SIZE) - size.filesize = UInt64(snapshot.offset) - guard libssh2_sftp_fstat_ex(handle, &size, 1) == 0 else { - throw error("Unable to trim incomplete file") - } - } - } - return RemoteWriteContext( - sftp: sftp, handle: handle, temporary: temporary, - destination: destination, overwrite: overwrite, resume: resume, - editPolicy: editPolicy) - } catch { - _ = libssh2_sftp_close_handle(handle) - throw error - } - } - - private func finishRemoteWrite(_ context: RemoteWriteContext, sourceStillValid: Bool) throws { - guard sourceStillValid else { throw SFTPResumeFailure.sourceChanged } - if let resume = context.resume { - guard try remoteIdentity(context.handle).byteCount == resume.snapshot.offset else { - throw SFTPResumeFailure.partialChanged - } - } - if let editPolicy = context.editPolicy { - var mode = LIBSSH2_SFTP_ATTRIBUTES() - mode.flags = UInt(LIBSSH2_SFTP_ATTR_PERMISSIONS) - mode.permissions = UInt(editPolicy.permissions & 0o7777) - guard libssh2_sftp_fstat_ex(context.handle, &mode, 1) == 0 else { - throw error(String(localized: "Unable to preserve remote file permissions")) - } - var committed = LIBSSH2_SFTP_ATTRIBUTES() - guard libssh2_sftp_fstat_ex(context.handle, &committed, 0) == 0 else { - throw error(String(localized: "Unable to inspect edited file")) - } - context.resume?.recordCommittedVersion( - SFTPFileVersion( - Self.remoteFile( - path: context.destination, - name: (context.destination as NSString).lastPathComponent, - attributes: committed))) - } - let closed = libssh2_sftp_close_handle(context.handle) - context.handleOpen = false - guard closed == 0 else { throw error("Unable to close \(context.temporary)") } - if let editPolicy = context.editPolicy { - try editPolicy.validate(try editTarget(sftp: context.sftp, path: context.destination)) - } - var attributes = LIBSSH2_SFTP_ATTRIBUTES() - let exists = - context.destination.withCString { - libssh2_sftp_stat_ex( - context.sftp, $0, UInt32(context.destination.utf8.count), - Int32(LIBSSH2_SFTP_LSTAT), &attributes) - } == 0 - if exists && (!context.overwrite || (attributes.permissions & 0o170000) == 0o040000) { - throw SFTPStreamUploadOperation.Failure.targetConflict - } - context.resume?.willCommit() - try renameOnQueue( - sftp: context.sftp, from: context.temporary, - to: context.destination, overwrite: context.overwrite) - context.resume?.didCommit() - } - - private func abortRemoteWrite(_ context: RemoteWriteContext, cause: any Error) throws { - if context.handleOpen { - _ = libssh2_sftp_close_handle(context.handle) - context.handleOpen = false - } - if context.resume != nil { throw cause } - let removed = context.temporary.withCString { - libssh2_sftp_unlink_ex(context.sftp, $0, UInt32(context.temporary.utf8.count)) - } - if removed != 0 { throw SFTPTemporaryFileCleanupError(path: context.temporary, cause: cause) } - throw cause - } - - private func withRemoteReadHandle( - at path: String, read: (OpaquePointer) throws -> T - ) throws -> T { - let sftp = try requireSFTP() - guard - let handle = path.withCString({ - libssh2_sftp_open_ex( - sftp, $0, UInt32(path.utf8.count), UInt(LIBSSH2_FXF_READ), 0, - Int32(LIBSSH2_SFTP_OPENFILE) - ) - }) - else { - throw error("Unable to open \(path)") - } - defer { _ = libssh2_sftp_close_handle(handle) } - return try read(handle) - } - - private func withAtomicRemoteWrite( - sftp: OpaquePointer, - destination: String, - overwrite: Bool, - temporaryTag: String = ".termind-part-", - resume: SFTPResumeState? = nil, - cancellation: TransferCancellationToken = TransferCancellationToken(), - validateCommit: () throws -> Void = {}, - write: (OpaquePointer) throws -> Void - ) throws { - if resume?.snapshot.committed == true { return } - let temporary = resume?.temporaryPath(for: destination) ?? (destination + temporaryTag + UUID().uuidString) - let continuing = resume?.snapshot.hasTemporaryFile == true - if continuing { - var attributes = LIBSSH2_SFTP_ATTRIBUTES() - let result = temporary.withCString { - libssh2_sftp_stat_ex(sftp, $0, UInt32(temporary.utf8.count), Int32(LIBSSH2_SFTP_LSTAT), &attributes) - } - if result != 0 { - guard libssh2_sftp_last_error(sftp) == LIBSSH2_FX_NO_SUCH_FILE else { - throw error("Unable to inspect incomplete file") - } - guard let resume, resume.snapshot.committing else { throw SFTPResumeFailure.partialChanged } - try withRemoteReadHandle(at: destination) { handle in - guard try remoteIdentity(handle).byteCount == resume.snapshot.offset else { - throw SFTPResumeFailure.partialChanged - } - try verifyRemotePrefix(handle, snapshot: resume.snapshot, cancellation: cancellation) - } - if let file = try editTarget(sftp: sftp, path: destination) { - resume.recordCommittedVersion(SFTPFileVersion(file)) - } - resume.didCommit() - return - } - guard attributes.flags & UInt(LIBSSH2_SFTP_ATTR_PERMISSIONS) != 0, - attributes.permissions & 0o170000 == 0o100000 - else { throw SFTPResumeFailure.partialChanged } - } - let editPolicy = resume?.snapshot.editPolicy - if let editPolicy { try editPolicy.validate(try editTarget(sftp: sftp, path: destination)) } - let flags = - resume == nil - ? UInt(LIBSSH2_FXF_WRITE | LIBSSH2_FXF_CREAT | LIBSSH2_FXF_TRUNC) - : UInt(LIBSSH2_FXF_READ | LIBSSH2_FXF_WRITE | (continuing ? 0 : (LIBSSH2_FXF_CREAT | LIBSSH2_FXF_EXCL))) - guard - let handle = temporary.withCString({ - libssh2_sftp_open_ex( - sftp, $0, UInt32(temporary.utf8.count), flags, 0o600, - Int32(LIBSSH2_SFTP_OPENFILE) - ) - }) - else { - throw error("Unable to create \(destination)") - } - resume?.createdTemporaryFile() - var handleOpen = true - do { - if let resume { - let snapshot = resume.snapshot - let identity = try remoteIdentity(handle) - guard identity.byteCount >= snapshot.offset else { throw SFTPResumeFailure.partialChanged } - try verifyRemotePrefix(handle, snapshot: snapshot, cancellation: cancellation) - if identity.byteCount != snapshot.offset { - var size = LIBSSH2_SFTP_ATTRIBUTES() - size.flags = UInt(LIBSSH2_SFTP_ATTR_SIZE) - size.filesize = UInt64(snapshot.offset) - guard libssh2_sftp_fstat_ex(handle, &size, 1) == 0 else { - throw error("Unable to trim incomplete file") - } - } - } - try write(handle) - if let resume { - guard try remoteIdentity(handle).byteCount == resume.snapshot.offset else { - throw SFTPResumeFailure.partialChanged - } - } - if let editPolicy { - var mode = LIBSSH2_SFTP_ATTRIBUTES() - mode.flags = UInt(LIBSSH2_SFTP_ATTR_PERMISSIONS) - mode.permissions = UInt(editPolicy.permissions & 0o7777) - guard libssh2_sftp_fstat_ex(handle, &mode, 1) == 0 else { - throw self.error(String(localized: "Unable to preserve remote file permissions")) - } - var committed = LIBSSH2_SFTP_ATTRIBUTES() - guard libssh2_sftp_fstat_ex(handle, &committed, 0) == 0 else { - throw self.error(String(localized: "Unable to inspect edited file")) - } - resume?.recordCommittedVersion( - SFTPFileVersion( - Self.remoteFile( - path: destination, name: (destination as NSString).lastPathComponent, attributes: committed) - )) - } - let closed = libssh2_sftp_close_handle(handle) - handleOpen = false - guard closed == 0 else { throw error("Unable to close \(temporary)") } - try validateCommit() - if let editPolicy { try editPolicy.validate(try editTarget(sftp: sftp, path: destination)) } - // Do not turn an upload into destructive directory replacement. - var attributes = LIBSSH2_SFTP_ATTRIBUTES() - let exists = - destination.withCString { - libssh2_sftp_stat_ex( - sftp, $0, UInt32(destination.utf8.count), Int32(LIBSSH2_SFTP_LSTAT), &attributes) - } == 0 - if exists && (!overwrite || (attributes.permissions & 0o170000) == 0o040000) { - throw SFTPStreamUploadOperation.Failure.targetConflict - } - resume?.willCommit() - try renameOnQueue(sftp: sftp, from: temporary, to: destination, overwrite: overwrite) - resume?.didCommit() - } catch { - if handleOpen { _ = libssh2_sftp_close_handle(handle) } - if resume != nil { throw error } - let removed = temporary.withCString { - libssh2_sftp_unlink_ex(sftp, $0, UInt32(temporary.utf8.count)) - } - if removed != 0 { throw SFTPTemporaryFileCleanupError(path: temporary, cause: error) } - throw error - } - } - - private func editTarget(sftp: OpaquePointer, path: String) throws -> RemoteFile? { - 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 != 0 { - if libssh2_sftp_last_error(sftp) == LIBSSH2_FX_NO_SUCH_FILE { return nil } - throw error(String(localized: "Unable to inspect remote edit destination")) - } - return Self.remoteFile(path: path, name: (path as NSString).lastPathComponent, attributes: attributes) - } - - private func renameOnQueue(sftp: OpaquePointer, from source: String, to destination: String, overwrite: Bool) throws - { - let flags = overwrite ? Int(LIBSSH2_SFTP_RENAME_OVERWRITE) : 0 - let result = SFTPRenameOperation.perform( - overwrite: overwrite, - posixRename: { - source.withCString { sourceBuffer in - destination.withCString { destinationBuffer in - libssh2_sftp_posix_rename_ex( - sftp, sourceBuffer, source.utf8.count, destinationBuffer, destination.utf8.count) - } - } - }, - standardRename: { - source.withCString { sourceBuffer in - destination.withCString { destinationBuffer in - libssh2_sftp_rename_ex( - sftp, sourceBuffer, UInt32(source.utf8.count), - destinationBuffer, UInt32(destination.utf8.count), flags) - } - } - }, - lastStatus: { libssh2_sftp_last_error(sftp) }) - guard result == 0 else { throw error("Unable to finish \(destination)") } - } - - private func requireSFTP() throws -> OpaquePointer { - guard let sftp else { throw SFTPError.notConnected } - return sftp - } - - private func error(_ prefix: String) -> SFTPError { - guard let sftp else { return .operation(prefix) } - let sftpCode = libssh2_sftp_last_error(sftp) - let sessionCode = session.map(libssh2_session_last_errno) ?? 0 - let message = "\(prefix): SFTP error \(sftpCode)" - let lostSFTPCodes = [ - UInt(LIBSSH2_FX_NO_CONNECTION), UInt(LIBSSH2_FX_CONNECTION_LOST), - ] - let lostSessionCodes = [ - LIBSSH2_ERROR_SOCKET_SEND, LIBSSH2_ERROR_SOCKET_RECV, - LIBSSH2_ERROR_TIMEOUT, LIBSSH2_ERROR_CHANNEL_CLOSED, - ] - if lostSFTPCodes.contains(sftpCode) || lostSessionCodes.contains(sessionCode) { - reportConnectionFailure(message) - return .connectionLost(message) - } - return .operation(message) - } - - private func cleanup() { - for entry in downloadHandles.values { _ = libssh2_sftp_close_handle(entry.handle) } - downloadHandles.removeAll() - if let sftp { - _ = libssh2_sftp_shutdown(sftp) - self.sftp = nil - owner.releaseChannel() - } - owner.closeIfUnused() - } - - private func reportConnectionFailure(_ message: String) { - owner.reportConnectionFailure(message) - } - - private static func join(_ base: String, _ component: String) -> String { - base == "/" ? "/\(component)" : "\(base)/\(component)" - } - - private static func remoteFile( - path: String, name: String, attributes: LIBSSH2_SFTP_ATTRIBUTES - ) -> RemoteFile { - let mode = UInt32(attributes.permissions) - let kind: RemoteFile.Kind - if mode & UInt32(LIBSSH2_SFTP_S_IFMT) == UInt32(LIBSSH2_SFTP_S_IFDIR) { - kind = .directory - } else if mode & UInt32(LIBSSH2_SFTP_S_IFMT) == UInt32(LIBSSH2_SFTP_S_IFREG) { - kind = .file - } else if mode & UInt32(LIBSSH2_SFTP_S_IFMT) == UInt32(LIBSSH2_SFTP_S_IFLNK) { - kind = .symbolicLink - } else if mode == 0 { - kind = .unknown - } else { - kind = .special - } - return RemoteFile( - path: path, name: name, kind: kind, size: Int64(attributes.filesize), - modifiedAt: attributes.mtime == 0 ? nil : Date(timeIntervalSince1970: TimeInterval(attributes.mtime)), - permissions: mode, ownerUID: UInt32(attributes.uid), ownerGID: UInt32(attributes.gid) - ) - } -} diff --git a/Frameworks/TermindRemoteInfrastructure/SSH/SSHMultiplexedSFTPChannelOwner.swift b/Frameworks/TermindRemoteInfrastructure/SSH/SSHMultiplexedSFTPChannelOwner.swift new file mode 100644 index 00000000..7ae07a11 --- /dev/null +++ b/Frameworks/TermindRemoteInfrastructure/SSH/SSHMultiplexedSFTPChannelOwner.swift @@ -0,0 +1,176 @@ +import CSSH2 +import Foundation + +/// Owns every SFTP subsystem Channel attached to one non-blocking SSH session. +/// +/// The terminal and Agent workers keep their feature-specific PTY/exec pumps, +/// while this type centralizes the shared Channel lifecycle and bounded +/// libssh2 turns. All mutable state is confined to the SSH owner's queue. +nonisolated final class SSHMultiplexedSFTPChannelOwner: @unchecked Sendable { + struct Context: @unchecked Sendable { + var session: OpaquePointer + var transport: SSHBufferedTransport + var socket: Int32 + } + + typealias ContextProvider = @Sendable () -> Context? + typealias EventUpdater = @Sendable (OpaquePointer, SSHBufferedTransport) -> Void + + private let queue: DispatchQueue + private let context: ContextProvider + private let updateEvents: EventUpdater + private let didCloseLastChannel: @Sendable () -> Void + private let identity = UUID() + private var channels: [UUID: OpaquePointer] = [:] + private var opening = false + private var openWaiters: [CheckedContinuation] = [] + + init( + queue: DispatchQueue, + context: @escaping ContextProvider, + updateEvents: @escaping EventUpdater, + didCloseLastChannel: @escaping @Sendable () -> Void = {} + ) { + self.queue = queue + self.context = context + self.updateEvents = updateEvents + self.didCloseLastChannel = didCloseLastChannel + } + + func openClient() async throws -> any SFTPClientProtocol { + let driver = SFTPChannelDriver( + identity: identity, + open: { [weak self] in + guard let self else { throw SFTPError.notConnected } + return try await self.openChannel() + }, + close: { [weak self] id in await self?.closeChannel(id) }, + perform: { [weak self] id, operation in + guard let self else { throw SFTPError.notConnected } + return try await self.perform(id, operation) + }) + return try await SFTPChannel.open(using: driver) + } + + /// Called by the enclosing SSH owner while already executing on `queue`. + func closeAllOnQueue() { + for channel in channels.values { _ = libssh2_sftp_shutdown(channel) } + channels.removeAll() + } + + var isEmptyOnQueue: Bool { channels.isEmpty } + + private func openChannel() async throws -> UUID { + await reserveOpen() + defer { releaseOpen() } + let id = UUID() + while true { + try Task.checkCancellation() + let attempt: SFTPChannelAttempt = try await queue.perform { [self] in + guard let context = context() else { throw SFTPError.notConnected } + context.transport.beginPass( + deadline: DispatchTime.now().uptimeNanoseconds + + SSHForwardingTuning.current.sliceNanoseconds) + try context.transport.flush(socket: context.socket) + if context.transport.isBackpressured { return .again } + guard let sftp = libssh2_sftp_init(context.session) else { + if libssh2_session_last_errno(context.session) == LIBSSH2_ERROR_EAGAIN { + return .again + } + throw SSHSessionWorker.sessionError( + context.session, prefix: "Unable to start SFTP") + } + channels[id] = sftp + try context.transport.flush(socket: context.socket) + updateEvents(context.session, context.transport) + return .value(()) + } + if case .value = attempt { + do { + try Task.checkCancellation() + return id + } catch { + await closeChannel(id) + throw error + } + } + try await Task.sleep(for: .milliseconds(2)) + } + } + + private func reserveOpen() async { + await withCheckedContinuation { continuation in + queue.async { [self] in + if opening { + openWaiters.append(continuation) + } else { + opening = true + continuation.resume() + } + } + } + } + + private func releaseOpen() { + queue.async { [self] in + if openWaiters.isEmpty { + opening = false + } else { + openWaiters.removeFirst().resume() + } + } + } + + private func closeChannel(_ id: UUID) async { + while true { + let attempt: SFTPChannelAttempt = + (try? await queue.perform { [self] in + guard let sftp = channels[id] else { return .value(()) } + guard let context = context() else { + channels[id] = nil + if channels.isEmpty { didCloseLastChannel() } + return .value(()) + } + context.transport.beginPass( + deadline: DispatchTime.now().uptimeNanoseconds + + SSHForwardingTuning.current.sliceNanoseconds) + try context.transport.flush(socket: context.socket) + let result = libssh2_sftp_shutdown(sftp) + if result == LIBSSH2_ERROR_EAGAIN { return .again } + channels[id] = nil + try context.transport.flush(socket: context.socket) + updateEvents(context.session, context.transport) + if channels.isEmpty { didCloseLastChannel() } + return .value(()) + }) ?? .value(()) + if case .value = attempt { return } + try? await Task.sleep(for: .milliseconds(2)) + } + } + + private func perform( + _ id: UUID, _ operation: @escaping SFTPChannelDriver.RawOperation + ) async throws -> SFTPChannelValue { + while true { + try Task.checkCancellation() + let attempt = try await queue.perform { [self] in + guard let sftp = channels[id], let context = context() else { + throw SFTPError.notConnected + } + context.transport.beginPass( + deadline: DispatchTime.now().uptimeNanoseconds + + SSHForwardingTuning.current.sliceNanoseconds) + try context.transport.flush(socket: context.socket) + if context.transport.isBackpressured { return SFTPChannelAnyAttempt.again } + let result = try operation(context.session, sftp) + try context.transport.flush(socket: context.socket) + updateEvents(context.session, context.transport) + return result + } + switch attempt { + case .value(let value): return value + case .again: try await Task.sleep(for: .milliseconds(2)) + } + } + } +} diff --git a/Frameworks/TermindRemoteInfrastructure/SSH/SSHMultiplexedSession.swift b/Frameworks/TermindRemoteInfrastructure/SSH/SSHMultiplexedSession.swift index 6aa8a62b..3c6934cb 100644 --- a/Frameworks/TermindRemoteInfrastructure/SSH/SSHMultiplexedSession.swift +++ b/Frameworks/TermindRemoteInfrastructure/SSH/SSHMultiplexedSession.swift @@ -25,7 +25,7 @@ public nonisolated struct SSHExecResult: Hashable, Sendable { /// The one failure that is safe to retry on a replacement transport. It is /// emitted only before the remote process has started. -public nonisolated enum SSHExecSessionError: LocalizedError, Sendable { +public nonisolated enum SSHCommandExecutionError: LocalizedError, Sendable { case transportUnavailable(any Error) case channelUnavailable(any Error) @@ -55,10 +55,6 @@ nonisolated extension SSHMultiplexedSessionProtocol { public func connect(using request: SSHConnectionRequest) async throws { try await connect(using: request.configuration) } - - public func openSFTPChannel() async throws -> any SFTPClientProtocol { - throw SFTPError.notConnected - } } /// Called on the connection's serial libssh2 owner. It must return promptly. @@ -70,7 +66,7 @@ public actor SSHMultiplexedSession: SSHMultiplexedSessionProtocol { public init( keyboardInteractive: SSHKeyboardInteractiveHandler? = nil, - logger: Logger = ApplicationLogger.disabled("termind.ssh-exec") + logger: Logger = ApplicationLogger.disabled("termind.ssh-multiplexed-session") ) { self.keyboardInteractive = keyboardInteractive worker = SSHMultiplexedSessionWorker(logger: logger) @@ -81,7 +77,17 @@ public actor SSHMultiplexedSession: SSHMultiplexedSessionProtocol { } public func connect(using request: SSHConnectionRequest) async throws { - try await worker.connect(using: request.withKeyboardInteractiveFallback(keyboardInteractive)) + try await connect(using: request, progress: nil, connectionFailure: nil) + } + + public func connect( + using request: SSHConnectionRequest, + progress: SSHRouteProgressHandler?, + connectionFailure: (@Sendable (String) -> Void)? + ) async throws { + try await worker.connect( + using: request.withKeyboardInteractiveFallback(keyboardInteractive), + progress: progress, connectionFailure: connectionFailure) } public func execute( diff --git a/Frameworks/TermindRemoteInfrastructure/SSH/SSHMultiplexedSessionWorker.swift b/Frameworks/TermindRemoteInfrastructure/SSH/SSHMultiplexedSessionWorker.swift index 8de9c5f2..6570b775 100644 --- a/Frameworks/TermindRemoteInfrastructure/SSH/SSHMultiplexedSessionWorker.swift +++ b/Frameworks/TermindRemoteInfrastructure/SSH/SSHMultiplexedSessionWorker.swift @@ -78,7 +78,7 @@ nonisolated final class SSHMultiplexedSessionWorker: @unchecked Sendable { } } - private let queue = DispatchQueue(label: "com.akinokaede.termind.ssh.exec-multiplexer") + private let queue = DispatchQueue(label: "com.akinokaede.termind.ssh.multiplexed-session") private let logger: Logger private var routeTransport: SSHRoutedSocket? private var session: OpaquePointer? @@ -96,22 +96,41 @@ nonisolated final class SSHMultiplexedSessionWorker: @unchecked Sendable { private var keepaliveDue = true private var nextKeepalive = DispatchTime.now() private var scratch = [CChar](repeating: 0, count: 16 * 1024) - private var sftpChannels: [UUID: OpaquePointer] = [:] - private var sftpOpening = false - private var sftpOpenWaiters: [CheckedContinuation] = [] - private let sharedSFTPIdentity = UUID() + private var sftpChannelOwner: SSHMultiplexedSFTPChannelOwner! + private var connectionFailure: (@Sendable (String) -> Void)? + private var didReportConnectionFailure = false - init(logger: Logger) { self.logger = logger } + init(logger: Logger) { + self.logger = logger + sftpChannelOwner = SSHMultiplexedSFTPChannelOwner( + queue: queue, + context: { [weak self] in + guard let self, let session = self.session, + let bufferedTransport = self.bufferedTransport, + let routeTransport = self.routeTransport + else { return nil } + return .init( + session: session, transport: bufferedTransport, + socket: routeTransport.socket) + }, + updateEvents: { [weak self] session, transport in + self?.updateSharedTransportEvents(session: session, transport: transport) + }) + } deinit { queue.sync { cleanup(error: CancellationError()) } } - func connect(using request: SSHConnectionRequest) async throws { + func connect( + using request: SSHConnectionRequest, + progress: SSHRouteProgressHandler? = nil, + connectionFailure: (@Sendable (String) -> Void)? = nil + ) async throws { let operationID = UUID().uuidString.lowercased() let hostID = request.configuration.hostID logger.info( - "ssh.exec.connect.started", + "ssh.multiplexed.connect.started", metadata: [ "host_id": RemoteLogging.hostID(hostID), "operation_id": "\(operationID)", "protocol": "ssh", "stage": "connecting", @@ -119,20 +138,22 @@ nonisolated final class SSHMultiplexedSessionWorker: @unchecked Sendable { await disconnect() do { let opened = try await SSHSessionWorker.openAuthenticatedSession( - using: request, timeout: 20, logger: logger) + using: request, timeout: 20, progress: progress, logger: logger) try await queue.perform { try Task.checkCancellation() self.routeTransport = opened.transport self.session = opened.session self.hostID = hostID self.bufferedTransport = SSHBufferedTransport(session: opened.session) + self.connectionFailure = connectionFailure + self.didReportConnectionFailure = false libssh2_session_set_blocking(opened.session, 0) libssh2_keepalive_config(opened.session, 0, 30) self.startEvents(socket: opened.transport.socket) } try Task.checkCancellation() logger.info( - "ssh.exec.connect.completed", + "ssh.multiplexed.connect.completed", metadata: [ "host_id": RemoteLogging.hostID(hostID), "operation_id": "\(operationID)", "outcome": "success", "protocol": "ssh", "stage": "connected", @@ -140,7 +161,7 @@ nonisolated final class SSHMultiplexedSessionWorker: @unchecked Sendable { } catch { await disconnect() logger.error( - "ssh.exec.connect.failed", + "ssh.multiplexed.connect.failed", metadata: [ "error_code": RemoteLogging.errorCode(error), "host_id": RemoteLogging.hostID(hostID), "operation_id": "\(operationID)", "outcome": "failure", "protocol": "ssh", @@ -171,7 +192,7 @@ nonisolated final class SSHMultiplexedSessionWorker: @unchecked Sendable { queue.async { [self] in guard session != nil else { continuation.resume( - throwing: SSHExecSessionError.transportUnavailable(TermindSSHError.disconnected)) + throwing: SSHCommandExecutionError.transportUnavailable(TermindSSHError.disconnected)) return } let context = Command( @@ -196,118 +217,7 @@ nonisolated final class SSHMultiplexedSessionWorker: @unchecked Sendable { } func openSFTPChannel() async throws -> any SFTPClientProtocol { - let driver = SSHSharedSFTPDriver( - identity: sharedSFTPIdentity, - open: { [weak self] in - guard let self else { throw SFTPError.notConnected } - return try await self.openRawSFTPChannel() - }, - close: { [weak self] id in await self?.closeSFTPChannel(id) }, - perform: { [weak self] id, operation in - guard let self else { throw SFTPError.notConnected } - return try await self.performSharedSFTP(id, operation) - }) - return try await SSHSharedSFTPClient.open(using: driver) - } - - private func openRawSFTPChannel() async throws -> UUID { - await reserveSFTPOpen() - defer { releaseSFTPOpen() } - let id = UUID() - while true { - try Task.checkCancellation() - let attempt: SSHSharedSFTPAttempt = try await queue.perform { [self] in - guard let session, let bufferedTransport, let routeTransport else { - throw SFTPError.notConnected - } - let deadline = DispatchTime.now().uptimeNanoseconds + SSHForwardingTuning.current.sliceNanoseconds - bufferedTransport.beginPass(deadline: deadline) - try bufferedTransport.flush(socket: routeTransport.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 SSHSessionWorker.sessionError(session, prefix: "Unable to start SFTP") - } - sftpChannels[id] = sftp - try bufferedTransport.flush(socket: routeTransport.socket) - updateSharedTransportEvents(session: session, transport: bufferedTransport) - return .value(()) - } - if case .value = attempt { return id } - 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() - } - } - } - - private func closeSFTPChannel(_ id: UUID) async { - while true { - let attempt: SSHSharedSFTPAttempt = - (try? await queue.perform { [self] in - guard let sftp = sftpChannels[id] else { return .value(()) } - guard let session, let bufferedTransport, let routeTransport else { - sftpChannels[id] = nil - return .value(()) - } - let deadline = DispatchTime.now().uptimeNanoseconds + SSHForwardingTuning.current.sliceNanoseconds - bufferedTransport.beginPass(deadline: deadline) - try bufferedTransport.flush(socket: routeTransport.socket) - let result = libssh2_sftp_shutdown(sftp) - if result == LIBSSH2_ERROR_EAGAIN { return .again } - sftpChannels[id] = nil - try bufferedTransport.flush(socket: routeTransport.socket) - updateSharedTransportEvents(session: session, transport: bufferedTransport) - return .value(()) - }) ?? .value(()) - if case .value = attempt { return } - try? await Task.sleep(for: .milliseconds(2)) - } - } - - private func performSharedSFTP( - _ id: UUID, _ operation: @escaping SSHSharedSFTPDriver.RawOperation - ) async throws -> SSHSharedSFTPValue { - while true { - try Task.checkCancellation() - let attempt = try await queue.perform { [self] in - guard let session, let sftp = sftpChannels[id], let bufferedTransport, let routeTransport else { - throw SFTPError.notConnected - } - let deadline = DispatchTime.now().uptimeNanoseconds + SSHForwardingTuning.current.sliceNanoseconds - bufferedTransport.beginPass(deadline: deadline) - try bufferedTransport.flush(socket: routeTransport.socket) - if bufferedTransport.isBackpressured { return SSHSharedSFTPAnyAttempt.again } - let result = try operation(session, sftp) - try bufferedTransport.flush(socket: routeTransport.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)) - } - } + try await sftpChannelOwner.openClient() } private func updateSharedTransportEvents(session: OpaquePointer, transport: SSHBufferedTransport) { @@ -447,7 +357,7 @@ nonisolated final class SSHMultiplexedSessionWorker: @unchecked Sendable { if isTransportError(code) { failTransport(session, error: error) } else { - finish(command, throwing: SSHExecSessionError.channelUnavailable(error)) + finish(command, throwing: SSHCommandExecutionError.channelUnavailable(error)) } return true } @@ -481,7 +391,7 @@ nonisolated final class SSHMultiplexedSessionWorker: @unchecked Sendable { guard result == 0 else { let code = libssh2_session_last_errno(session) let error = SSHSessionWorker.sessionError(session, prefix: "Unable to start remote command") - command.terminalError = SSHExecSessionError.channelUnavailable(error) + command.terminalError = SSHCommandExecutionError.channelUnavailable(error) command.phase = .closing if isTransportError(code) { failTransport(session, error: error) } ready.insert(command.id) @@ -657,6 +567,10 @@ nonisolated final class SSHMultiplexedSessionWorker: @unchecked Sendable { private func failTransport(_ session: OpaquePointer, error: (any Error)? = nil) { let failure = error ?? SSHSessionWorker.sessionError(session, prefix: "The SSH connection closed") + if !didReportConnectionFailure { + didReportConnectionFailure = true + connectionFailure?(failure.localizedDescription) + } cleanup(error: failure) } @@ -670,8 +584,7 @@ nonisolated final class SSHMultiplexedSessionWorker: @unchecked Sendable { opening = nil pendingOpens = SSHReadyQueue() ready = SSHReadyQueue() - for sftp in sftpChannels.values { _ = libssh2_sftp_shutdown(sftp) } - sftpChannels.removeAll() + sftpChannelOwner.closeAllOnQueue() let transport = routeTransport routeTransport = nil if let session { @@ -683,6 +596,8 @@ nonisolated final class SSHMultiplexedSessionWorker: @unchecked Sendable { } session = nil bufferedTransport = nil + connectionFailure = nil + didReportConnectionFailure = false hostID = nil if let events = socketEvents { events.cancel { transport?.close() } diff --git a/Frameworks/TermindRemoteInfrastructure/SSH/SSHSessionActor.swift b/Frameworks/TermindRemoteInfrastructure/SSH/SSHSessionActor.swift index 7dc566c9..d7702e94 100644 --- a/Frameworks/TermindRemoteInfrastructure/SSH/SSHSessionActor.swift +++ b/Frameworks/TermindRemoteInfrastructure/SSH/SSHSessionActor.swift @@ -677,10 +677,7 @@ 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] = [] - private let sharedSFTPIdentity = UUID() + private var sftpChannelOwner: SSHMultiplexedSFTPChannelOwner! public init( output: @escaping SSHSessionActor.OutputHandler, @@ -693,6 +690,24 @@ public nonisolated final class SSHSessionWorker: @unchecked Sendable { self.exit = exit self.shellIntegrationHandler = shellIntegrationHandler self.logger = logger + sftpChannelOwner = SSHMultiplexedSFTPChannelOwner( + queue: queue, + context: { [weak self] in + guard let self, let session = self.session, + let bufferedTransport = self.bufferedTransport, + self.socket >= 0 + else { return nil } + return .init( + session: session, transport: bufferedTransport, + socket: self.socket) + }, + updateEvents: { [weak self] session, transport in + self?.updateSharedTransportEvents(session: session, transport: transport) + }, + didCloseLastChannel: { [weak self] in + guard let self, self.channel == nil else { return } + self.cleanup() + }) } /// Opens a transport, handshakes, validates the host key, and authenticates — @@ -1367,7 +1382,7 @@ public nonisolated final class SSHSessionWorker: @unchecked Sendable { _ = libssh2_channel_free(channel) self.channel = nil } - if sftpChannels.isEmpty { cleanup() } + if sftpChannelOwner.isEmptyOnQueue { cleanup() } } private func cleanup() { @@ -1386,8 +1401,7 @@ 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() + sftpChannelOwner.closeAllOnQueue() if let session { if let bufferedTransport { bufferedTransport.dispose(session: session) @@ -1536,137 +1550,7 @@ public nonisolated final class SSHSessionWorker: @unchecked Sendable { nonisolated extension SSHSessionWorker { fileprivate func openSFTPChannel() async throws -> any SFTPClientProtocol { - let driver = SSHSharedSFTPDriver( - identity: sharedSFTPIdentity, - open: { [weak self] in - guard let self else { throw SFTPError.notConnected } - return try await self.openRawSFTPChannel() - }, - close: { [weak self] id in await self?.closeSFTPChannel(id) }, - perform: { [weak self] id, operation in - guard let self else { throw SFTPError.notConnected } - return try await self.performSharedSFTP(id, operation) - }) - return try await SSHSharedSFTPClient.open(using: driver) - } - - private func openRawSFTPChannel() async throws -> UUID { - await reserveSFTPOpen() - defer { releaseSFTPOpen() } - let id = UUID() - while true { - try Task.checkCancellation() - let attempt: SSHSharedSFTPAttempt = 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 id - } catch { - await closeSFTPChannel(id) - throw error - } - } - try await Task.sleep(for: .milliseconds(2)) - } - } - - private func performSharedSFTP( - _ id: UUID, _ operation: @escaping SSHSharedSFTPDriver.RawOperation - ) async throws -> SSHSharedSFTPValue { - try await performSFTP(id) { session, sftp in - switch try operation(session, sftp) { - case .value(let value): .value(value) - case .again: .again - } - } - } - - 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: SSHSharedSFTPAttempt = - (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 -> SSHSharedSFTPAttempt - ) 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 SSHSharedSFTPAttempt.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)) - } - } + try await sftpChannelOwner.openClient() } private func updateSharedTransportEvents(session: OpaquePointer, transport: SSHBufferedTransport) { diff --git a/Memory/ARCHITECTURE.md b/Memory/ARCHITECTURE.md index a3d8414c..72555c58 100644 --- a/Memory/ARCHITECTURE.md +++ b/Memory/ARCHITECTURE.md @@ -166,6 +166,10 @@ AgentKit is a separately versioned remote package; its sibling working tree is n - Domain owns `RemoteFile`, Application `RemoteFileSystem`/`MobileSFTPSessionModel`, Remote Infrastructure libssh2 protocols, Presentation symbols/kinds. Device-local History changes only after fresh successful listings/downloads, never stale/failed/ cancelled/type-changed work. +- `SFTPClient` is the connect/disconnect facade and owns its authenticated multiplexed SSH session; `SFTPChannel` is the sole + file-operation implementation. Standalone SFTP, Agent primary SFTP and terminal fallback therefore + share Channel open/close/EAGAIN, atomic-write, resume, edit, streaming and copy behavior; feature owners retain their distinct + authentication, pooling and retirement lifetimes. - 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 diff --git a/Memory/REFACTOR.md b/Memory/REFACTOR.md index 10438130..c22a99ae 100644 --- a/Memory/REFACTOR.md +++ b/Memory/REFACTOR.md @@ -208,3 +208,19 @@ never justifies extraction. Keep schemas, native lists/terminal teardown and loc Host retirement, route changes and fatal primary failures close the shared lifetime atomically, without replaying operations. - **Verification:** focused pool/authentication/copy tests and macOS build passed during implementation; final cross-platform, localization and format status is recorded in the task handoff. + +### 2026-09-20 — one SFTP Channel implementation (complete) + +- **Evidence:** the standalone `SFTPWorker` and multiplexed `SSHSharedSFTPClient` separately implemented directory, atomic write, + resume, edit, streaming and copy semantics, while the Agent and terminal workers duplicated SFTP Channel ownership and EAGAIN + loops. The new type name also described reuse (`Shared`) rather than responsibility. +- **Action:** keep `SFTPClient` as the public connection facade and `SSHMultiplexedSession` owner, move all file operations to + `SFTPChannel`, and share `SSHMultiplexedSFTPChannelOwner` across standalone, Agent and terminal transports. Rename + Agent pool APIs from exec-only terms to `session`, `connection` and `sftpClient`. +- **Result:** removed the second SFTP operation engine and both feature-local Channel owners. One implementation now preserves + atomic rename, partial-file verification, edit permissions/version capture, streamed download validation and same-transport copy + for every entry point. Tracked production plus the two new files changed +1,523/−2,695 lines (net −1,172). +- **Verification:** macOS and iOS builds, 87 focused pool/auth/edit/resume/rename/copy tests, 14 opt-in loopback SSH/SFTP wire + tests and all 363 Framework tests passed. The 1,465-test TFM run retained two unrelated concurrent timing failures in + `AgentRuntimeTests`/`MCPUploadTests`; both passed together on immediate focused rerun. Changed-file strict formatting and Chinese + localization checks passed; recursive formatting retains the pre-existing `LoggingFilesSettings.swift:110` finding. diff --git a/TFMTests/AgentConnectionPoolTests.swift b/TFMTests/AgentConnectionPoolTests.swift index af8d1c54..07a87bef 100644 --- a/TFMTests/AgentConnectionPoolTests.swift +++ b/TFMTests/AgentConnectionPoolTests.swift @@ -66,11 +66,11 @@ struct AgentConnectionPoolTests { let pool = makePool(backend, openerGate: gate) let host = UUID() - async let first = pool.exec(hostID: host) + async let first = pool.session(hostID: host) // Only start the second caller once the first is parked mid-connect, so // it is guaranteed to find work already in flight rather than racing it. while await gate.arrivals == 0 { await Task.yield() } - async let second = pool.exec(hostID: host) + async let second = pool.session(hostID: host) await gate.open() let sessions = try await [first, second] @@ -84,8 +84,8 @@ struct AgentConnectionPoolTests { let pool = makePool(backend) let host = UUID() - _ = try await pool.exec(hostID: host) - _ = try await pool.sftp(hostID: host) + _ = try await pool.session(hostID: host) + _ = try await pool.sftpClient(hostID: host) #expect(await pool.openConnectionCount == 1) #expect(await backend.execOpens == 1) #expect(await backend.sftpOpens == 0) @@ -102,8 +102,8 @@ struct AgentConnectionPoolTests { let pool = makePool(backend) let host = UUID() - _ = try await pool.sftp(hostID: host) - _ = try await pool.exec(hostID: host) + _ = try await pool.sftpClient(hostID: host) + _ = try await pool.session(hostID: host) #expect(await backend.execOpens == 1) #expect(await backend.sftpOpens == 0) @@ -117,9 +117,9 @@ struct AgentConnectionPoolTests { let pool = makePool(backend, openerGate: gate) let host = UUID() - async let exec = pool.exec(hostID: host) + async let exec = pool.session(hostID: host) while await gate.arrivals == 0 { await Task.yield() } - async let sftp = pool.sftp(hostID: host) + async let sftp = pool.sftpClient(hostID: host) await gate.open() _ = try await (exec, sftp) @@ -134,7 +134,7 @@ struct AgentConnectionPoolTests { let pool = makePool(backend) let host = UUID() - _ = try await pool.exec(hostID: host) + _ = try await pool.session(hostID: host) #expect(await pool.disconnect(hostID: host)) #expect(await pool.disconnect(hostID: host) == false) #expect(await backend.execDisconnects == 1) @@ -146,7 +146,7 @@ struct AgentConnectionPoolTests { let pool = makePool(backend) let host = UUID() - _ = try await pool.exec(hostID: host) + _ = try await pool.session(hostID: host) let remaining = await pool.sweepIdle( now: .now.advanced(by: AgentConnectionPool.defaultIdleTimeout + .seconds(1)) ) @@ -155,7 +155,7 @@ struct AgentConnectionPoolTests { #expect(await backend.execDisconnects == 1) #expect(await pool.isConnected(hostID: host) == false) - _ = try await pool.exec(hostID: host) + _ = try await pool.session(hostID: host) #expect(await backend.execOpens == 2) } @@ -165,7 +165,7 @@ struct AgentConnectionPoolTests { let pool = makePool(backend) let host = UUID() - _ = try await pool.exec(hostID: host) + _ = try await pool.session(hostID: host) let remaining = await pool.sweepIdle(now: .now) #expect(remaining == 1) @@ -177,7 +177,7 @@ struct AgentConnectionPoolTests { let backend = StubBackend() let pool = makePool(backend) let host = UUID() - let session = try await pool.exec(hostID: host) + let session = try await pool.session(hostID: host) let gate = Gate() let running = Task { try await pool.withExecLease(hostID: host, session: session) { @@ -202,9 +202,9 @@ struct AgentConnectionPoolTests { let pool = makePool(backend) let host = UUID() - let before = try await pool.exec(hostID: host) + let before = try await pool.session(hostID: host) await backend.setFingerprint("route-b", for: host) - let after = try await pool.exec(hostID: host) + let after = try await pool.session(hostID: host) // Not the same machine any more, so not the same session — otherwise a // command aimed at the edited host would run on the old one. @@ -220,10 +220,10 @@ struct AgentConnectionPoolTests { let pool = makePool(backend) let host = UUID() - _ = try await pool.exec(hostID: host) + _ = try await pool.session(hostID: host) await backend.forget(host) - await #expect(throws: AgentToolError.self) { _ = try await pool.exec(hostID: host) } + await #expect(throws: AgentToolError.self) { _ = try await pool.session(hostID: host) } #expect(await backend.execDisconnects == 1) #expect(await pool.openConnectionCount == 0) } @@ -235,7 +235,7 @@ struct AgentConnectionPoolTests { let host = UUID() await backend.queueExecFailures([ - SSHExecSessionError.transportUnavailable(TermindSSHError.disconnected) + SSHCommandExecutionError.transportUnavailable(TermindSSHError.disconnected) ]) let result = try await pool.execute( "uptime", workingDirectory: nil, stdin: nil, @@ -274,8 +274,8 @@ struct AgentConnectionPoolTests { for _ in 0..<3 { let host = UUID() - _ = try await pool.exec(hostID: host) - _ = try await pool.sftp(hostID: host) + _ = try await pool.session(hostID: host) + _ = try await pool.sftpClient(hostID: host) } await pool.closeAll() @@ -288,7 +288,7 @@ struct AgentConnectionPoolTests { /// Which stub connection the pool handed back. Distinguishing "reused" from /// "reconnected" is the whole subject of these tests. private func serial(_ session: any SSHMultiplexedSessionProtocol) throws -> Int { - try #require(session as? StubExecSession).stubSerial + try #require(session as? StubMultiplexedSession).stubSerial } /// `sweepInterval: nil` keeps the background sweeper out of the tests; every @@ -339,7 +339,7 @@ private actor StubBackend { private var fingerprints: [UUID: String] = [:] private var forgotten: Set = [] private var nextExecFailures: [any Error] = [] - private var execSessions: [StubExecSession] = [] + private var sessions: [StubMultiplexedSession] = [] func fingerprint(for hostID: UUID) throws -> String { guard !forgotten.contains(hostID) else { throw AgentToolError.invalidHost } @@ -353,13 +353,13 @@ private actor StubBackend { /// connection to fail once and the replacement to succeed. func queueExecFailures(_ errors: [any Error]) { nextExecFailures = errors } - func openSession() -> StubExecSession { + func openSession() -> StubMultiplexedSession { execOpens += 1 - let session = StubExecSession( + let session = StubMultiplexedSession( serial: execOpens, backend: self, failures: nextExecFailures ) nextExecFailures = [] - execSessions.append(session) + sessions.append(session) return session } @@ -375,12 +375,12 @@ private actor StubBackend { func totalExecuteCount() async -> Int { var total = 0 - for session in execSessions { total += await session.executeCount } + for session in sessions { total += await session.executeCount } return total } } -private actor StubExecSession: SSHMultiplexedSessionProtocol { +private actor StubMultiplexedSession: SSHMultiplexedSessionProtocol { nonisolated let stubSerial: Int private let backend: StubBackend private var failures: [any Error] diff --git a/TFMTests/AgentHostAuthenticationTests.swift b/TFMTests/AgentHostAuthenticationTests.swift index b6260633..879f1fad 100644 --- a/TFMTests/AgentHostAuthenticationTests.swift +++ b/TFMTests/AgentHostAuthenticationTests.swift @@ -311,7 +311,7 @@ struct AgentHostAuthenticationTests { )) } - _ = try await registry.sftp( + _ = try await registry.sftpClient( hostID: destination.id, invocationID: UUID(), interaction: interaction.sshInteraction, pool: pool ) @@ -390,11 +390,11 @@ struct AgentHostAuthenticationTests { let setup = authenticationSetup() let interaction = GatedAuthenticationResponder() - async let first = setup.registry.sftp( + async let first = setup.registry.sftpClient( hostID: setup.host.id, invocationID: UUID(), interaction: interaction.sshInteraction, pool: setup.pool ) - async let second = setup.registry.sftp( + async let second = setup.registry.sftpClient( hostID: setup.host.id, invocationID: UUID(), interaction: interaction.sshInteraction, pool: setup.pool ) @@ -416,11 +416,11 @@ struct AgentHostAuthenticationTests { let setup = authenticationSetup() let interaction = GatedAuthenticationResponder() - async let exec = setup.registry.exec( + async let exec = setup.registry.session( hostID: setup.host.id, invocationID: UUID(), interaction: interaction.sshInteraction, pool: setup.pool ) - async let sftp = setup.registry.sftp( + async let sftp = setup.registry.sftpClient( hostID: setup.host.id, invocationID: UUID(), interaction: interaction.sshInteraction, pool: setup.pool ) @@ -444,13 +444,13 @@ struct AgentHostAuthenticationTests { let setup = authenticationSetup() let interaction = GatedAuthenticationResponder() let first = Task { - try await setup.registry.sftp( + try await setup.registry.sftpClient( hostID: setup.host.id, invocationID: UUID(), interaction: interaction.sshInteraction, pool: setup.pool ) } let second = Task { - try await setup.registry.sftp( + try await setup.registry.sftpClient( hostID: setup.host.id, invocationID: UUID(), interaction: interaction.sshInteraction, pool: setup.pool ) @@ -479,7 +479,7 @@ struct AgentHostAuthenticationTests { )) } - _ = try await setup.registry.sftp( + _ = try await setup.registry.sftpClient( hostID: setup.host.id, invocationID: UUID(), interaction: interaction.sshInteraction, pool: setup.pool ) @@ -1234,14 +1234,14 @@ private actor AuthenticationConnectionBackend { self.remoteFiles = remoteFiles } - func openSession(configuration: SSHConnectionConfiguration) throws -> AuthenticationExecSession { + func openSession(configuration: SSHConnectionConfiguration) throws -> AuthenticationMultiplexedSession { execOpens += 1 sftpConfigurations.append(configuration) if sftpAuthenticationFailures > 0 { sftpAuthenticationFailures -= 1 throw TermindSSHError.authenticationFailed("Rejected credential") } - return AuthenticationExecSession(backend: self) + return AuthenticationMultiplexedSession(backend: self) } func openSFTP( @@ -1286,7 +1286,7 @@ private actor AuthenticationConnectionBackend { func remoteFile(at path: String) -> Data? { remoteFiles[path] } } -private actor AuthenticationExecSession: SSHMultiplexedSessionProtocol { +private actor AuthenticationMultiplexedSession: SSHMultiplexedSessionProtocol { private let backend: AuthenticationConnectionBackend init(backend: AuthenticationConnectionBackend) { self.backend = backend } diff --git a/TFMTests/SSHKeyboardInteractiveTests.swift b/TFMTests/SSHKeyboardInteractiveTests.swift index b460cd34..510ed980 100644 --- a/TFMTests/SSHKeyboardInteractiveTests.swift +++ b/TFMTests/SSHKeyboardInteractiveTests.swift @@ -46,6 +46,15 @@ struct SSHKeyboardInteractiveWireTests { do { let files = try await client.listDirectory(at: "/") #expect(files.map(\.name) == ["fixture.txt"]) + let localURL = FileManager.default.temporaryDirectory + .appending(path: "termind-sftp-version-\(UUID().uuidString.lowercased())") + defer { try? FileManager.default.removeItem(at: localURL) } + let resume = SFTPResumeState(capturesFileVersion: true) + try await client.downloadFile( + from: "/fixture.txt", to: localURL, overwrite: false, + cancellation: TransferCancellationToken(), resume: resume, + progress: { _ in }) + #expect(resume.snapshot.committedVersion != nil) #expect(await recorder.challenges.count == 1) } catch { await client.disconnect()