252 lines
11 KiB
Swift
252 lines
11 KiB
Swift
// SPDX-FileCopyrightText: 2026 Kaede Akino
|
|
// SPDX-License-Identifier: GPL-3.0-only
|
|
|
|
import Darwin
|
|
import Foundation
|
|
import TermindApplication
|
|
import TermindDomain
|
|
import Testing
|
|
|
|
@testable import TermindRemoteInfrastructure
|
|
|
|
@Suite
|
|
struct TerminalTCPBridgeTests {
|
|
@Test func boundedBridgePreservesBytesAndTheFinalResponseAfterHalfClose() async throws {
|
|
var pair: [Int32] = [-1, -1]
|
|
#expect(socketpair(AF_UNIX, SOCK_STREAM, 0, &pair) == 0)
|
|
let peer = pair[1]
|
|
defer { Darwin.close(peer) }
|
|
configure(peer)
|
|
let route = try await routed(pair[0])
|
|
let bridge = try TerminalTCPBridge(open: { _ in throw CancellationError() }, initial: route)
|
|
let client = try connect(to: bridge.port)
|
|
defer { Darwin.close(client) }
|
|
let payload = Data((0..<131_072).map { UInt8($0 % 251) })
|
|
async let received = read(peer, count: payload.count)
|
|
try await write(client, payload)
|
|
#expect(try await received == payload)
|
|
#expect(Darwin.shutdown(client, SHUT_WR) == 0)
|
|
let response = Data("final response after request EOF".utf8)
|
|
try await write(peer, response)
|
|
#expect(Darwin.shutdown(peer, SHUT_WR) == 0)
|
|
#expect(try await read(client, count: response.count) == response)
|
|
await bridge.close()
|
|
#expect(fcntl(route.socket, F_GETFD) == -1)
|
|
}
|
|
|
|
@Test func replacementConnectionGetsANewRouteAndCloseReleasesBoth() async throws {
|
|
var first: [Int32] = [-1, -1]
|
|
var second: [Int32] = [-1, -1]
|
|
#expect(socketpair(AF_UNIX, SOCK_STREAM, 0, &first) == 0)
|
|
#expect(socketpair(AF_UNIX, SOCK_STREAM, 0, &second) == 0)
|
|
defer {
|
|
Darwin.close(first[1])
|
|
Darwin.close(second[1])
|
|
}
|
|
configure(first[1])
|
|
configure(second[1])
|
|
let initial = try await routed(first[0])
|
|
let replacement = try await routed(second[0])
|
|
let snapshots = BridgeSnapshots()
|
|
let reporter = TerminalRouteReporter(hostIDs: [UUID()], changed: snapshots.record)
|
|
let initialAttempt = reporter.begin()
|
|
reporter.ready(attemptID: initialAttempt)
|
|
let bridge = try TerminalTCPBridge(
|
|
open: { _ in replacement }, initial: initial, reporter: reporter, initialAttemptID: initialAttempt)
|
|
let oldClient = try connect(to: bridge.port)
|
|
try await write(oldClient, Data([1]))
|
|
#expect(try await read(first[1], count: 1) == Data([1]))
|
|
#expect(snapshots.values.map(\.stage) == [.opening, .ready])
|
|
Darwin.close(oldClient)
|
|
let newClient = try connect(to: bridge.port)
|
|
defer { Darwin.close(newClient) }
|
|
try await write(newClient, Data([2]))
|
|
#expect(try await read(second[1], count: 1) == Data([2]))
|
|
await bridge.close()
|
|
#expect(fcntl(initial.socket, F_GETFD) == -1)
|
|
#expect(fcntl(replacement.socket, F_GETFD) == -1)
|
|
let reported = snapshots.values
|
|
#expect(reported.map(\.stage) == [.opening, .ready, .unavailable, .opening, .ready, .unavailable])
|
|
#expect(reported[2].attemptID == initialAttempt)
|
|
#expect(reported[3].attemptID != initialAttempt)
|
|
#expect(reported[3].establishedHostIDs.isEmpty)
|
|
#expect(reported.map(\.revision) == Array(1...UInt64(reported.count)))
|
|
}
|
|
|
|
@Test func closeDuringOpeningRejectsLateProgressAndReleasesTheLateSocket() async throws {
|
|
var pair: [Int32] = [-1, -1]
|
|
#expect(socketpair(AF_UNIX, SOCK_STREAM, 0, &pair) == 0)
|
|
defer { Darwin.close(pair[1]) }
|
|
let lateRoute = try await routed(pair[0])
|
|
let gate = OpeningGate()
|
|
let snapshots = BridgeSnapshots()
|
|
let hop = UUID()
|
|
let reporter = TerminalRouteReporter(hostIDs: [hop, UUID()], changed: snapshots.record)
|
|
let bridge = try TerminalTCPBridge(open: { try await gate.open($0) }, reporter: reporter)
|
|
let client = try connect(to: bridge.port)
|
|
defer { Darwin.close(client) }
|
|
#expect(await eventually { await gate.isOpening })
|
|
await gate.emit(.authenticated(hop))
|
|
#expect(await eventually { snapshots.values.last?.establishedHostIDs == [hop] })
|
|
await bridge.close()
|
|
let retired = snapshots.values
|
|
#expect(retired.last?.stage == .unavailable)
|
|
#expect(retired.last?.establishedHostIDs == [])
|
|
await gate.emit(.connected(hop))
|
|
await gate.finish(lateRoute)
|
|
#expect(await eventually { fcntl(lateRoute.socket, F_GETFD) == -1 })
|
|
#expect(snapshots.values == retired)
|
|
}
|
|
|
|
@Test func blockedOpeningRemainsBlockedUntilExplicitRetry() async throws {
|
|
var pair: [Int32] = [-1, -1]
|
|
#expect(socketpair(AF_UNIX, SOCK_STREAM, 0, &pair) == 0)
|
|
defer { Darwin.close(pair[1]) }
|
|
configure(pair[1])
|
|
let route = try await routed(pair[0])
|
|
let gate = OpeningGate()
|
|
let snapshots = BridgeSnapshots()
|
|
let reporter = TerminalRouteReporter(hostIDs: [UUID()], changed: snapshots.record)
|
|
let bridge = try TerminalTCPBridge(open: { try await gate.open($0) }, reporter: reporter)
|
|
let first = try connect(to: bridge.port)
|
|
defer { Darwin.close(first) }
|
|
#expect(await eventually { await gate.isOpening })
|
|
await gate.fail()
|
|
#expect(await eventually { snapshots.values.last?.stage == .unavailable })
|
|
let blocked = try connect(to: bridge.port)
|
|
defer { Darwin.close(blocked) }
|
|
#expect(
|
|
await eventually {
|
|
var byte: UInt8 = 0
|
|
return Darwin.recv(blocked, &byte, 1, 0) == 0
|
|
})
|
|
#expect(await gate.openCount == 1)
|
|
await bridge.retry()
|
|
let retried = try connect(to: bridge.port)
|
|
defer { Darwin.close(retried) }
|
|
#expect(await eventually { await gate.openCount == 2 })
|
|
await gate.finish(route)
|
|
try await write(retried, Data([7]))
|
|
#expect(try await read(pair[1], count: 1) == Data([7]))
|
|
#expect(snapshots.values.map(\.stage) == [.opening, .unavailable, .opening, .ready])
|
|
await bridge.close()
|
|
}
|
|
|
|
private func eventually(_ predicate: () async -> Bool) async -> Bool {
|
|
for _ in 0..<200 {
|
|
if await predicate() { return true }
|
|
try? await Task.sleep(for: .milliseconds(10))
|
|
}
|
|
return await predicate()
|
|
}
|
|
|
|
private nonisolated func routed(_ socket: Int32) async throws -> SSHRoutedSocket {
|
|
try await SSHRoutedSocket.open(
|
|
host: "routed.invalid", port: 2022, proxy: nil, jumps: [], timeout: 2,
|
|
authorizeHostKey: nil, network: Network(socket: socket))
|
|
}
|
|
|
|
private nonisolated struct Network: SSHNetworkDialing {
|
|
let socket: Int32
|
|
func open(host: String, port: Int, timeout: TimeInterval) async throws -> Int32 { socket }
|
|
func hostKeys(stableNodeID: String) async throws -> Set<SSHHostKeyIdentity> { [] }
|
|
}
|
|
|
|
private nonisolated func connect(to port: UInt16) throws -> Int32 {
|
|
let socket = Darwin.socket(AF_INET, SOCK_STREAM, 0)
|
|
guard socket >= 0 else { throw POSIXError(.ENOBUFS) }
|
|
var address = sockaddr_in()
|
|
address.sin_len = UInt8(MemoryLayout<sockaddr_in>.size)
|
|
address.sin_family = sa_family_t(AF_INET)
|
|
address.sin_addr.s_addr = inet_addr("127.0.0.1")
|
|
address.sin_port = port.bigEndian
|
|
let result = withUnsafePointer(to: &address) {
|
|
$0.withMemoryRebound(to: sockaddr.self, capacity: 1) {
|
|
Darwin.connect(socket, $0, socklen_t(MemoryLayout<sockaddr_in>.size))
|
|
}
|
|
}
|
|
guard result == 0 else {
|
|
Darwin.close(socket)
|
|
throw POSIXError(.ECONNREFUSED)
|
|
}
|
|
configure(socket)
|
|
return socket
|
|
}
|
|
|
|
private nonisolated func configure(_ socket: Int32) {
|
|
_ = fcntl(socket, F_SETFL, fcntl(socket, F_GETFL) | O_NONBLOCK)
|
|
var one: Int32 = 1
|
|
_ = setsockopt(socket, SOL_SOCKET, SO_NOSIGPIPE, &one, socklen_t(MemoryLayout<Int32>.size))
|
|
}
|
|
|
|
private nonisolated func read(_ socket: Int32, count: Int) async throws -> Data {
|
|
var result = Data()
|
|
let deadline = ContinuousClock.now.advanced(by: .seconds(5))
|
|
while result.count < count {
|
|
var bytes = [UInt8](repeating: 0, count: min(4096, count - result.count))
|
|
let received = Darwin.recv(socket, &bytes, bytes.count, 0)
|
|
if received > 0 {
|
|
result.append(contentsOf: bytes.prefix(received))
|
|
} else if received == 0 {
|
|
throw POSIXError(.ECONNRESET)
|
|
} else {
|
|
guard errno == EWOULDBLOCK || errno == EAGAIN || errno == EINTR else {
|
|
throw POSIXError(.EIO)
|
|
}
|
|
guard ContinuousClock.now < deadline else { throw POSIXError(.ETIMEDOUT) }
|
|
try await Task.sleep(for: .milliseconds(5))
|
|
}
|
|
}
|
|
return result
|
|
}
|
|
|
|
private nonisolated func write(_ socket: Int32, _ data: Data) async throws {
|
|
var remaining = data
|
|
let deadline = ContinuousClock.now.advanced(by: .seconds(5))
|
|
while !remaining.isEmpty {
|
|
let sent = remaining.withUnsafeBytes { Darwin.send(socket, $0.baseAddress, $0.count, 0) }
|
|
if sent > 0 {
|
|
remaining.removeFirst(sent)
|
|
} else {
|
|
guard errno == EWOULDBLOCK || errno == EAGAIN || errno == EINTR else {
|
|
throw POSIXError(.EIO)
|
|
}
|
|
guard ContinuousClock.now < deadline else { throw POSIXError(.ETIMEDOUT) }
|
|
try await Task.sleep(for: .milliseconds(5))
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
private nonisolated final class BridgeSnapshots: @unchecked Sendable {
|
|
private let lock = NSLock()
|
|
private var recorded: [TerminalRouteSnapshot] = []
|
|
var values: [TerminalRouteSnapshot] { lock.withLock { recorded } }
|
|
func record(_ snapshot: TerminalRouteSnapshot) { lock.withLock { recorded.append(snapshot) } }
|
|
}
|
|
|
|
private actor OpeningGate {
|
|
private var continuation: CheckedContinuation<SSHRoutedSocket, any Error>?
|
|
private var progress: SSHRouteProgressHandler?
|
|
private(set) var openCount = 0
|
|
var isOpening: Bool { continuation != nil }
|
|
|
|
func open(_ progress: SSHRouteProgressHandler?) async throws -> SSHRoutedSocket {
|
|
self.progress = progress
|
|
openCount += 1
|
|
return try await withCheckedThrowingContinuation { continuation = $0 }
|
|
}
|
|
|
|
func emit(_ event: SSHRouteEvent) { progress?(event) }
|
|
|
|
func finish(_ socket: SSHRoutedSocket) {
|
|
continuation?.resume(returning: socket)
|
|
continuation = nil
|
|
}
|
|
|
|
func fail() {
|
|
continuation?.resume(throwing: CancellationError())
|
|
continuation = nil
|
|
}
|
|
}
|