okooooooo/Termind
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Unify SFTP channel implementation

Browse Source
This commit is contained in:
Kaede Akino committed 2026-09-20 00:34:30 +08:00
1 parent 15d2eec23a
commit 8fdf2eb0d3
16 files changed
+679 -1821
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

FrameworkTests/LaunchFileLogServiceTests.swift
+3 -2
View File
@@ -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 {}
}
Frameworks/TermindRemoteInfrastructure/Agent/AgentBuiltInSSHTools.swift
+4 -4
View File
@@ -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
)
Frameworks/TermindRemoteInfrastructure/Agent/AgentConnections.swift
+23 -21
View File
@@ -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(
Frameworks/TermindRemoteInfrastructure/Agent/AgentHostServices.swift
+1 -1
View File
@@ -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
),
Frameworks/TermindRemoteInfrastructure/SSH/RemoteConnectionCoordinator.swift
+3 -3
View File
@@ -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()
Frameworks/TermindRemoteInfrastructure/SSH/SSHSharedSFTPClient.swift → Frameworks/TermindRemoteInfrastructure/SSH/SFTPChannel.swift
+177 -68
View File
@@ -6,37 +6,37 @@ import Foundation
import TermindApplication
import TermindDomain
nonisolated enum SSHSharedSFTPAttempt<Value: Sendable>: Sendable {
nonisolated enum SFTPChannelAttempt<Value: Sendable>: 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<Value: Sendable>(
_ id: UUID,
_ operation: @escaping @Sendable (OpaquePointer, OpaquePointer) throws -> SSHSharedSFTPAttempt<Value>
_ operation: @escaping @Sendable (OpaquePointer, OpaquePointer) throws -> SFTPChannelAttempt<Value>
) 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
Frameworks/TermindRemoteInfrastructure/SSH/SFTPClient.swift
+147 -1411
View File
File diff suppressed because it is too large. Load diff
Frameworks/TermindRemoteInfrastructure/SSH/SSHMultiplexedSFTPChannelOwner.swift
+176
View File
@@ -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<Void, Never>] = []
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<Void> = 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<Void> =
(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))
}
}
}
}
Frameworks/TermindRemoteInfrastructure/SSH/SSHMultiplexedSession.swift
+13 -7
View File
@@ -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(
Frameworks/TermindRemoteInfrastructure/SSH/SSHMultiplexedSessionWorker.swift
+43 -128
View File
@@ -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<Void, Never>] = []
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<Void> = 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<Void> =
(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() }
Frameworks/TermindRemoteInfrastructure/SSH/SSHSessionActor.swift
+22 -138
View File
@@ -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<Void, Never>] = []
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<Void> = 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<Void> =
(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<Value: Sendable>(
_ id: UUID,
_ operation: @escaping @Sendable (OpaquePointer, OpaquePointer) throws -> SSHSharedSFTPAttempt<Value>
) 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<Value>.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) {
Memory/ARCHITECTURE.md
+4
View File
@@ -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
Memory/REFACTOR.md
+16
View File
@@ -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.
TFMTests/AgentConnectionPoolTests.swift
+27 -27
View File
@@ -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<UUID> = []
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]
TFMTests/AgentHostAuthenticationTests.swift
+11 -11
View File
@@ -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 }
TFMTests/SSHKeyboardInteractiveTests.swift
+9
View File
@@ -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()