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