Fix false SSH host-death verdict and remote diagnostics (#9971)

* test(ssh): cover false remote death recovery

* ci: exercise SSH regression baseline

* fix(ssh): preserve remote PTY recovery

* fix(ssh): harden recovery diagnostics

* fix(remote): keep log routing portable

* test: follow renamed resume command field

* test: cover recovery preservation edge cases

* fix: preserve recovery state through inconclusive probes

* test: exercise terminal teardown through app fixture

* test(remote): cover closed daemon output descriptors

* fix: harden recovery review edge cases

* test: substitute pinned SSH in teardown fixture

* test: cover final recovery review edge cases

* fix: preserve recovery through final reconciliation edges

* test: distinguish automatic and manual PTY recovery

* fix: report PTY recovery state accurately

* test: make reconciliation parameter type explicit

* test: require reconciliation acknowledgement contracts

* Fix SSH startup fixtures after main merge

* Test confirmed PTY end cleanup failure

* Finalize cleanup for confirmed PTY exit

* Keep SSH auth fixtures off live network

* Fix SSH tests after reconnect merge

* Test terminal reset before no-progress reattach

* Reset terminal modes before no-progress reattach

* Run no-progress reset regression in package tests

* Fix SSH test fixtures after reconnect merge
This commit is contained in:
Austin Wang
2026-08-12 00:43:25 +00:00
committed by GitHub
parent 16c3e69e8b
commit c8b9afc23d
27 changed files with 2204 additions and 320 deletions
+151 -48
View File
@@ -130,72 +130,70 @@ extension CMUXCLI {
surfaceID: String?,
sessionID: String,
lifecycleID: String,
reconciliationConfirmedSessionEnded: inout Bool,
intentionalOnly: Bool,
sessionRunningExitCode: SSHPTYAttachExitCode = .bridgeClosedSessionRunning
sessionRunningExitCode: SSHPTYAttachExitCode = .bridgeClosedSessionRunning,
reconciliationUnavailableExitCode: SSHPTYAttachExitCode = .retryableTransient
) throws -> Bool {
let reconciliationFailure = "ssh-pty-attach: bridge closed before remote PTY exit could be confirmed"
let response: [String: Any]
do {
var params: [String: Any] = [
"workspace_id": workspaceId,
"session_id": sessionID,
"lifecycle_id": lifecycleID,
"acknowledge_lifecycle_if_session_absent": !intentionalOnly,
]
if let surfaceID {
params["surface_id"] = surfaceID
params["allow_moved_surface"] = true
}
response = try client.sendV2(method: "workspace.remote.pty_sessions", params: params)
} catch {
throw CLIError(
message: "\(reconciliationFailure): \(userFacingRemotePTYErrorMessage(error))",
exitCode: SSHPTYAttachExitCode.retryableTransient
)
var params: [String: Any] = [
"workspace_id": workspaceId,
"session_id": sessionID,
"lifecycle_id": lifecycleID,
"acknowledge_lifecycle_if_session_absent": false,
]
if let surfaceID {
params["surface_id"] = surfaceID
params["allow_moved_surface"] = true
}
let requestedLifecycle = (response["requested_session_lifecycle"] as? String)?
.trimmingCharacters(in: .whitespacesAndNewlines)
let intentionalCleanup = requestedLifecycle == "intentional_cleanup_requested" ||
requestedLifecycle == "intentionally_closed"
guard let sessions = response["sessions"] as? [[String: Any]] else {
throw CLIError(message: reconciliationFailure, exitCode: SSHPTYAttachExitCode.retryableTransient)
}
let errors: [[String: Any]]
if let rawErrors = response["errors"] {
guard let parsedErrors = rawErrors as? [[String: Any]] else {
throw CLIError(message: reconciliationFailure, exitCode: SSHPTYAttachExitCode.retryableTransient)
}
errors = parsedErrors
} else {
errors = []
}
if !intentionalCleanup, !errors.isEmpty {
throw CLIError(
message: "\(reconciliationFailure)\n\(sshSessionListFailureMessage(errors))",
exitCode: SSHPTYAttachExitCode.retryableTransient
var reconciliation = try requestValidatedSSHPTYReconciliation(
client: client,
params: params,
unavailableExitCode: reconciliationUnavailableExitCode
)
if !intentionalOnly,
!reconciliation.intentionalCleanup,
!reconciliation.sessionIDs.contains(sessionID) {
// Keep the first liveness read side-effect free. Only after its
// response is validated may the server atomically recheck absence
// and acknowledge this exact lifecycle generation.
params["acknowledge_lifecycle_if_session_absent"] = true
reconciliation = try requestValidatedSSHPTYReconciliation(
client: client,
params: params,
unavailableExitCode: reconciliationUnavailableExitCode
)
}
if intentionalOnly, !intentionalCleanup { return false }
if intentionalOnly, !reconciliation.intentionalCleanup { return false }
let sessionStillRunning = sessions.contains {
(($0["session_id"] as? String)?.trimmingCharacters(in: .whitespacesAndNewlines) ?? "") == sessionID
}
if !intentionalCleanup, sessionStillRunning {
if !reconciliation.intentionalCleanup,
reconciliation.sessionIDs.contains(sessionID) {
let message: String
if sessionRunningExitCode == .bridgeClosedWithoutProgress {
message = String(
localized: "cli.sshPtyAttach.bridgeClosedWithoutProgress",
defaultValue: "ssh-pty-attach: bridge closed without receiving new output while the remote PTY session is still running"
defaultValue: "ssh-pty-attach: bridge closed without receiving new output while the remote PTY session is still running",
bundle: CLIExecutableLocator.enclosingAppBundle() ?? .main
)
} else if sshPTYAttachWrapperWillRetry(sessionRunningExitCode) {
message = String(
localized: "cli.sshPtyAttach.bridgeClosedSessionRunningReconnecting",
defaultValue: "The SSH terminal connection ended while the remote session is still running; cmux is reconnecting.",
bundle: CLIExecutableLocator.enclosingAppBundle() ?? .main
)
} else {
message = "ssh-pty-attach: bridge closed while remote PTY session is still running"
message = String(
localized: "cli.sshPtyAttach.bridgeClosedSessionRunning",
defaultValue: "The SSH terminal connection ended; the remote session may still be running.",
bundle: CLIExecutableLocator.enclosingAppBundle() ?? .main
)
}
throw CLIError(
message: message,
exitCode: sessionRunningExitCode
)
}
reconciliationConfirmedSessionEnded = true
guard let surfaceID else { return true }
do {
_ = try client.sendV2(method: "workspace.remote.pty_attach_end", params: [
@@ -206,12 +204,117 @@ extension CMUXCLI {
} catch {
throw CLIError(
message: "ssh-pty-attach: remote PTY exited but local session cleanup failed: \(userFacingRemotePTYErrorMessage(error))",
exitCode: SSHPTYAttachExitCode.retryableTransient
exitCode: SSHPTYAttachExitCode.fatal
)
}
return true
}
private func requestValidatedSSHPTYReconciliation(
client: SocketClient,
params: [String: Any],
unavailableExitCode: SSHPTYAttachExitCode
) throws -> (intentionalCleanup: Bool, sessionIDs: [String]) {
let response: [String: Any]
do {
response = try client.sendV2(method: "workspace.remote.pty_sessions", params: params)
} catch {
throw CLIError(
message: sshPTYReconciliationUnavailableMessage(
detail: userFacingRemotePTYErrorMessage(error)
),
exitCode: unavailableExitCode
)
}
return try validatedSSHPTYReconciliation(
response,
unavailableExitCode: unavailableExitCode
)
}
private func validatedSSHPTYReconciliation(
_ response: [String: Any],
unavailableExitCode: SSHPTYAttachExitCode
) throws -> (intentionalCleanup: Bool, sessionIDs: [String]) {
let requestedLifecycle: String?
if let rawRequestedLifecycle = response["requested_session_lifecycle"] {
guard let rawRequestedLifecycle = rawRequestedLifecycle as? String else {
throw CLIError(
message: sshPTYReconciliationUnavailableMessage(detail: nil),
exitCode: unavailableExitCode
)
}
let normalizedLifecycle = rawRequestedLifecycle
.trimmingCharacters(in: .whitespacesAndNewlines)
guard [
"active",
"intentional_cleanup_requested",
"intentionally_closed",
].contains(normalizedLifecycle) else {
throw CLIError(
message: sshPTYReconciliationUnavailableMessage(detail: nil),
exitCode: unavailableExitCode
)
}
requestedLifecycle = normalizedLifecycle
} else {
requestedLifecycle = nil
}
let intentionalCleanup = requestedLifecycle == "intentional_cleanup_requested" ||
requestedLifecycle == "intentionally_closed"
guard let sessions = response["sessions"] as? [[String: Any]] else {
throw CLIError(
message: sshPTYReconciliationUnavailableMessage(detail: nil),
exitCode: unavailableExitCode
)
}
let sessionIDs = sessions.compactMap { session -> String? in
guard let rawSessionID = session["session_id"] as? String else { return nil }
let normalizedSessionID = rawSessionID.trimmingCharacters(in: .whitespacesAndNewlines)
return normalizedSessionID.isEmpty ? nil : normalizedSessionID
}
guard sessionIDs.count == sessions.count else {
throw CLIError(
message: sshPTYReconciliationUnavailableMessage(detail: nil),
exitCode: unavailableExitCode
)
}
let errors: [[String: Any]]
if let rawErrors = response["errors"] {
guard let parsedErrors = rawErrors as? [[String: Any]] else {
throw CLIError(
message: sshPTYReconciliationUnavailableMessage(detail: nil),
exitCode: unavailableExitCode
)
}
errors = parsedErrors
} else {
errors = []
}
if !intentionalCleanup, !errors.isEmpty {
throw CLIError(
message: sshPTYReconciliationUnavailableMessage(
detail: sshSessionListFailureMessage()
),
exitCode: unavailableExitCode
)
}
return (intentionalCleanup, sessionIDs)
}
private func sshPTYReconciliationUnavailableMessage(detail: String?) -> String {
let message = String(
localized: "cli.sshPtyAttach.reconciliationUnavailableReattach",
defaultValue: "The SSH terminal connection ended before the remote session state could be confirmed; preserving the remote session for reconnection.",
bundle: CLIExecutableLocator.enclosingAppBundle() ?? .main
)
guard let detail = detail?.trimmingCharacters(in: .whitespacesAndNewlines),
!detail.isEmpty else {
return message
}
return "\(message): \(detail)"
}
func readSSHPTYBridgeReady(fd: Int32) throws -> (attachmentToken: String, replayBytes: Int) {
let maxStatusBytes = 4096
// Bound only the pre-ready status wait: a bridge that accepts the TCP
+15 -5
View File
@@ -4,15 +4,25 @@ import Foundation
extension CMUXCLI {
func sshAutoReconnectNoteFormat() -> String {
let status = String(localized: "cli.ssh.autoReconnect.status", defaultValue: "[cmux] ssh exited with status %s; reconnecting (attempt %s/%s).")
let stopHint = String(localized: "cli.ssh.autoReconnect.stopHint", defaultValue: "[cmux] close this pane or press Ctrl-C to stop reconnecting.")
let bundle = CLIExecutableLocator.enclosingAppBundle() ?? .main
let status = String(localized: "cli.ssh.autoReconnect.status", defaultValue: "[cmux] ssh exited with status %s; reconnecting (attempt %s/%s).", bundle: bundle)
let stopHint = String(localized: "cli.ssh.autoReconnect.stopHint", defaultValue: "[cmux] close this pane or press Ctrl-C to stop reconnecting.", bundle: bundle)
return "\\n\\033[33m\(status)\\033[0m\\n\\033[2m\(stopHint)\\033[0m\\n"
}
func sshManualReconnectExitPromptFormat() -> String {
let status = String(localized: "cli.ssh.manualReconnectPrompt.status", defaultValue: "[cmux] ssh exited with status %s.")
let detail = String(localized: "cli.ssh.manualReconnectPrompt.detail", defaultValue: "[cmux] the remote VM may have been paused, destroyed, or lost network.")
let prompt = String(localized: "cli.ssh.manualReconnectPrompt.prompt", defaultValue: "[cmux] press Enter to close this pane. Press r then Enter to reconnect.")
let bundle = CLIExecutableLocator.enclosingAppBundle() ?? .main
let status = String(localized: "cli.ssh.manualReconnectPrompt.status", defaultValue: "[cmux] ssh exited with status %s.", bundle: bundle)
let detail = String(localized: "cli.ssh.manualReconnectPrompt.detail", defaultValue: "[cmux] the SSH connection ended; the remote session may still be running.", bundle: bundle)
let prompt = String(localized: "cli.ssh.manualReconnectPrompt.prompt", defaultValue: "[cmux] press Enter to close this pane. Press r then Enter to reconnect.", bundle: bundle)
return "\\n\\033[31m\(status)\\033[0m\\n\\033[2m\(detail)\\033[0m\\n\\033[2m\(prompt)\\033[0m\\n"
}
func sshTerminalExitPromptFormat() -> String {
let bundle = CLIExecutableLocator.enclosingAppBundle() ?? .main
let status = String(localized: "cli.ssh.manualReconnectPrompt.status", defaultValue: "[cmux] ssh exited with status %s.", bundle: bundle)
let detail = String(localized: "cli.ssh.manualReconnectPrompt.detail", defaultValue: "[cmux] the SSH connection ended; the remote session may still be running.", bundle: bundle)
let prompt = String(localized: "cli.ssh.terminalExitPrompt.prompt", defaultValue: "[cmux] press Enter to close this pane.", bundle: bundle)
return "\\n\\033[31m\(status)\\033[0m\\n\\033[2m\(detail)\\033[0m\\n\\033[2m\(prompt)\\033[0m\\n"
}
+5 -1
View File
@@ -306,6 +306,8 @@ extension CMUXCLI {
terminalFailureCommand: "break"
)
let backoffBuilder = SSHRetryBackoffScriptBuilder(context: .startup)
let terminalModeReset = shellQuote(SSHTerminalModeResetSequence().shellPrintfFormat)
let terminalExitPrompt = shellQuote(sshTerminalExitPromptFormat())
let terminalExitPromptCommand = [
shellQuote(resolvedExecutableURL()?.path ?? (args.first ?? "cmux")),
"__ssh-terminal-exit-prompt",
@@ -377,6 +379,7 @@ extension CMUXCLI {
"CMUX_SSH_CHILD_PID=; CMUX_SSH_AUTH_PID=; CMUX_SSH_PENDING_SIGNAL=; CMUX_SSH_PENDING_SIGNAL_NAME=",
] + backoffBuilder.stateInitializationLines + [
"cmux_ssh_note() { if [ -t 2 ]; then printf \"$@\" >&2 || true; fi; }",
"cmux_ssh_reset_terminal_modes() { if [ -t 2 ]; then printf \(terminalModeReset) >&2 || true; fi; }",
"cmux_ssh_register_attempt() { \(lifecycleLaunching); }",
"cmux_ssh_begin_attempt() { CMUX_SSH_ATTEMPT_ID=$(/usr/bin/uuidgen | /usr/bin/tr '[:upper:]' '[:lower:]') || return 1; export CMUX_SSH_ATTEMPT_ID; cmux_ssh_attempt_registration_retry=0; while ! cmux_ssh_register_attempt; do cmux_ssh_attempt_registration_retry=$((cmux_ssh_attempt_registration_retry + 1)); if [ \"$cmux_ssh_attempt_registration_retry\" -ge 3 ]; then return 1; fi; /bin/sleep 0.1; done; }",
"cmux_ssh_session_end() { if [ \"${CMUX_SSH_SESSION_ENDED:-0}\" = 1 ]; then return; fi; CMUX_SSH_SESSION_ENDED=1; cmux_ssh_cleanup_password; \(lifecycleCleanup); }",
@@ -428,6 +431,7 @@ extension CMUXCLI {
" cmux_ssh_status=$?",
" CMUX_SSH_CHILD_PID=",
" if [ \"$cmux_ssh_status\" -eq 0 ]; then break; fi",
" cmux_ssh_reset_terminal_modes",
" case \"$cmux_ssh_status\" in \(retryableStatusPattern)) ;; *) break ;; esac",
]
if retryPTYAttachStatus {
@@ -466,7 +470,7 @@ extension CMUXCLI {
" trap 'cmux_ssh_prompt_signal_exit 129' HUP",
" trap 'cmux_ssh_prompt_signal_exit 130' INT",
" trap 'cmux_ssh_prompt_signal_exit 143' TERM",
" printf '\\n\\033[31m[cmux] ssh exited with status %s.\\033[0m\\n\\033[2m[cmux] the remote VM may have been paused, destroyed, or lost network.\\033[0m\\n\\033[2m[cmux] press Enter to close this pane.\\033[0m\\n' \"$cmux_ssh_status\" >&2 || true",
" printf \(terminalExitPrompt) \"$cmux_ssh_status\" >&2 || true",
" if [ -t 0 ]; then \(terminalExitPromptCommand) <&0; else exec \(terminalExitPromptCommand) <&0; fi",
" cmux_ssh_prompt_restore_tty",
" trap - EXIT HUP INT TERM",
+55 -52
View File
@@ -12452,7 +12452,7 @@ struct CMUXCLI {
if jsonOutput {
print(jsonString(formatIDs(response, mode: idFormat)))
if !errors.isEmpty {
throw CLIError(message: sshSessionListFailureMessage(errors))
throw CLIError(message: sshSessionListFailureMessage())
}
return
}
@@ -12477,21 +12477,16 @@ struct CMUXCLI {
print("\(workspacePrefix)\(sessionID) attachments=\(attachments.count) size=\(effectiveCols)x\(effectiveRows) scrollback_bytes=\(scrollbackBytes)")
}
if !errors.isEmpty {
throw CLIError(message: sshSessionListFailureMessage(errors))
throw CLIError(message: sshSessionListFailureMessage())
}
}
func sshSessionListFailureMessage(_ errors: [[String: Any]]) -> String {
let count = errors.count
let summary = "ssh-session-list failed for \(count) remote workspace\(count == 1 ? "" : "s")"
let details = errors.map { error in
let workspace = debugString(error["workspace_ref"])
?? debugString(error["workspace_id"])
?? "workspace:?"
let message = userFacingRemotePTYErrorMessage(error["error"])
return "- \(workspace): \(message)"
}
return ([summary] + details).joined(separator: "\n")
func sshSessionListFailureMessage() -> String {
return String(
localized: "cli.sshSessionList.remoteStateUnavailable",
defaultValue: "Remote PTY session state is unavailable for one or more workspaces.",
bundle: CLIExecutableLocator.enclosingAppBundle() ?? .main
)
}
private func runSSHSessionCleanup(
@@ -12861,24 +12856,36 @@ struct CMUXCLI {
var bridgeReachedReady = false
var sessionLostWillRespawn = false
var wrapperWillRetrySameSurface = false
var preserveLifecycleForRecovery = false
var noProgressRetryExhausted = false
var reconciliationConfirmedSessionEnded = false
var attachFinished = false
var attachmentToken = ""
var readinessDelivery: Task<Void, Never>?
func reconcileBridgeEnd(
intentionalOnly: Bool,
sessionRunningExitCode: SSHPTYAttachExitCode = .bridgeClosedSessionRunning
sessionRunningExitCode: SSHPTYAttachExitCode = .bridgeClosedSessionRunning,
reconciliationUnavailableExitCode: SSHPTYAttachExitCode = .retryableTransient
) throws -> Bool {
do {
return try reconcileSSHPTYBridgeEnd(
client: client, workspaceId: workspaceId, surfaceID: surfaceID,
sessionID: sessionID,
lifecycleID: lifecycleID,
reconciliationConfirmedSessionEnded: &reconciliationConfirmedSessionEnded,
intentionalOnly: intentionalOnly,
sessionRunningExitCode: sessionRunningExitCode
sessionRunningExitCode: sessionRunningExitCode,
reconciliationUnavailableExitCode: reconciliationUnavailableExitCode
)
} catch let error as CLIError {
if reconciliationConfirmedSessionEnded {
preserveLifecycleForRecovery = false
wrapperWillRetrySameSurface = false
}
if let exitCode = SSHPTYAttachExitCode(rawValue: error.exitCode) {
if exitCode == .bridgeClosedSessionRunning || exitCode == .retryableTransient {
preserveLifecycleForRecovery = true
}
if sshPTYAttachWrapperWillRetry(exitCode) {
wrapperWillRetrySameSurface = true
} else if exitCode == .bridgeClosedWithoutProgress {
@@ -12899,10 +12906,15 @@ struct CMUXCLI {
lifecycleID: lifecycleID,
attachmentID: attachmentID,
attachmentToken: attachmentToken,
retireLifecycle: !sessionLostWillRespawn && !wrapperWillRetrySameSurface,
clearLocalSurface: (!bridgeReachedReady || noProgressRetryExhausted)
&& !sessionLostWillRespawn
&& !wrapperWillRetrySameSurface
retireLifecycle: reconciliationConfirmedSessionEnded
|| (!sessionLostWillRespawn
&& !wrapperWillRetrySameSurface
&& !preserveLifecycleForRecovery),
clearLocalSurface: reconciliationConfirmedSessionEnded
|| ((!bridgeReachedReady || noProgressRetryExhausted)
&& !sessionLostWillRespawn
&& !wrapperWillRetrySameSurface
&& !preserveLifecycleForRecovery)
)
}
}
@@ -12943,23 +12955,19 @@ struct CMUXCLI {
wrapperWillRetrySameSurface = true
}
if closedGeneration {
if (try? reconcileBridgeEnd(intentionalOnly: true)) == true {
if try reconcileBridgeEnd(intentionalOnly: true) {
attachFinished = true
return
}
cleanupFailedSSHPTYAttach(
client: client,
workspaceId: workspaceId,
surfaceID: surfaceID,
sessionID: sessionID,
lifecycleID: lifecycleID,
attachmentID: attachmentID,
attachmentToken: attachmentToken,
retireLifecycle: true,
clearLocalSurface: true
preserveLifecycleForRecovery = true
let recoveryExitCode = SSHPTYAttachExitCode.retryableTransient
if sshPTYAttachWrapperWillRetry(recoveryExitCode) {
wrapperWillRetrySameSurface = true
}
throw CLIError(
message: "ssh-pty-attach: \(userFacingRemotePTYErrorMessage(error))",
exitCode: recoveryExitCode
)
attachFinished = true
return
}
if exitCode.isWrapperRetryable {
if try reconcileBridgeEnd(intentionalOnly: true) {
@@ -13087,6 +13095,19 @@ struct CMUXCLI {
}
var reconnectInputFilterStopRequested = false
var outputProgress = SSHPTYAttachOutputProgress(replayBytes: bridgeReplayBytes)
func finishBridgeClosedNormally() throws {
resizeMonitor.cancel()
readinessDelivery?.cancel()
_ = try reconcileBridgeEnd(
intentionalOnly: false,
sessionRunningExitCode: sshPTYAttachBridgeClosedExitCode(
receivedLiveOutput: outputProgress.receivedLiveOutput,
readyUptime: bridgeReadyUptime
),
reconciliationUnavailableExitCode: .bridgeClosedSessionRunning
)
attachFinished = true
}
var outputBuffer = [UInt8](repeating: 0, count: 32768)
while true {
@@ -13096,29 +13117,11 @@ struct CMUXCLI {
reconnectInputFilterControl?.stopFilteringBeforeFirstOutput(unlessAlreadyRequested: &reconnectInputFilterStopRequested)
cliWriteStdout(Data(outputBuffer.prefix(count)))
} else if count == 0 {
resizeMonitor.cancel()
readinessDelivery?.cancel()
_ = try reconcileBridgeEnd(
intentionalOnly: false,
sessionRunningExitCode: sshPTYAttachBridgeClosedExitCode(
receivedLiveOutput: outputProgress.receivedLiveOutput,
readyUptime: bridgeReadyUptime
)
)
attachFinished = true
try finishBridgeClosedNormally()
return
} else if errno != EINTR {
if sshPTYBridgeReadErrorIsEOF(errno) {
resizeMonitor.cancel()
readinessDelivery?.cancel()
_ = try reconcileBridgeEnd(
intentionalOnly: false,
sessionRunningExitCode: sshPTYAttachBridgeClosedExitCode(
receivedLiveOutput: outputProgress.receivedLiveOutput,
readyUptime: bridgeReadyUptime
)
)
attachFinished = true
try finishBridgeClosedNormally()
return
}
throw CLIError(message: "ssh-pty-attach: bridge read failed")
@@ -19,7 +19,12 @@ public enum SSHPTYAttachExitCode: Int32 {
/// A persistent PTY session that no longer exists and must be respawned.
case sessionNotFound = 253
/// A closed bridge whose persistent PTY session is still running.
/// A closed established bridge that must preserve its persistent PTY session for reattach.
///
/// This covers both a session confirmed running and a post-close liveness
/// query made inconclusive by the tunnel replacement race. In either case,
/// retiring the session would destroy recoverable state; the next
/// `--require-existing` attach is the authoritative probe.
case bridgeClosedSessionRunning = 254
/// A transient transport or daemon failure that may succeed after reconnecting.
@@ -77,6 +82,7 @@ public enum SSHPTYAttachExitCode: Int32 {
let retryWithoutReauthenticationStatus = retryableWithoutReauthentication.rawValue
let sessionRunningStatus = bridgeClosedSessionRunning.rawValue
let transientStatus = retryableTransient.rawValue
let terminalModeReset = SSHTerminalModeResetSequence().shellPrintfFormat.remoteCommandShellQuoted
return [
"cmux_ssh_attach_reconnect_limit=\"${CMUX_SSH_RECONNECT_LIMIT:-}\"",
@@ -101,6 +107,7 @@ public enum SSHPTYAttachExitCode: Int32 {
" if [ \"$cmux_ssh_attach_reconnect_unbounded\" -eq 1 ] || [ \"$cmux_ssh_attach_retry\" -lt \"$cmux_ssh_attach_reconnect_limit\" ]; then cmux_ssh_attach_can_retry=1; else cmux_ssh_attach_can_retry=0; fi",
" CMUX_SSH_PTY_ATTACH_WRAPPER_CAN_RETRY=\"$cmux_ssh_attach_can_retry\" CMUX_SSH_PTY_ATTACH_NO_PROGRESS_RETRY=\"$cmux_ssh_attach_no_progress_retry\" CMUX_SSH_PTY_ATTACH_NO_PROGRESS_LIMIT=\"$cmux_ssh_attach_no_progress_limit\" \(command)",
" cmux_ssh_attach_status=$?",
" if [ \"$cmux_ssh_attach_status\" -ne 0 ] && [ -t 2 ]; then printf \(terminalModeReset) >&2 || true; fi",
" case \"$cmux_ssh_attach_status\" in",
" \(noProgressPolicy.status)) cmux_ssh_attach_no_progress_retry=$((cmux_ssh_attach_no_progress_retry + 1)); cmux_ssh_attach_reconnect_delay=\"$cmux_ssh_attach_reconnect_initial_delay\"; \(noProgressPolicy.limitReachedCommand) ;;",
" \(retryWithoutReauthenticationStatus)) cmux_ssh_attach_no_progress_retry=0 ;;",
@@ -121,10 +128,11 @@ public enum SSHPTYAttachExitCode: Int32 {
/// Builds a bounded no-progress sub-loop for a wrapper that already owns
/// general reconnect and foreground-authentication policy.
///
/// Status 252 is consumed until its health budget is exhausted. All other
/// statuses, including 251, 254, and 255, return unchanged to the enclosing
/// wrapper so its existing reconnect and reauthentication behavior remains
/// the single owner of those transitions.
/// Status 252 is consumed until its health budget is exhausted, with terminal
/// reporting modes reset before each reattach. All other statuses, including
/// 251, 254, and 255, return unchanged to the enclosing wrapper so its existing
/// reconnect and reauthentication behavior remains the single owner of those
/// transitions.
///
/// The attach environment is exported on its own lines rather than as an
/// assignment prefix, because a prefix is only legal before a simple command
@@ -134,6 +142,7 @@ public enum SSHPTYAttachExitCode: Int32 {
/// - Returns: Shell source lines implementing the no-progress budget.
public static func noProgressRetryLoopLines(command: String) -> [String] {
let policy = noProgressShellPolicy()
let terminalModeReset = SSHTerminalModeResetSequence().shellPrintfFormat.remoteCommandShellQuoted
return policy.configurationLines + [
"cmux_ssh_attach_no_progress_retry=0",
@@ -143,6 +152,7 @@ public enum SSHPTYAttachExitCode: Int32 {
" export CMUX_SSH_PTY_ATTACH_NO_PROGRESS_RETRY CMUX_SSH_PTY_ATTACH_NO_PROGRESS_LIMIT",
" \(command)",
" cmux_ssh_attach_status=$?",
" if [ \"$cmux_ssh_attach_status\" -eq \(policy.status) ] && [ -t 2 ]; then printf \(terminalModeReset) >&2 || true; fi",
" if [ \"$cmux_ssh_attach_status\" -ne \(policy.status) ]; then exit \"$cmux_ssh_attach_status\"; fi",
" cmux_ssh_attach_no_progress_retry=$((cmux_ssh_attach_no_progress_retry + 1))",
" \(policy.limitReachedCommand)",
@@ -38,6 +38,7 @@ public struct SSHPTYAttachRetryScriptBuilder: Sendable {
let noProgressStatus = SSHPTYAttachExitCode.bridgeClosedWithoutProgress.rawValue
let sessionRunningStatus = SSHPTYAttachExitCode.bridgeClosedSessionRunning.rawValue
let transientStatus = SSHPTYAttachExitCode.retryableTransient.rawValue
let terminalModeReset = SSHTerminalModeResetSequence().shellPrintfFormat.remoteCommandShellQuoted
var lines = [
"cmux_ssh_attach_reconnect_limit=\"${CMUX_SSH_RECONNECT_LIMIT:-}\"",
"case \"$cmux_ssh_attach_reconnect_limit\" in '') cmux_ssh_attach_reconnect_limit='∞'; cmux_ssh_attach_reconnect_unbounded=1 ;; *[!0-9]*) cmux_ssh_attach_reconnect_limit=20; cmux_ssh_attach_reconnect_unbounded=0 ;; *) cmux_ssh_attach_reconnect_unbounded=0 ;; esac",
@@ -74,6 +75,7 @@ public struct SSHPTYAttachRetryScriptBuilder: Sendable {
" if [ \"$cmux_ssh_attach_reconnect_unbounded\" -eq 1 ] || [ \"$cmux_ssh_attach_retry\" -lt \"$cmux_ssh_attach_reconnect_limit\" ]; then cmux_ssh_attach_can_retry=1; else cmux_ssh_attach_can_retry=0; fi",
" CMUX_SSH_PTY_ATTACH_WRAPPER_CAN_RETRY=\"$cmux_ssh_attach_can_retry\" CMUX_SSH_PTY_ATTACH_NO_PROGRESS_RETRY=\"$cmux_ssh_attach_no_progress_retry\" CMUX_SSH_PTY_ATTACH_NO_PROGRESS_LIMIT=\"$cmux_ssh_attach_no_progress_limit\" \(command)",
" cmux_ssh_attach_status=$?",
" if [ \"$cmux_ssh_attach_status\" -ne 0 ] && [ -t 2 ]; then printf \(terminalModeReset) >&2 || true; fi",
" case \"$cmux_ssh_attach_status\" in",
" \(noProgressStatus)) cmux_ssh_attach_no_progress_retry=$((cmux_ssh_attach_no_progress_retry + 1)); cmux_ssh_attach_reconnect_delay=\"$cmux_ssh_attach_reconnect_initial_delay\"; \(noProgressPolicy.limitReachedCommand) ;;",
" \(retryWithoutReauthenticationStatus)) cmux_ssh_attach_no_progress_retry=0 ;;",
@@ -0,0 +1,17 @@
/// Builds the terminal-mode reset emitted between a remote PTY and a local SSH prompt.
public struct SSHTerminalModeResetSequence: Sendable {
/// Creates a terminal-mode reset sequence builder.
public init() {}
/// A `printf` format that disables input and reporting modes a remote TUI may leave enabled.
///
/// The value uses shell `printf` escapes rather than literal control bytes
/// so generated startup scripts remain readable and safely quotable.
public var shellPrintfFormat: String {
var format = "\\033[?1000l\\033[?1002l\\033[?1003l\\033[?1004l"
format += "\\033[?1005l\\033[?1006l\\033[?1015l\\033[?1016l"
format += "\\033[?2004l\\033[999<u\\033[0;1=u\\033[>m"
format += "\\033[?2031l\\033[?2048l\\033[?2026l"
return format
}
}
@@ -65,6 +65,71 @@ struct SSHPTYAttachExitCodeTests {
#expect(!fileManager.fileExists(atPath: authAttempts.path))
}
@Test("A no-progress retry resets terminal reporting modes before reattaching")
func noProgressRetryResetsTerminalModesBeforeReattaching() throws {
let fileManager = FileManager.default
let root = fileManager.temporaryDirectory
.appendingPathComponent("cmux-no-progress-terminal-reset-\(UUID().uuidString)")
let attemptFile = root.appendingPathComponent("attempts")
let scriptFile = root.appendingPathComponent("loop.sh")
try fileManager.createDirectory(at: root, withIntermediateDirectories: true)
defer { try? fileManager.removeItem(at: root) }
let retryLines = SSHPTYAttachExitCode.noProgressRetryLoopLines(
command: "cmux_test_attach"
)
try Self.writeExecutable(
scriptFile,
([
"#!/bin/sh",
"cmux_test_attach() {",
" count=$(cat \"$CMUX_TEST_ATTEMPT_FILE\" 2>/dev/null || printf 0)",
" count=$((count + 1))",
" printf '%s' \"$count\" > \"$CMUX_TEST_ATTEMPT_FILE\"",
" printf 'attempt:%s\\n' \"$count\" >&2",
" if [ \"$count\" -eq 1 ]; then return \(SSHPTYAttachExitCode.bridgeClosedWithoutProgress.rawValue); fi",
" return 0",
"}",
] + retryLines).joined(separator: "\n")
)
let transcriptPipe = Pipe()
let process = Process()
process.executableURL = URL(fileURLWithPath: "/usr/bin/script")
process.arguments = ["-q", "-F", "/dev/null", "/bin/sh", scriptFile.path]
process.environment = ProcessInfo.processInfo.environment.merging([
"CMUX_TEST_ATTEMPT_FILE": attemptFile.path,
]) { _, override in override }
process.standardInput = FileHandle.nullDevice
process.standardOutput = transcriptPipe
process.standardError = transcriptPipe
try process.run()
let transcriptData = transcriptPipe.fileHandleForReading.readDataToEndOfFile()
process.waitUntilExit()
let transcript = String(data: transcriptData, encoding: .utf8) ?? ""
#expect(process.terminationStatus == 0, Comment(rawValue: transcript))
let secondAttempt = transcript.range(of: "attempt:2")
#expect(secondAttempt != nil, Comment(rawValue: transcript))
let requiredResets = [
"\u{1B}[?1004l", // focus reporting
"\u{1B}[?1000l", // mouse reporting
"\u{1B}[?2004l", // bracketed paste
"\u{1B}[999<u", // Kitty keyboard stack
"\u{1B}[0;1=u", // Kitty keyboard flags
"\u{1B}[?2048l", // in-band resize reports
"\u{1B}[?2026l", // synchronized output
]
for reset in requiredResets {
let resetRange = transcript.range(of: reset)
#expect(resetRange != nil, Comment(rawValue: transcript))
if let resetRange, let secondAttempt {
#expect(resetRange.lowerBound < secondAttempt.lowerBound)
}
}
}
@Test("lifecycle codes retain precedence over transient-looking messages")
func lifecycleCodesRetainPrecedence() {
#expect(
@@ -20,7 +20,7 @@ struct SSHPTYAttachRetryScriptBuilderTests {
"sleep() { printf 'sleep:%s\\n' \"$1\" >> \"$CMUX_TEST_LOG\"; }",
"cmux_test_attach() { printf '%s\\n' attach >> \"$CMUX_TEST_LOG\"; return 7; }",
"cmux_ssh_attach_foreground_auth() {",
" count=$(test -f \"$CMUX_TEST_LOG\" && grep -c '^auth$' \"$CMUX_TEST_LOG\" 2>/dev/null || printf 0)",
" count=$(grep -c '^auth$' \"$CMUX_TEST_LOG\" 2>/dev/null) || count=0",
" printf '%s\\n' auth >> \"$CMUX_TEST_LOG\"",
" if [ \"$count\" -eq 0 ]; then return 254; fi",
" return 0",
@@ -53,7 +53,7 @@ struct SSHPTYAttachRetryScriptBuilderTests {
"sleep() { printf 'sleep:%s\\n' \"$1\" >> \"$CMUX_TEST_LOG\"; }",
"cmux_test_attach() {",
"""
count=$(test -f "$CMUX_TEST_LOG" && grep -c '^attach$' "$CMUX_TEST_LOG" 2>/dev/null || printf 0)
count=$(grep -c '^attach$' "$CMUX_TEST_LOG" 2>/dev/null) || count=0
printf '%s\\n' attach >> "$CMUX_TEST_LOG"
if [ "$count" -eq 0 ]; then return 255; fi
return 253
@@ -95,7 +95,7 @@ struct SSHPTYAttachRetryScriptBuilderTests {
" return 7",
"}",
"cmux_ssh_attach_foreground_auth() {",
" count=$(test -f \"$CMUX_TEST_LOG\" && grep -c '^auth$' \"$CMUX_TEST_LOG\" 2>/dev/null || printf 0)",
" count=$(grep -c '^auth$' \"$CMUX_TEST_LOG\" 2>/dev/null) || count=0",
" printf '%s\\n' auth >> \"$CMUX_TEST_LOG\"",
" if [ \"$count\" -eq 0 ]; then return 0; fi",
" if [ \"$count\" -eq 1 ]; then return 252; fi",
@@ -134,7 +134,7 @@ struct SSHPTYAttachRetryScriptBuilderTests {
"sleep() { :; }",
"cmux_test_attach() { printf '%s\\n' attach >> \"$CMUX_TEST_LOG\"; return 255; }",
"cmux_ssh_attach_foreground_auth() {",
" count=$(test -f \"$CMUX_TEST_LOG\" && grep -c '^auth$' \"$CMUX_TEST_LOG\" 2>/dev/null || printf 0)",
" count=$(grep -c '^auth$' \"$CMUX_TEST_LOG\" 2>/dev/null) || count=0",
" printf '%s\\n' auth >> \"$CMUX_TEST_LOG\"",
" if [ \"$count\" -eq 0 ]; then return 0; fi",
" return 255",
@@ -279,7 +279,7 @@ struct SSHPTYAttachRetryScriptBuilderTests {
let script = ([
"cmux_ssh_attach_signal_exit() { exit \"$1\"; }",
"cmux_test_attach() {",
" count=$(test -f \"$CMUX_TEST_LOG\" && grep -c '^attach$' \"$CMUX_TEST_LOG\" 2>/dev/null || printf 0)",
" count=$(grep -c '^attach$' \"$CMUX_TEST_LOG\" 2>/dev/null) || count=0",
" printf '%s\\n' attach >> \"$CMUX_TEST_LOG\"",
" if [ \"$count\" -eq 0 ]; then return 255; fi",
" IFS= read -r cmux_test_input || return 42",
+87 -110
View File
@@ -53654,124 +53654,16 @@
"cli.ssh.manualReconnectPrompt.detail": {
"extractionState": "manual",
"localizations": {
"ar": {
"stringUnit": {
"state": "translated",
"value": "[cmux] ربما تم إيقاف الجهاز الافتراضي البعيد مؤقتًا أو حذفه أو فقد الاتصال بالشبكة."
}
},
"bs": {
"stringUnit": {
"state": "translated",
"value": "[cmux] udaljeni VM je možda pauziran, uništen ili je izgubio mrežu."
}
},
"da": {
"stringUnit": {
"state": "translated",
"value": "[cmux] den eksterne VM kan være sat på pause, slettet eller have mistet netværket."
}
},
"de": {
"stringUnit": {
"state": "translated",
"value": "[cmux] die Remote-VM wurde möglicherweise angehalten, gelöscht oder hat die Netzwerkverbindung verloren."
}
},
"en": {
"stringUnit": {
"state": "translated",
"value": "[cmux] the remote VM may have been paused, destroyed, or lost network."
}
},
"es": {
"stringUnit": {
"state": "translated",
"value": "[cmux] la VM remota puede estar pausada, destruida o sin conexión de red."
}
},
"fr": {
"stringUnit": {
"state": "translated",
"value": "[cmux] la VM distante a peut-être été suspendue, détruite ou a perdu le réseau."
}
},
"it": {
"stringUnit": {
"state": "translated",
"value": "[cmux] la VM remota potrebbe essere stata messa in pausa, eliminata o aver perso la rete."
"value": "[cmux] the SSH connection ended; the remote session may still be running."
}
},
"ja": {
"stringUnit": {
"state": "translated",
"value": "[cmux] リモート VM は一時停止、削除、またはネットワーク切断された可能性があります。"
}
},
"km": {
"stringUnit": {
"state": "translated",
"value": "[cmux] VM ពីចម្ងាយអាចត្រូវបានផ្អាក លុបចោល ឬបាត់បង់បណ្តាញ។"
}
},
"ko": {
"stringUnit": {
"state": "translated",
"value": "[cmux] 원격 VM이 일시 중지, 삭제되었거나 네트워크 연결이 끊겼을 수 있습니다."
}
},
"nb": {
"stringUnit": {
"state": "translated",
"value": "[cmux] den eksterne VM-en kan være satt på pause, slettet eller ha mistet nettverket."
}
},
"pl": {
"stringUnit": {
"state": "translated",
"value": "[cmux] zdalna VM mogła zostać wstrzymana, usunięta lub utracić sieć."
}
},
"pt-BR": {
"stringUnit": {
"state": "translated",
"value": "[cmux] a VM remota pode ter sido pausada, destruída ou perdido a rede."
}
},
"ru": {
"stringUnit": {
"state": "translated",
"value": "[cmux] удаленная VM могла быть приостановлена, удалена или потерять сеть."
}
},
"th": {
"stringUnit": {
"state": "translated",
"value": "[cmux] VM ระยะไกลอาจถูกพัก ถูกลบ หรือสูญเสียการเชื่อมต่อเครือข่าย"
}
},
"tr": {
"stringUnit": {
"state": "translated",
"value": "[cmux] uzak VM duraklatılmış, silinmiş veya ağ bağlantısını kaybetmiş olabilir."
}
},
"uk": {
"stringUnit": {
"state": "translated",
"value": "[cmux] віддалену VM могло бути призупинено, видалено або вона втратила мережу."
}
},
"zh-Hans": {
"stringUnit": {
"state": "translated",
"value": "[cmux] 远程 VM 可能已暂停、被销毁或失去网络连接。"
}
},
"zh-Hant": {
"stringUnit": {
"state": "translated",
"value": "[cmux] 遠端 VM 可能已暫停、被刪除或失去網路連線。"
"value": "[cmux] SSH 接続が終了しました。リモートセッションはまだ実行中の可能性があります。"
}
}
}
@@ -54776,6 +54668,23 @@
}
}
},
"cli.ssh.terminalExitPrompt.prompt": {
"extractionState": "manual",
"localizations": {
"en": {
"stringUnit": {
"state": "translated",
"value": "[cmux] press Enter to close this pane."
}
},
"ja": {
"stringUnit": {
"state": "translated",
"value": "[cmux] 閉じるには Enter を押してください。"
}
}
}
},
"cli.sshPtyAttach.bridgeClosedReattaching": {
"extractionState": "manual",
"localizations": {
@@ -54793,6 +54702,40 @@
}
}
},
"cli.sshPtyAttach.bridgeClosedSessionRunning": {
"extractionState": "manual",
"localizations": {
"en": {
"stringUnit": {
"state": "translated",
"value": "The SSH terminal connection ended; the remote session may still be running."
}
},
"ja": {
"stringUnit": {
"state": "translated",
"value": "SSH ターミナル接続が終了しました。リモートセッションはまだ実行中の可能性があります。"
}
}
}
},
"cli.sshPtyAttach.bridgeClosedSessionRunningReconnecting": {
"extractionState": "manual",
"localizations": {
"en": {
"stringUnit": {
"state": "translated",
"value": "The SSH terminal connection ended while the remote session is still running; cmux is reconnecting."
}
},
"ja": {
"stringUnit": {
"state": "translated",
"value": "SSH ターミナル接続が終了しましたが、リモートセッションは引き続き実行中です。再接続します。"
}
}
}
},
"cli.sshPtyAttach.bridgeClosedWithoutProgress": {
"extractionState": "manual",
"localizations": {
@@ -54844,6 +54787,23 @@
}
}
},
"cli.sshPtyAttach.reconciliationUnavailableReattach": {
"extractionState": "manual",
"localizations": {
"en": {
"stringUnit": {
"state": "translated",
"value": "The SSH terminal connection ended before the remote session state could be confirmed; preserving the remote session for reconnection."
}
},
"ja": {
"stringUnit": {
"state": "translated",
"value": "リモートセッションの状態を確認できる前に SSH ターミナル接続が終了しました。再接続のためセッションを維持します。"
}
}
}
},
"cli.sshPtyAttach.remoteSessionLostRespawn": {
"extractionState": "manual",
"localizations": {
@@ -54861,6 +54821,23 @@
}
}
},
"cli.sshSessionList.remoteStateUnavailable": {
"extractionState": "manual",
"localizations": {
"en": {
"stringUnit": {
"state": "translated",
"value": "Remote PTY session state is unavailable for one or more workspaces."
}
},
"ja": {
"stringUnit": {
"state": "translated",
"value": "1 つ以上のワークスペースでリモート PTY セッションの状態を取得できません。"
}
}
}
},
"cli.tmux-compat.error.downstreamTmuxMissing": {
"extractionState": "manual",
"localizations": {
@@ -3687,7 +3687,11 @@ final class CLINotifyProcessIntegrationRegressionTests: XCTestCase {
line: line
)
XCTAssertTrue(
script.contains("case \"$cmux_ssh_status\" in 254|255"),
script.contains("case \"$cmux_ssh_status\" in 254)")
&& script.contains(
"252) cmux_ssh_status=255; if [ \"$cmux_ssh_auth_succeeded\" -eq 0 ]; then break; fi"
)
&& script.contains("*) break ;; esac; fi"),
script,
file: file,
line: line
@@ -4200,6 +4204,7 @@ final class CLINotifyProcessIntegrationRegressionTests: XCTestCase {
"workspace.remote.pty_bridge",
"workspace.remote.pty_resize",
"workspace.remote.pty_sessions",
"workspace.remote.pty_sessions",
"workspace.remote.pty_attach_end",
]
)
@@ -4300,7 +4305,12 @@ final class CLINotifyProcessIntegrationRegressionTests: XCTestCase {
XCTAssertTrue(result.stdout.isEmpty, result.stdout)
XCTAssertTrue(result.stderr.isEmpty, result.stderr)
let methods = state.snapshot().compactMap { self.jsonObject($0)?["method"] as? String }
XCTAssertEqual(methods, ["workspace.remote.pty_bridge", "workspace.remote.pty_resize", "workspace.remote.pty_sessions"])
XCTAssertEqual(methods, [
"workspace.remote.pty_bridge",
"workspace.remote.pty_resize",
"workspace.remote.pty_sessions",
"workspace.remote.pty_sessions",
])
}
func testSSHPTYAttachBridgeResetWhenSessionGoneClearsLocalState() throws {
@@ -4313,6 +4323,8 @@ final class CLINotifyProcessIntegrationRegressionTests: XCTestCase {
let surfaceId = "33333333-3333-3333-3333-333333333333"
let sessionId = "ssh-\(workspaceId)-\(surfaceId)"
let token = "bridge-token"
let resizeObserved = DispatchSemaphore(value: 0)
let readinessObserved = DispatchSemaphore(value: 0)
defer {
Darwin.close(listenerFD)
@@ -4346,6 +4358,7 @@ final class CLINotifyProcessIntegrationRegressionTests: XCTestCase {
let params = payload["params"] as? [String: Any] ?? [:]
XCTAssertEqual(params["attachment_token"] as? String, "attach-token")
XCTAssertEqual(params["surface_id"] as? String, surfaceId)
resizeObserved.signal()
return self.v2Response(id: id, ok: true, result: ["resized": true])
case "workspace.remote.terminal_session_connected":
let params = payload["params"] as? [String: Any] ?? [:]
@@ -4356,6 +4369,7 @@ final class CLINotifyProcessIntegrationRegressionTests: XCTestCase {
XCTAssertNotNil(
(params["lifecycle_id"] as? String).flatMap(UUID.init(uuidString:))
)
readinessObserved.signal()
return self.v2Response(id: id, ok: true, result: ["connected": true])
case "workspace.remote.pty_sessions":
return self.v2Response(
@@ -4388,7 +4402,10 @@ final class CLINotifyProcessIntegrationRegressionTests: XCTestCase {
)
}
}
let bridgeHandled = startBridgeReadyThenResetAfterClientEOFServer(listenerFD: bridge.fd)
let bridgeHandled = startBridgeReadyThenResetAfterClientEOFServer(
listenerFD: bridge.fd,
waitBeforeClientEOF: [resizeObserved, readinessObserved]
)
var environment = ProcessInfo.processInfo.environment
environment["CMUX_SOCKET_PATH"] = socketPath
@@ -4419,6 +4436,7 @@ final class CLINotifyProcessIntegrationRegressionTests: XCTestCase {
"workspace.remote.pty_bridge",
"workspace.remote.pty_resize",
"workspace.remote.pty_sessions",
"workspace.remote.pty_sessions",
"workspace.remote.pty_attach_end",
]
)
@@ -4856,7 +4874,7 @@ final class CLINotifyProcessIntegrationRegressionTests: XCTestCase {
)
}
XCTAssertEqual(methods.filter { $0 == "workspace.remote.pty_resize" }.count, 1)
XCTAssertEqual(methods.filter { $0 == "workspace.remote.pty_sessions" }.count, 1)
XCTAssertEqual(methods.filter { $0 == "workspace.remote.pty_sessions" }.count, 2)
XCTAssertEqual(methods.filter { $0 == "workspace.remote.pty_attach_end" }.count, 1)
}
@@ -5069,7 +5087,13 @@ final class CLINotifyProcessIntegrationRegressionTests: XCTestCase {
XCTAssertTrue(initialCommand.contains(sessionId), initialCommand)
XCTAssertTrue(initialCommand.contains("CMUX_WORKSPACE_ID"), initialCommand)
XCTAssertTrue(initialCommand.contains("CMUX_SURFACE_ID"), initialCommand)
XCTAssertTrue(initialCommand.contains("251|254|255") && initialCommand.contains("CMUX_SSH_RECONNECT_MAX_DELAY_SECONDS") && initialCommand.contains(""), initialCommand)
let retryStatuses = ["251)", "252)", "254)", "255)"]
XCTAssertTrue(
retryStatuses.allSatisfy { initialCommand.contains($0) }
&& initialCommand.contains("CMUX_SSH_RECONNECT_MAX_DELAY_SECONDS")
&& initialCommand.contains(""),
initialCommand
)
XCTAssertEqual(initialCommand.components(separatedBy: "/usr/bin/uuidgen").count - 1, 2, initialCommand)
XCTAssertTrue(initialCommand.contains("ssh-session-end --lifecycle-only"), initialCommand)
return self.v2Response(
@@ -5725,7 +5749,7 @@ final class CLINotifyProcessIntegrationRegressionTests: XCTestCase {
XCTAssertFalse(state.snapshot().contains { $0.contains("workspace.remote.pty_close") })
XCTAssertTrue(result.stderr.contains("ssh-session-cleanup failed for 1 persisted SSH PTY session"), result.stderr)
XCTAssertTrue(result.stderr.contains(sessionId), result.stderr)
XCTAssertTrue(result.stderr.contains("persistent SSH PTY session is no longer running"), result.stderr)
XCTAssertTrue(result.stderr.contains("remote PTY operation failed"), result.stderr)
}
func testSSHSessionCleanupAllWorkspacesSessionIDCountsDuplicateIDsPerWorkspace() throws {
+8 -1
View File
@@ -244,7 +244,10 @@ extension CLINotifyProcessIntegrationRegressionTests {
return handled
}
func startBridgeReadyThenResetAfterClientEOFServer(listenerFD: Int32) -> XCTestExpectation {
func startBridgeReadyThenResetAfterClientEOFServer(
listenerFD: Int32,
waitBeforeClientEOF: [DispatchSemaphore] = []
) -> XCTestExpectation {
let handled = expectation(description: "pty bridge ready reset server handled")
DispatchQueue.global(qos: .userInitiated).async {
defer { handled.fulfill() }
@@ -291,6 +294,10 @@ extension CLINotifyProcessIntegrationRegressionTests {
}
}
for semaphore in waitBeforeClientEOF {
guard semaphore.wait(timeout: .now() + 5) == .success else { return }
}
while true {
let count = Darwin.read(clientFD, &buffer, buffer.count)
if count > 0 {
@@ -10,12 +10,153 @@ import Testing
#endif
extension CLINotifyProcessIntegrationRegressionTests {
func testSSHPTYReconciliationPreservesSessionForReattach() throws {
let cliPath = try bundledCLIPath()
let scenarios: [(name: String, wrapperRetry: String?, confirmsRunning: Bool)] = [
("pending-inconclusive", "1", false),
("exhausted-inconclusive", "0", false),
("direct-inconclusive", nil, false),
("pending-confirmed", "1", true),
("exhausted-confirmed", "0", true),
("direct-confirmed", nil, true),
]
for (index, scenario) in scenarios.enumerated() {
let socketPath = makeSocketPath("sshptyreconcile\(index)")
let listenerFD = try bindUnixSocket(at: socketPath)
let bridge = try bindLoopbackTCP()
let state = MockSocketServerState()
let workspaceID = "22222222-2222-2222-2222-222222222222"
let surfaceID = "33333333-3333-3333-3333-333333333333"
let sessionID = "ssh-\(workspaceID)-\(surfaceID)"
defer {
Darwin.close(listenerFD)
Darwin.close(bridge.fd)
unlink(socketPath)
}
let socketHandled = startMockServer(listenerFD: listenerFD, state: state) { line in
guard let payload = self.jsonObject(line),
let id = payload["id"] as? String,
let method = payload["method"] as? String else {
return self.malformedRequestResponse(raw: line)
}
switch method {
case "workspace.remote.pty_bridge":
return self.v2Response(id: id, ok: true, result: [
"host": "127.0.0.1",
"port": bridge.port,
"token": "bridge-token",
"session_id": sessionID,
"attachment_id": surfaceID,
])
case "workspace.remote.pty_resize":
return self.v2Response(id: id, ok: true, result: ["resized": true])
case "workspace.remote.pty_sessions":
if scenario.confirmsRunning {
return self.v2Response(id: id, ok: true, result: [
"sessions": [["session_id": sessionID]],
"errors": [],
])
}
// This is the exact callback-order race from issue 9965: the
// per-channel bridge has closed and the broker has already
// removed its ready tunnel, even though the daemon and PTY
// session can still be healthy behind the replacement tunnel.
return self.v2Response(
id: id,
ok: false,
error: [
"code": "remote_pty_error",
"message": "remote daemon tunnel is not ready",
]
)
case "workspace.remote.pty_detach":
return self.v2Response(id: id, ok: true, result: ["detached": true])
case "workspace.remote.pty_attach_end":
return self.v2Response(id: id, ok: true, result: ["ended": true])
default:
return self.v2Response(
id: id,
ok: false,
error: ["code": "unexpected_method", "message": "Unexpected method \(method)"]
)
}
}
let bridgeHandled = startBridgeReadyThenCloseServer(listenerFD: bridge.fd)
var environment = ProcessInfo.processInfo.environment
environment["CMUX_SOCKET_PATH"] = socketPath
environment["CMUX_CLI_SENTRY_DISABLED"] = "1"
environment.removeValue(forKey: "CMUX_SSH_PTY_ATTACH_WRAPPER_CAN_RETRY")
environment["CMUX_SSH_PTY_ATTACH_WRAPPER_CAN_RETRY"] = scenario.wrapperRetry
let result = runProcess(
executablePath: cliPath,
arguments: [
"ssh-pty-attach",
"--wait",
"--require-existing",
"--workspace", workspaceID,
"--session-id", sessionID,
"--attachment-id", surfaceID,
],
environment: environment,
timeout: 5
)
wait(for: [socketHandled, bridgeHandled], timeout: 5)
#expect(!result.timedOut, Comment(rawValue: scenario.name))
#expect(
result.status == SSHPTYAttachExitCode.bridgeClosedSessionRunning.rawValue,
Comment(rawValue: scenario.name)
)
if scenario.confirmsRunning {
let claimsAutomaticReconnect = result.stderr.contains("cmux is reconnecting")
#expect(
claimsAutomaticReconnect == (scenario.wrapperRetry == "1"),
Comment(rawValue: "\(scenario.name): \(result.stderr)")
)
if scenario.wrapperRetry != "1" {
#expect(result.stderr.contains("remote session may still be running"))
}
} else {
#expect(result.stderr.contains("remote session state could be confirmed"))
}
let requests = state.snapshot().compactMap { self.jsonObject($0) }
let reconciliationRequests = requests.filter {
$0["method"] as? String == "workspace.remote.pty_sessions"
}
#expect(reconciliationRequests.count == 1, Comment(rawValue: scenario.name))
for request in reconciliationRequests {
guard let params = request["params"] as? [String: Any] else {
#expect(Bool(false), Comment(rawValue: scenario.name))
continue
}
#expect(
params["acknowledge_lifecycle"] as? Bool != true,
Comment(rawValue: scenario.name)
)
#expect(
params["acknowledge_lifecycle_if_session_absent"] as? Bool != true,
Comment(rawValue: scenario.name)
)
}
let methods = requests.compactMap { $0["method"] as? String }
#expect(!methods.contains("workspace.remote.pty_attach_end"))
}
}
func testSSHPTYReconciliationRejectsMalformedLifecyclePayloads() throws {
let malformedResults: [[String: Any]] = [
["errors": []],
["sessions": [], "errors": "invalid"],
["sessions": [[:]], "errors": []],
["sessions": [["session_id": " "]], "errors": []],
["sessions": [["session_id": 42]], "errors": []],
["requested_session_lifecycle": 42, "sessions": [], "errors": []],
["requested_session_lifecycle": " ", "sessions": [], "errors": []],
["requested_session_lifecycle": "unknown", "sessions": [], "errors": []],
]
for (index, malformedResult) in malformedResults.enumerated() {
let malformedCase = "malformed result \(index): \(malformedResult)"
let cliPath = try bundledCLIPath()
let socketPath = makeSocketPath("sshptymalformed\(index)")
let listenerFD = try bindUnixSocket(at: socketPath)
@@ -77,12 +218,215 @@ extension CLINotifyProcessIntegrationRegressionTests {
)
wait(for: [socketHandled, bridgeHandled], timeout: 5)
#expect(!result.timedOut)
#expect(result.status == SSHPTYAttachExitCode.retryableTransient.rawValue)
#expect(result.stderr.contains("bridge closed before remote PTY exit could be confirmed"))
#expect(!result.timedOut, Comment(rawValue: malformedCase))
#expect(
result.status == SSHPTYAttachExitCode.bridgeClosedSessionRunning.rawValue,
Comment(rawValue: malformedCase)
)
#expect(
result.stderr.contains("remote session state could be confirmed"),
Comment(rawValue: malformedCase)
)
let requests = state.snapshot().compactMap { self.jsonObject($0) }
let methods = requests.compactMap { $0["method"] as? String }
#expect(
!methods.contains("workspace.remote.pty_attach_end"),
Comment(rawValue: malformedCase)
)
let reconciliationRequests = requests.filter {
$0["method"] as? String == "workspace.remote.pty_sessions"
}
#expect(reconciliationRequests.count == 1, Comment(rawValue: malformedCase))
for request in reconciliationRequests {
guard let params = request["params"] as? [String: Any] else {
#expect(Bool(false), Comment(rawValue: malformedCase))
continue
}
#expect(
params["acknowledge_lifecycle"] as? Bool != true,
Comment(rawValue: malformedCase)
)
#expect(
params["acknowledge_lifecycle_if_session_absent"] as? Bool != true,
Comment(rawValue: malformedCase)
)
}
}
}
func testSSHPTYAttachClosedGenerationPreservesActiveLifecycleWithoutSessionProof() throws {
let cliPath = try bundledCLIPath()
let socketPath = makeSocketPath("sshptyclosedactive")
let listenerFD = try bindUnixSocket(at: socketPath)
let state = MockSocketServerState()
let workspaceID = "22222222-2222-2222-2222-222222222222"
let surfaceID = "33333333-3333-3333-3333-333333333333"
let sessionID = "ssh-\(workspaceID)-\(surfaceID)"
let lifecycleID = "44444444-4444-4444-4444-444444444444"
defer {
Darwin.close(listenerFD)
unlink(socketPath)
}
let socketHandled = startMockServer(listenerFD: listenerFD, state: state) { line in
guard let payload = self.jsonObject(line),
let id = payload["id"] as? String,
let method = payload["method"] as? String else {
return self.malformedRequestResponse(raw: line)
}
switch method {
case "workspace.remote.pty_bridge":
return self.v2Response(
id: id,
ok: false,
error: ["code": "pty_lifecycle_closed", "message": "remote PTY operation failed"]
)
case "workspace.remote.pty_sessions":
return self.v2Response(id: id, ok: true, result: [
"sessions": [],
"errors": [],
])
case "workspace.remote.pty_attach_end":
return self.v2Response(id: id, ok: true, result: ["ended": true])
default:
return self.v2Response(
id: id,
ok: false,
error: ["code": "unexpected_method", "message": "Unexpected method \(method)"]
)
}
}
var environment = ProcessInfo.processInfo.environment
environment["CMUX_SOCKET_PATH"] = socketPath
environment["CMUX_CLI_SENTRY_DISABLED"] = "1"
environment["CMUX_SSH_PTY_ATTACH_WRAPPER_CAN_RETRY"] = "1"
let result = runProcess(
executablePath: cliPath,
arguments: [
"ssh-pty-attach", "--wait", "--require-existing",
"--workspace", workspaceID, "--session-id", sessionID,
"--lifecycle-id", lifecycleID, "--attachment-id", surfaceID,
],
environment: environment,
timeout: 5
)
wait(for: [socketHandled], timeout: 5)
#expect(!result.timedOut, Comment(rawValue: result.stderr))
#expect(
result.status == SSHPTYAttachExitCode.retryableTransient.rawValue,
Comment(rawValue: result.stderr)
)
let requests = state.snapshot().compactMap { self.jsonObject($0) }
let methods = requests.compactMap { $0["method"] as? String }
#expect(methods.filter { $0 == "workspace.remote.pty_bridge" }.count == 1)
#expect(methods.filter { $0 == "workspace.remote.pty_sessions" }.count == 1)
#expect(!methods.contains("workspace.remote.pty_attach_end"), Comment(rawValue: "\(methods)"))
let reconciliationParams: [[String: Any]] = requests.compactMap { request in
guard request["method"] as? String == "workspace.remote.pty_sessions" else { return nil }
return request["params"] as? [String: Any]
}
#expect(reconciliationParams.count == 1)
guard let reconciliationParams = reconciliationParams.first else { return }
#expect(reconciliationParams["acknowledge_lifecycle"] as? Bool != true)
#expect(reconciliationParams["acknowledge_lifecycle_if_session_absent"] as? Bool != true)
}
func testSSHPTYAttachCleanupFailureDoesNotRetryConfirmedEndedSession() throws {
let cliPath = try bundledCLIPath()
let socketPath = makeSocketPath("sshptyendedcleanup")
let listenerFD = try bindUnixSocket(at: socketPath)
let bridge = try bindLoopbackTCP()
let state = MockSocketServerState()
let workspaceID = "22222222-2222-2222-2222-222222222222"
let surfaceID = "33333333-3333-3333-3333-333333333333"
let sessionID = "ssh-\(workspaceID)-\(surfaceID)"
let lifecycleID = "44444444-4444-4444-4444-444444444444"
defer {
Darwin.close(listenerFD)
Darwin.close(bridge.fd)
unlink(socketPath)
}
let socketHandled = startMockServer(listenerFD: listenerFD, state: state) { line in
guard let payload = self.jsonObject(line),
let id = payload["id"] as? String,
let method = payload["method"] as? String else {
return self.malformedRequestResponse(raw: line)
}
switch method {
case "workspace.remote.pty_bridge":
return self.v2Response(id: id, ok: true, result: [
"host": "127.0.0.1",
"port": bridge.port,
"token": "bridge-token",
"session_id": sessionID,
"attachment_id": surfaceID,
])
case "workspace.remote.pty_sessions":
return self.v2Response(id: id, ok: true, result: [
"sessions": [],
"errors": [],
])
case "workspace.remote.pty_resize", "workspace.remote.pty_detach":
return self.v2Response(id: id, ok: true, result: [:])
case "workspace.remote.pty_attach_end":
let attemptCount = state.snapshot().compactMap { self.jsonObject($0) }.filter {
$0["method"] as? String == "workspace.remote.pty_attach_end"
}.count
if attemptCount == 1 {
return self.v2Response(
id: id,
ok: false,
error: ["code": "cleanup_failed", "message": "transient cleanup failure"]
)
}
return self.v2Response(id: id, ok: true, result: ["ended": true])
default:
return self.v2Response(
id: id,
ok: false,
error: ["code": "unexpected_method", "message": "Unexpected method \(method)"]
)
}
}
let bridgeHandled = startBridgeReadyThenCloseServer(listenerFD: bridge.fd)
var environment = ProcessInfo.processInfo.environment
environment["CMUX_SOCKET_PATH"] = socketPath
environment["CMUX_CLI_SENTRY_DISABLED"] = "1"
environment["CMUX_SSH_PTY_ATTACH_WRAPPER_CAN_RETRY"] = "1"
environment.removeValue(forKey: "CMUX_TERMINAL_LIFECYCLE_ID")
environment.removeValue(forKey: "CMUX_SSH_ATTEMPT_ID")
let result = runProcess(
executablePath: cliPath,
arguments: [
"ssh-pty-attach", "--wait", "--require-existing",
"--workspace", workspaceID, "--session-id", sessionID,
"--lifecycle-id", lifecycleID, "--attachment-id", surfaceID,
],
environment: environment,
timeout: 5
)
wait(for: [socketHandled, bridgeHandled], timeout: 5)
#expect(!result.timedOut, Comment(rawValue: result.stderr))
#expect(result.status == SSHPTYAttachExitCode.fatal.rawValue, Comment(rawValue: result.stderr))
let requests = state.snapshot().compactMap { self.jsonObject($0) }
let methods = requests.compactMap { $0["method"] as? String }
#expect(methods.filter { $0 == "workspace.remote.pty_bridge" }.count == 1)
#expect(methods.filter { $0 == "workspace.remote.pty_sessions" }.count == 3)
#expect(methods.filter { $0 == "workspace.remote.pty_attach_end" }.count == 2)
#expect(methods.filter { $0 == "workspace.remote.pty_detach" }.count == 1)
let lifecycleRetirements = requests.filter { request in
guard request["method"] as? String == "workspace.remote.pty_sessions",
let params = request["params"] as? [String: Any] else { return false }
return params["acknowledge_lifecycle"] as? Bool == true
}
#expect(lifecycleRetirements.count == 1)
}
func testSSHPTYAttachPreservesPipedProbeLikeInputBeforeForwardingInput() throws {
let cliPath = try bundledCLIPath()
let socketPath = makeSocketPath("sshptyprobe")
@@ -181,11 +525,21 @@ extension CLINotifyProcessIntegrationRegressionTests {
#expect(result.stderr.isEmpty)
let forwardedBridgeInput = bridgeInput.snapshot()
#expect(String(data: forwardedBridgeInput, encoding: .utf8) == queuedProbeReplies + forwardedInput)
let methods = state.snapshot().compactMap { self.jsonObject($0)?["method"] as? String }
let requests = state.snapshot().compactMap { self.jsonObject($0) }
let methods = requests.compactMap { $0["method"] as? String }
#expect(methods == [
"workspace.remote.pty_bridge",
"workspace.remote.pty_sessions",
"workspace.remote.pty_sessions",
"workspace.remote.pty_attach_end",
])
let reconciliationParams: [[String: Any]] = requests.compactMap { request in
guard request["method"] as? String == "workspace.remote.pty_sessions" else { return nil }
return request["params"] as? [String: Any]
}
#expect(reconciliationParams.count == 2)
guard reconciliationParams.count == 2 else { return }
#expect(reconciliationParams[0]["acknowledge_lifecycle_if_session_absent"] as? Bool != true)
#expect(reconciliationParams[1]["acknowledge_lifecycle_if_session_absent"] as? Bool == true)
}
}
@@ -236,7 +236,7 @@ extension CLINotifyProcessIntegrationRegressionTests {
XCTAssertEqual(methods.filter { $0 == "workspace.remote.pty_attach_end" }.count, 1, "\(methods)")
}
func testSSHPTYAttachClosedGenerationBeforeReadyEndsWithoutWrapperRetry() throws {
func testSSHPTYAttachClosedGenerationPreservesStateWhenReconciliationIsUnavailable() throws {
let cliPath = try bundledCLIPath()
let socketPath = makeSocketPath("sshptyclosedstart")
let listenerFD = try bindUnixSocket(at: socketPath)
@@ -301,10 +301,22 @@ extension CLINotifyProcessIntegrationRegressionTests {
wait(for: [socketHandled], timeout: 5)
XCTAssertFalse(result.timedOut, result.stderr)
XCTAssertEqual(result.status, 0, result.stderr)
let methods = state.snapshot().compactMap { self.jsonObject($0)?["method"] as? String }
XCTAssertEqual(result.status, SSHPTYAttachExitCode.retryableTransient.rawValue, result.stderr)
let requests = state.snapshot().compactMap { self.jsonObject($0) }
let methods = requests.compactMap { $0["method"] as? String }
XCTAssertEqual(methods.filter { $0 == "workspace.remote.pty_bridge" }.count, 1, "\(methods)")
XCTAssertEqual(methods.filter { $0 == "workspace.remote.pty_attach_end" }.count, 1, "\(methods)")
XCTAssertEqual(methods.filter { $0 == "workspace.remote.pty_sessions" }.count, 1, "\(methods)")
XCTAssertFalse(methods.contains("workspace.remote.pty_attach_end"), "\(methods)")
let reconciliationParams = requests.compactMap { request -> [String: Any]? in
guard request["method"] as? String == "workspace.remote.pty_sessions" else { return nil }
return request["params"] as? [String: Any]
}
XCTAssertEqual(reconciliationParams.count, 1)
XCTAssertNotEqual(reconciliationParams.first?["acknowledge_lifecycle"] as? Bool, true)
XCTAssertNotEqual(
reconciliationParams.first?["acknowledge_lifecycle_if_session_absent"] as? Bool,
true
)
}
func testSSHPTYAttachCapacityFailureKeepsSurfaceForWrapperRetry() throws {
@@ -408,7 +420,7 @@ extension CLINotifyProcessIntegrationRegressionTests {
guard request["method"] as? String == "workspace.remote.pty_sessions" else { return nil }
return (request["params"] as? [String: Any])?["acknowledge_lifecycle_if_session_absent"] as? Bool
}
XCTAssertEqual(reconciliationFlags, [false, true])
XCTAssertEqual(reconciliationFlags, [false, false, true])
let methods = requests.compactMap { $0["method"] as? String }
XCTAssertEqual(methods.filter { $0 == "workspace.remote.pty_bridge" }.count, 2, "\(methods)")
XCTAssertEqual(methods.filter { $0 == "workspace.remote.pty_attach_end" }.count, 1, "\(methods)")
@@ -205,7 +205,7 @@ extension CLINotifyProcessIntegrationRegressionTests {
environment["CMUX_SSH_RECONNECT_DELAY_SECONDS"] = "2"
environment["CMUX_SSH_RECONNECT_MAX_DELAY_SECONDS"] = "2"
let command = SSHPTYAttachStartupCommandBuilder.command(
let generatedCommand = SSHPTYAttachStartupCommandBuilder.command(
sessionID: "ssh-test-session",
foregroundAuth: SSHPTYAttachStartupCommandBuilder.ForegroundAuth(
destination: "[email protected]",
@@ -215,6 +215,11 @@ extension CLINotifyProcessIntegrationRegressionTests {
token: "foreground-auth-token"
)
)
XCTAssertTrue(generatedCommand.contains("/usr/bin/ssh"), generatedCommand)
let command = generatedCommand.replacingOccurrences(
of: "/usr/bin/ssh",
with: fakeSSH.path
)
let result = runProcess(
executablePath: "/bin/sh",
arguments: ["-c", command],
@@ -275,7 +280,10 @@ extension CLINotifyProcessIntegrationRegressionTests {
let generatedScript = try persistentSSHInitialStartupScriptForReconnectTest()
let bundledCLI = try bundledCLIPath()
let rewrittenScript = generatedScript.replacingOccurrences(of: bundledCLI, with: fakeAttach.path)
XCTAssertTrue(generatedScript.contains("/usr/bin/ssh"), generatedScript)
let rewrittenScript = generatedScript
.replacingOccurrences(of: bundledCLI, with: fakeAttach.path)
.replacingOccurrences(of: "/usr/bin/ssh", with: fakeAuth.path)
XCTAssertNotEqual(rewrittenScript, generatedScript, "Expected generated wrapper to reference the bundled CLI")
try writeSSHPTYReconnectTestShell(at: fakeStartup, contents: rewrittenScript)
for executable in [fakeStartup, fakeAuth, fakeAttach, fakeSleep] {
@@ -53,7 +53,7 @@ struct SSHForegroundAuthenticationMarkerCleanupTests {
sshOptions: ["ControlMaster=no"],
token: "foreground-auth-token"
)
)
).replacingOccurrences(of: "/usr/bin/ssh", with: fakeSSH.path)
let process = Process()
process.executableURL = URL(fileURLWithPath: "/bin/sh")
process.arguments = ["-c", "exec \(command)"]
@@ -155,7 +155,7 @@ struct SSHForegroundAuthenticationMarkerCleanupTests {
sshOptions: ["ControlMaster=no"],
token: "foreground-auth-token"
)
)
).replacingOccurrences(of: "/usr/bin/ssh", with: fakeSSH.path)
let result = try Self.runProcess(command: command, environment: environment)
#expect(result.status == 253, Comment(rawValue: result.stderr))
@@ -251,7 +251,7 @@ struct SSHForegroundAuthenticationMarkerCleanupTests {
sshOptions: sshOptions,
token: "foreground-auth-token"
)
)
).replacingOccurrences(of: "/usr/bin/ssh", with: fakeSSH.path)
let result = try Self.runProcess(command: command, environment: environment)
#expect(result.status == 253, Comment(rawValue: result.stderr))
@@ -318,7 +318,7 @@ struct SSHForegroundAuthenticationMarkerCleanupTests {
sshOptions: ["ControlMaster=no"],
token: "foreground-auth-token"
)
),
).replacingOccurrences(of: "/usr/bin/ssh", with: fakeSSH.path),
environment: environment
)
+152 -19
View File
@@ -79,7 +79,9 @@ struct SSHStartupManualReconnectTests {
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeCLI.path)
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeSSH.path)
let startupCommand = try Self.generatedVMSSHInitialStartupCommand()
let startupCommand = try Self.generatedVMSSHInitialStartupCommand(
replacingSystemSSHWith: fakeSSH
)
#expect(!startupCommand.contains("workspace.remote.terminal_session_connected"))
var environment = ProcessInfo.processInfo.environment
environment["PATH"] = "\(root.path):\(environment["PATH"] ?? "/usr/bin:/bin")"
@@ -116,6 +118,73 @@ struct SSHStartupManualReconnectTests {
)
}
@Test func terminalTeardownDisablesRemoteInputReportingModesBeforePrompt() throws {
let fileManager = FileManager.default
let root = fileManager.temporaryDirectory
.appendingPathComponent("cmux-ssh-terminal-mode-reset-\(UUID().uuidString)", isDirectory: true)
let fakeCLI = root.appendingPathComponent("cmux")
let fakeSSH = root.appendingPathComponent("ssh")
try fileManager.createDirectory(at: root, withIntermediateDirectories: true)
defer { try? fileManager.removeItem(at: root) }
try Self.writeShellFile(at: fakeCLI, lines: ["#!/bin/sh", "exit 0"])
try Self.writeShellFile(at: fakeSSH, lines: ["#!/bin/sh", "exit 7"])
for executable in [fakeCLI, fakeSSH] {
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: executable.path)
}
let generatedStartupCommand = try Self.generatedVMSSHInitialStartupCommand(
replacingSystemSSHWith: fakeSSH
)
let generatedStartupURL = URL(
fileURLWithPath: generatedStartupCommand.trimmingCharacters(in: .whitespacesAndNewlines)
)
defer { try? fileManager.removeItem(at: generatedStartupURL) }
let generatedStartupScript = try String(contentsOf: generatedStartupURL, encoding: .utf8)
try #require(generatedStartupScript.contains(fakeSSH.path))
let startupURL = root.appendingPathComponent("startup-with-fake-ssh.sh")
try generatedStartupScript.write(to: startupURL, atomically: true, encoding: .utf8)
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: startupURL.path)
try fileManager.removeItem(at: generatedStartupURL)
var environment = ProcessInfo.processInfo.environment
environment["CMUX_BUNDLED_CLI_PATH"] = fakeCLI.path
environment["CMUX_SOCKET_PATH"] = "/tmp/cmux-debug-test.sock"
environment["CMUX_WORKSPACE_ID"] = "11111111-1111-1111-1111-111111111111"
environment["CMUX_SURFACE_ID"] = "22222222-2222-2222-2222-222222222222"
environment["CMUX_SSH_RECONNECT_LIMIT"] = "0"
let result = Self.runProcess(
executablePath: "/usr/bin/script",
arguments: ["-q", "-F", "/dev/null", "/bin/sh", startupURL.path],
environment: environment,
standardInput: "\n",
timeout: 5
)
let transcript = result.stdout + result.stderr
#expect(!result.timedOut, Comment(rawValue: transcript))
#expect(result.status == 7, Comment(rawValue: transcript))
let requiredResets = [
"\u{1B}[?1004l", // focus reporting
"\u{1B}[?1000l", // mouse reporting
"\u{1B}[?2004l", // bracketed paste
"\u{1B}[999<u", // Kitty keyboard stack
"\u{1B}[0;1=u", // Kitty keyboard flags
"\u{1B}[?2048l", // in-band resize reports
"\u{1B}[?2026l", // synchronized output
]
let closePrompt = transcript.range(of: "press Enter to close this pane")
#expect(closePrompt != nil)
for reset in requiredResets {
let resetRange = transcript.range(of: reset)
#expect(resetRange != nil, Comment(rawValue: transcript))
if let resetRange, let closePrompt {
#expect(resetRange.lowerBound < closePrompt.lowerBound)
}
}
}
@Test func directSignalTerminatesForegroundAuthenticationProcessTree() throws {
let fileManager = FileManager.default
let root = fileManager.temporaryDirectory
@@ -141,7 +210,9 @@ struct SSHStartupManualReconnectTests {
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: executable.path)
}
let startupCommand = try Self.generatedPersistentSSHForegroundAuthenticationStartupCommand()
let startupCommand = try Self.generatedPersistentSSHForegroundAuthenticationStartupCommand(
replacingSystemSSHWith: fakeSSH
)
var environment = ProcessInfo.processInfo.environment
environment["PATH"] = "\(root.path):\(environment["PATH"] ?? "/usr/bin:/bin")"
environment["CMUX_BUNDLED_CLI_PATH"] = fakeCLI.path
@@ -194,7 +265,9 @@ struct SSHStartupManualReconnectTests {
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: executable.path)
}
let startupCommand = try Self.generatedPersistentSSHForegroundAuthenticationStartupCommand()
let startupCommand = try Self.generatedPersistentSSHForegroundAuthenticationStartupCommand(
replacingSystemSSHWith: fakeSSH
)
var environment = ProcessInfo.processInfo.environment
environment["PATH"] = "\(root.path):\(environment["PATH"] ?? "/usr/bin:/bin")"
environment["CMUX_BUNDLED_CLI_PATH"] = fakeCLI.path
@@ -274,7 +347,9 @@ struct SSHStartupManualReconnectTests {
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: executable.path)
}
let startupCommand = try Self.generatedPersistentSSHForegroundAuthenticationStartupCommand()
let startupCommand = try Self.generatedPersistentSSHForegroundAuthenticationStartupCommand(
replacingSystemSSHWith: fakeSSH
)
var environment = ProcessInfo.processInfo.environment
environment["PATH"] = "\(root.path):\(environment["PATH"] ?? "/usr/bin:/bin")"
environment["CMUX_BUNDLED_CLI_PATH"] = fakeCLI.path
@@ -356,7 +431,9 @@ struct SSHStartupManualReconnectTests {
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: executable.path)
}
let startupCommand = try Self.generatedPersistentSSHForegroundAuthenticationStartupCommand()
let startupCommand = try Self.generatedPersistentSSHForegroundAuthenticationStartupCommand(
replacingSystemSSHWith: fakeSSH
)
var environment = ProcessInfo.processInfo.environment
environment["PATH"] = "\(root.path):\(environment["PATH"] ?? "/usr/bin:/bin")"
environment["CMUX_BUNDLED_CLI_PATH"] = fakeCLI.path
@@ -408,7 +485,9 @@ struct SSHStartupManualReconnectTests {
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: executable.path)
}
let startupCommand = try Self.generatedPersistentSSHForegroundAuthenticationStartupCommand()
let startupCommand = try Self.generatedPersistentSSHForegroundAuthenticationStartupCommand(
replacingSystemSSHWith: fakeSSH
)
var environment = ProcessInfo.processInfo.environment
environment["PATH"] = "\(root.path):\(environment["PATH"] ?? "/usr/bin:/bin")"
environment["CMUX_BUNDLED_CLI_PATH"] = fakeCLI.path
@@ -503,7 +582,9 @@ struct SSHStartupManualReconnectTests {
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: executable.path)
}
let startupCommand = try Self.generatedPersistentSSHForegroundAuthenticationStartupCommand()
let startupCommand = try Self.generatedPersistentSSHForegroundAuthenticationStartupCommand(
replacingSystemSSHWith: fakeSSH
)
var environment = ProcessInfo.processInfo.environment
environment["PATH"] = "\(root.path):\(environment["PATH"] ?? "/usr/bin:/bin")"
environment["CMUX_BUNDLED_CLI_PATH"] = fakeCLI.path
@@ -686,7 +767,9 @@ struct SSHStartupManualReconnectTests {
)
}
private static func generatedPersistentSSHForegroundAuthenticationStartupCommand() throws -> String {
private static func generatedPersistentSSHForegroundAuthenticationStartupCommand(
replacingSystemSSHWith fakeSSH: URL
) throws -> String {
let cliPath = try BundledCLITestSupport.bundledCLIPath(for: BundleToken.self)
let socketPath = makeSocketPath("ssh-foreground-auth")
let listenerFD = try bindUnixSocket(at: socketPath)
@@ -769,14 +852,12 @@ struct SSHStartupManualReconnectTests {
)
let configureParams = try #require(configureRequest["params"] as? [String: Any])
let startupCommand = try #require(configureParams["terminal_startup_command"] as? String)
#expect(
startupCommand.contains("cmux_ssh_foreground_auth"),
"Expected the persistent SSH foreground-auth startup path: \(startupCommand)"
)
return startupCommand
return try rewritingSystemSSH(in: startupCommand, with: fakeSSH)
}
private static func generatedVMSSHInitialStartupCommand() throws -> String {
private static func generatedVMSSHInitialStartupCommand(
replacingSystemSSHWith fakeSSH: URL
) throws -> String {
let cliPath = try BundledCLITestSupport.bundledCLIPath(for: BundleToken.self)
let socketPath = makeSocketPath("vm-ssh-startup")
let listenerFD = try bindUnixSocket(at: socketPath)
@@ -798,10 +879,9 @@ struct SSHStartupManualReconnectTests {
}
switch method {
case "vm.attach_info":
case "vm.ssh_info":
let params = payload["params"] as? [String: Any] ?? [:]
guard params["id"] as? String == vmID,
params["require_daemon"] as? Bool == true else {
guard params["id"] as? String == vmID else {
return v2Response(id: id, ok: false, error: ["code": "invalid_params", "message": "unexpected attach params"])
}
return v2Response(
@@ -864,7 +944,58 @@ struct SSHStartupManualReconnectTests {
requests.first { ($0["method"] as? String) == "workspace.create" }
)
let createParams = try #require(createRequest["params"] as? [String: Any])
return try #require(createParams["initial_command"] as? String)
let startupCommand = try #require(createParams["initial_command"] as? String)
return try rewritingSystemSSH(in: startupCommand, with: fakeSSH)
}
private static func rewritingSystemSSH(
in startupCommand: String,
with fakeSSH: URL
) throws -> String {
let systemSSHPath = "/usr/bin/ssh"
let commandURL = URL(
fileURLWithPath: startupCommand.trimmingCharacters(in: .whitespacesAndNewlines)
)
var isDirectory: ObjCBool = false
if FileManager.default.fileExists(atPath: commandURL.path, isDirectory: &isDirectory),
!isDirectory.boolValue {
let script = try String(contentsOf: commandURL, encoding: .utf8)
try #require(script.contains(systemSSHPath))
try script
.replacingOccurrences(of: systemSSHPath, with: fakeSSH.path)
.write(to: commandURL, atomically: true, encoding: .utf8)
try FileManager.default.setAttributes(
[.posixPermissions: 0o700],
ofItemAtPath: commandURL.path
)
return startupCommand
}
if startupCommand.contains(systemSSHPath) {
return startupCommand.replacingOccurrences(of: systemSSHPath, with: fakeSSH.path)
}
let encodedPrefix = "(printf %s "
let encodedSuffix = " | base64"
let prefixRange = try #require(startupCommand.range(of: encodedPrefix))
let suffixRange = try #require(
startupCommand.range(
of: encodedSuffix,
range: prefixRange.upperBound..<startupCommand.endIndex
)
)
let encodedRange = prefixRange.upperBound..<suffixRange.lowerBound
let encodedScript = String(startupCommand[encodedRange])
let scriptData = try #require(Data(base64Encoded: encodedScript))
let script = try #require(String(data: scriptData, encoding: .utf8))
try #require(script.contains(systemSSHPath))
let rewrittenScript = script.replacingOccurrences(of: systemSSHPath, with: fakeSSH.path)
var rewrittenCommand = startupCommand
rewrittenCommand.replaceSubrange(
encodedRange,
with: Data(rewrittenScript.utf8).base64EncodedString()
)
return rewrittenCommand
}
private static func makeTerminalExitPromptFixture() throws -> TerminalExitPromptFixture {
@@ -888,7 +1019,9 @@ struct SSHStartupManualReconnectTests {
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: executable.path)
}
let startupCommand = try generatedPersistentSSHForegroundAuthenticationStartupCommand()
let startupCommand = try generatedPersistentSSHForegroundAuthenticationStartupCommand(
replacingSystemSSHWith: fakeSSH
)
var environment = ProcessInfo.processInfo.environment
environment["PATH"] = "\(root.path):\(environment["PATH"] ?? "/usr/bin:/bin")"
environment["CMUX_BUNDLED_CLI_PATH"] = fakeCLI.path
+116 -27
View File
@@ -34,7 +34,9 @@ extension CLINotifyProcessIntegrationRegressionTests {
]
for signal in ["HUP", "INT", "TERM"] {
try? fileManager.removeItem(at: logFile)
let startupCommand = try generatedVMSSHInitialStartupCommand()
let startupCommand = try generatedVMSSHInitialStartupCommand(
replacingSystemSSHWith: fakeSSH
)
var environment = ProcessInfo.processInfo.environment
environment["PATH"] = "\(root.path):\(environment["PATH"] ?? "/usr/bin:/bin")"
@@ -105,7 +107,9 @@ extension CLINotifyProcessIntegrationRegressionTests {
try? fileManager.removeItem(at: logFile)
try? fileManager.removeItem(at: childSignalLog)
let startupCommand = try generatedVMSSHInitialStartupCommand()
let startupCommand = try generatedVMSSHInitialStartupCommand(
replacingSystemSSHWith: fakeSSH
)
var environment = ProcessInfo.processInfo.environment
environment["PATH"] = "\(root.path):\(environment["PATH"] ?? "/usr/bin:/bin")"
environment["CMUX_BUNDLED_CLI_PATH"] = fakeCLI.path
@@ -183,7 +187,9 @@ extension CLINotifyProcessIntegrationRegressionTests {
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeSSH.path)
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeSleep.path)
let startupCommand = try generatedSSHStartupCommand()
let startupCommand = try generatedSSHStartupCommand(
replacingSystemSSHWith: fakeSSH
)
var environment = ProcessInfo.processInfo.environment
environment["PATH"] = "\(root.path):\(environment["PATH"] ?? "/usr/bin:/bin")"
environment["CMUX_BUNDLED_CLI_PATH"] = fakeCLI.path
@@ -255,7 +261,9 @@ extension CLINotifyProcessIntegrationRegressionTests {
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeCLI.path)
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeSSH.path)
let startupCommand = try generatedSSHStartupCommand()
let startupCommand = try generatedSSHStartupCommand(
replacingSystemSSHWith: fakeSSH
)
var environment = ProcessInfo.processInfo.environment
environment["PATH"] = "\(root.path):\(environment["PATH"] ?? "/usr/bin:/bin")"
environment["CMUX_BUNDLED_CLI_PATH"] = fakeCLI.path
@@ -327,7 +335,9 @@ extension CLINotifyProcessIntegrationRegressionTests {
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeCLI.path)
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeSSH.path)
let startupCommand = try generatedSSHStartupCommand()
let startupCommand = try generatedSSHStartupCommand(
replacingSystemSSHWith: fakeSSH
)
var environment = ProcessInfo.processInfo.environment
environment["PATH"] = "\(root.path):\(environment["PATH"] ?? "/usr/bin:/bin")"
environment["CMUX_BUNDLED_CLI_PATH"] = fakeCLI.path
@@ -442,11 +452,14 @@ extension CLINotifyProcessIntegrationRegressionTests {
)
}
let startupCommand = try generatedSSHStartupCommand(sshOptions: [
"ControlMaster no",
"ControlPath /tmp/cmux-ssh-%C",
"RequestTTY no",
])
let startupCommand = try generatedSSHStartupCommand(
replacingSystemSSHWith: fakeSSH,
sshOptions: [
"ControlMaster no",
"ControlPath /tmp/cmux-ssh-%C",
"RequestTTY no",
]
)
var environment = ProcessInfo.processInfo.environment
environment["PATH"] = "\(root.path):\(environment["PATH"] ?? "/usr/bin:/bin")"
environment["HOME"] = home.path
@@ -531,6 +544,7 @@ extension CLINotifyProcessIntegrationRegressionTests {
}
let startupCommand = try generatedSSHStartupCommand(
replacingSystemSSHWith: fakeSSH,
remoteCommandArguments: [
"touch",
"\"$CMUX_TEST_RAW_COMMAND_LOG\"",
@@ -621,6 +635,7 @@ extension CLINotifyProcessIntegrationRegressionTests {
echo user-end >> "$CMUX_TEST_RAW_EVENT_LOG"
"""
let startupCommand = try generatedSSHStartupCommand(
replacingSystemSSHWith: fakeSSH,
remoteCommandArguments: [
"/bin/sh",
"-c",
@@ -702,6 +717,7 @@ extension CLINotifyProcessIntegrationRegressionTests {
}
let startupCommand = try generatedSSHStartupCommand(
replacingSystemSSHWith: fakeSSH,
additionalArguments: ["--transport", "mosh"]
)
var environment = ProcessInfo.processInfo.environment
@@ -793,11 +809,14 @@ extension CLINotifyProcessIntegrationRegressionTests {
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeCLI.path)
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeSSH.path)
let startupCommand = try generatedSSHStartupCommand(sshOptions: [
"ControlMaster auto",
"ControlPersist 600",
"ControlPath \(staleControlPath.path)",
])
let startupCommand = try generatedSSHStartupCommand(
replacingSystemSSHWith: fakeSSH,
sshOptions: [
"ControlMaster auto",
"ControlPersist 600",
"ControlPath \(staleControlPath.path)",
]
)
var environment = ProcessInfo.processInfo.environment
environment["PATH"] = "\(root.path):\(environment["PATH"] ?? "/usr/bin:/bin")"
environment["CMUX_BUNDLED_CLI_PATH"] = fakeCLI.path
@@ -865,7 +884,10 @@ extension CLINotifyProcessIntegrationRegressionTests {
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeCLI.path)
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeSSH.path)
let startupCommand = try generatedSSHStartupCommand(sshOptions: sshOptions)
let startupCommand = try generatedSSHStartupCommand(
replacingSystemSSHWith: fakeSSH,
sshOptions: sshOptions
)
var environment = ProcessInfo.processInfo.environment
environment["PATH"] = "\(root.path):\(environment["PATH"] ?? "/usr/bin:/bin")"
environment["CMUX_BUNDLED_CLI_PATH"] = fakeCLI.path
@@ -917,7 +939,9 @@ extension CLINotifyProcessIntegrationRegressionTests {
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeCLI.path)
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeSSH.path)
let startupCommand = try generatedVMSSHInitialStartupCommand()
let startupCommand = try generatedVMSSHInitialStartupCommand(
replacingSystemSSHWith: fakeSSH
)
var environment = ProcessInfo.processInfo.environment
environment["PATH"] = "\(root.path):\(environment["PATH"] ?? "/usr/bin:/bin")"
environment["CMUX_BUNDLED_CLI_PATH"] = fakeCLI.path
@@ -972,7 +996,9 @@ extension CLINotifyProcessIntegrationRegressionTests {
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeCLI.path)
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeSSH.path)
let startupCommand = try generatedVMSSHInitialStartupCommand()
let startupCommand = try generatedVMSSHInitialStartupCommand(
replacingSystemSSHWith: fakeSSH
)
var environment = ProcessInfo.processInfo.environment
environment["PATH"] = "\(root.path):\(environment["PATH"] ?? "/usr/bin:/bin")"
environment["CMUX_BUNDLED_CLI_PATH"] = fakeCLI.path
@@ -1021,7 +1047,9 @@ extension CLINotifyProcessIntegrationRegressionTests {
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeCLI.path)
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeSSH.path)
let startupCommand = try generatedVMSSHInitialStartupCommand()
let startupCommand = try generatedVMSSHInitialStartupCommand(
replacingSystemSSHWith: fakeSSH
)
var environment = ProcessInfo.processInfo.environment
environment["PATH"] = "\(root.path):\(environment["PATH"] ?? "/usr/bin:/bin")"
environment["CMUX_BUNDLED_CLI_PATH"] = fakeCLI.path
@@ -1078,7 +1106,9 @@ extension CLINotifyProcessIntegrationRegressionTests {
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeCLI.path)
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeSSH.path)
let startupCommand = try generatedVMSSHInitialStartupCommand()
let startupCommand = try generatedVMSSHInitialStartupCommand(
replacingSystemSSHWith: fakeSSH
)
var environment = ProcessInfo.processInfo.environment
environment["PATH"] = "\(root.path):\(environment["PATH"] ?? "/usr/bin:/bin")"
environment["CMUX_BUNDLED_CLI_PATH"] = fakeCLI.path
@@ -1129,7 +1159,9 @@ extension CLINotifyProcessIntegrationRegressionTests {
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeCLI.path)
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeSSH.path)
let startupCommand = try generatedVMSSHInitialStartupCommand()
let startupCommand = try generatedVMSSHInitialStartupCommand(
replacingSystemSSHWith: fakeSSH
)
var environment = ProcessInfo.processInfo.environment
environment["PATH"] = "\(root.path):\(environment["PATH"] ?? "/usr/bin:/bin")"
environment["CMUX_BUNDLED_CLI_PATH"] = fakeCLI.path
@@ -1185,7 +1217,9 @@ extension CLINotifyProcessIntegrationRegressionTests {
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeCLI.path)
try fileManager.setAttributes([.posixPermissions: 0o700], ofItemAtPath: fakeSSH.path)
let startupCommand = try generatedSSHStartupCommand()
let startupCommand = try generatedSSHStartupCommand(
replacingSystemSSHWith: fakeSSH
)
var environment = ProcessInfo.processInfo.environment
environment["PATH"] = "\(root.path):\(environment["PATH"] ?? "/usr/bin:/bin")"
environment["CMUX_BUNDLED_CLI_PATH"] = fakeCLI.path
@@ -1216,6 +1250,7 @@ extension CLINotifyProcessIntegrationRegressionTests {
}
private func generatedSSHStartupCommand(
replacingSystemSSHWith fakeSSH: URL,
sshOptions: [String] = [
"ControlMaster no",
"ControlPath /tmp/cmux-ssh-%C",
@@ -1319,10 +1354,13 @@ extension CLINotifyProcessIntegrationRegressionTests {
requests.first { ($0["method"] as? String) == "workspace.remote.configure" }
)
let configureParams = try XCTUnwrap(configureRequest["params"] as? [String: Any])
return try XCTUnwrap(configureParams["terminal_startup_command"] as? String)
let startupCommand = try XCTUnwrap(configureParams["terminal_startup_command"] as? String)
return try rewritingSystemSSH(in: startupCommand, with: fakeSSH)
}
private func generatedVMSSHInitialStartupCommand() throws -> String {
private func generatedVMSSHInitialStartupCommand(
replacingSystemSSHWith fakeSSH: URL
) throws -> String {
let cliPath = try bundledCLIPath()
let socketPath = makeSocketPath("vm-ssh-startup")
let listenerFD = try bindUnixSocket(at: socketPath)
@@ -1344,10 +1382,9 @@ extension CLINotifyProcessIntegrationRegressionTests {
}
switch method {
case "vm.attach_info":
case "vm.ssh_info":
let params = payload["params"] as? [String: Any] ?? [:]
XCTAssertEqual(params["id"] as? String, vmID)
XCTAssertEqual(params["require_daemon"] as? Bool, true)
return self.v2Response(
id: id,
ok: true,
@@ -1421,7 +1458,59 @@ extension CLINotifyProcessIntegrationRegressionTests {
requests.first { ($0["method"] as? String) == "workspace.create" }
)
let createParams = try XCTUnwrap(createRequest["params"] as? [String: Any])
return try XCTUnwrap(createParams["initial_command"] as? String)
let startupCommand = try XCTUnwrap(createParams["initial_command"] as? String)
return try rewritingSystemSSH(in: startupCommand, with: fakeSSH)
}
private func rewritingSystemSSH(
in startupCommand: String,
with fakeSSH: URL
) throws -> String {
let systemSSHPath = "/usr/bin/ssh"
let commandURL = URL(
fileURLWithPath: startupCommand.trimmingCharacters(in: .whitespacesAndNewlines)
)
var isDirectory: ObjCBool = false
if FileManager.default.fileExists(atPath: commandURL.path, isDirectory: &isDirectory),
!isDirectory.boolValue {
let script = try String(contentsOf: commandURL, encoding: .utf8)
XCTAssertTrue(script.contains(systemSSHPath), script)
try script
.replacingOccurrences(of: systemSSHPath, with: fakeSSH.path)
.write(to: commandURL, atomically: true, encoding: .utf8)
try FileManager.default.setAttributes(
[.posixPermissions: 0o700],
ofItemAtPath: commandURL.path
)
return startupCommand
}
if startupCommand.contains(systemSSHPath) {
return startupCommand.replacingOccurrences(of: systemSSHPath, with: fakeSSH.path)
}
let encodedPrefix = "(printf %s "
let encodedSuffix = " | base64"
guard let prefixRange = startupCommand.range(of: encodedPrefix),
let suffixRange = startupCommand.range(
of: encodedSuffix,
range: prefixRange.upperBound..<startupCommand.endIndex
) else {
XCTFail("Generated startup command did not pin \(systemSSHPath): \(startupCommand)")
return startupCommand
}
let encodedRange = prefixRange.upperBound..<suffixRange.lowerBound
let encodedScript = String(startupCommand[encodedRange])
let scriptData = try XCTUnwrap(Data(base64Encoded: encodedScript))
let script = try XCTUnwrap(String(data: scriptData, encoding: .utf8))
XCTAssertTrue(script.contains(systemSSHPath), script)
let rewrittenScript = script.replacingOccurrences(of: systemSSHPath, with: fakeSSH.path)
var rewrittenCommand = startupCommand
rewrittenCommand.replaceSubrange(
encodedRange,
with: Data(rewrittenScript.utf8).base64EncodedString()
)
return rewrittenCommand
}
private func writeShellFile(at url: URL, lines: [String]) throws {
+15
View File
@@ -70,6 +70,21 @@ PTY lifecycle:
5. Sessions with no attachments keep their last-known size and are reaped by the daemon idle TTL.
6. Closing the owning workspace sends an authenticated slot-shutdown request, waits a bounded interval for the daemon lock to be released, and removes the relay's shell-state directory. As defense in depth, a daemon launched with `--persistent-lease-port` observes that exact `~/.cmux/relay/<port>.slot` lease, but retires passively only after the observed lease disappears and both stdio connections and live PTY sessions are empty. A detached live PTY survives lease loss until it exits or is closed explicitly. Older callers that omit the flag retain the prior behavior without unsafe broad lease scanning.
### Persistent daemon diagnostics
Persistent-daemon logging is always enabled. The current log is
`~/.cmux/daemon/<version>/<slot>/daemon.log`, mode `0600`. It records daemon
start/readiness/stop, authenticated connection lifecycle, PTY attach/detach/
close/exit, channel or PTY-operation faults, and process-level stdout/stderr
occurrence and rate-limited aggregate byte counts. Arbitrary process output is
discarded rather than persisted. Faults are recorded as bounded codes or
categories rather than raw error text. Tokens, commands, terminal input, RPC
request identifiers, and raw process output are never logged.
The log rotates at 2 MiB. `daemon.log` is the newest file, with at most two
older generations in `daemon.log.1` and `daemon.log.2`, so one slot uses at
most approximately 6 MiB of diagnostics.
## Cloud WebSocket PTY transport
The WebSocket PTY transport is locked until the backend writes a short-lived
+152 -17
View File
@@ -1073,7 +1073,7 @@ func waitPersistentDaemonReady(reader *os.File, logFile string) error {
}
}
func runPersistentDaemonServer(slot string, leasePort int, stderr io.Writer) error {
func runPersistentDaemonServer(slot string, leasePort int, stderr io.Writer) (resultErr error) {
paths, err := persistentDaemonPathsForSlot(slot)
if err != nil {
return err
@@ -1096,6 +1096,44 @@ func runPersistentDaemonServer(slot string, leasePort int, stderr io.Writer) err
}
defer syscall.Flock(int(lockFile.Fd()), syscall.LOCK_UN)
daemonLog, err := openPersistentDaemonLog(paths.logFile)
if err != nil {
return fmt.Errorf("open persistent daemon log: %w", err)
}
routeProcessOutput := shouldRoutePersistentDaemonProcessOutput(stderr)
stderr = daemonLog
var processOutputRoute *persistentDaemonProcessOutputRoute
defer func() {
if processOutputRoute != nil {
if closeErr := processOutputRoute.Close(); closeErr != nil && resultErr == nil {
resultErr = fmt.Errorf("restore persistent daemon process output: %w", closeErr)
}
}
fields := []string{"status", "clean"}
if resultErr != nil {
fields = []string{
"status", "error",
"error_category", persistentDaemonErrorCategory(resultErr),
}
}
logPersistentDaemonEvent(stderr, "daemon_stop", fields...)
_ = daemonLog.Close()
}()
if routeProcessOutput {
processOutputRoute, err = routePersistentDaemonProcessOutput(daemonLog)
if err != nil {
return fmt.Errorf("route persistent daemon process output: %w", err)
}
}
logPersistentDaemonEvent(
stderr,
"daemon_start",
"version", version,
"slot", paths.slot,
"lease_port", strconv.Itoa(leasePort),
"pid", strconv.Itoa(os.Getpid()),
)
_ = os.Remove(paths.socket)
listener, err := net.Listen("unix", paths.socket)
if err != nil {
@@ -1104,6 +1142,7 @@ func runPersistentDaemonServer(slot string, leasePort int, stderr io.Writer) err
defer listener.Close()
defer os.Remove(paths.socket)
_ = os.Chmod(paths.socket, 0o600)
logPersistentDaemonEvent(stderr, "daemon_ready", "socket", paths.socket)
signalPersistentDaemonReady()
config := persistentDaemonServerConfig{emptyIdleTimeout: persistentDaemonEmptyIdleTimeout}
@@ -1344,27 +1383,67 @@ func handlePersistentDaemonConnWithAuthTimeout(
requestShutdown func(),
) {
defer conn.Close()
defer logPersistentDaemonEvent(stderr, "connection_closed")
logPersistentDaemonEvent(stderr, "connection_accepted")
if timeout > 0 {
if err := conn.SetDeadline(time.Now().Add(timeout)); err != nil {
logPersistentDaemonEvent(
stderr,
"connection_fault",
"phase", "auth_deadline",
"error_category", persistentDaemonErrorCategory(err),
)
return
}
}
reader := bufio.NewReaderSize(conn, 64*1024)
writer := &stdioFrameWriter{writer: bufio.NewWriter(conn)}
if err := authenticatePersistentDaemonConn(reader, writer, verifier); err != nil {
if stderr != nil {
_, _ = fmt.Fprintf(stderr, "persistent daemon connection rejected: %v\n", err)
}
logPersistentDaemonEvent(
stderr,
"connection_rejected",
"reason", persistentDaemonAuthenticationFailureReason(err),
)
return
}
logPersistentDaemonEvent(stderr, "connection_authenticated")
if timeout > 0 {
if err := conn.SetDeadline(time.Time{}); err != nil {
logPersistentDaemonEvent(
stderr,
"connection_fault",
"phase", "clear_auth_deadline",
"error_category", persistentDaemonErrorCategory(err),
)
return
}
}
_ = runRPCServerWithReader(reader, writer, hub, false, requestShutdown, func() {
if err := runRPCServerWithReader(reader, writer, hub, false, requestShutdown, func() {
_ = conn.Close()
})
}); err != nil {
logPersistentDaemonEvent(
stderr,
"connection_fault",
"phase", "rpc",
"error_category", persistentDaemonErrorCategory(err),
)
}
}
func persistentDaemonAuthenticationFailureReason(err error) string {
if err == nil {
return "unknown"
}
switch err.Error() {
case "authentication frame exceeds size limit",
"authentication frame is invalid JSON",
"authentication method is missing",
"authentication method is invalid",
"authentication token is invalid":
return err.Error()
default:
return persistentDaemonErrorCategory(err)
}
}
func authenticatePersistentDaemonConn(reader *bufio.Reader, writer *stdioFrameWriter, verifier persistentDaemonTokenVerifier) error {
@@ -1811,18 +1890,16 @@ func (s *rpcServer) handleNotificationResponse(req rpcRequest, resp rpcResponse)
if !rpcRequestIsPTYAttachmentNotification(req) || resp.OK {
return nil
}
errorCode := "unknown"
if resp.Error != nil {
errorCode = persistentDaemonDiagnosticCode(resp.Error.Code)
}
if s.frameWriter == nil {
detail := "unknown error"
if resp.Error != nil {
detail = strings.TrimSpace(resp.Error.Code)
if message := strings.TrimSpace(resp.Error.Message); message != "" {
if detail != "" {
detail += ": "
}
detail += message
}
}
_, _ = fmt.Fprintf(os.Stderr, "cmuxd-remote: %s notification failed without response writer: %s\n", req.Method, detail)
s.logPTYEvent(
"pty_notification_fault",
"operation", persistentDaemonDiagnosticCode(req.Method),
"error_code", errorCode,
)
return nil
}
sessionID, attachmentID, attachmentToken, badResp := parsePTYAttachmentIdentity(req, req.Method)
@@ -1838,6 +1915,13 @@ func (s *rpcServer) handleNotificationResponse(req rpcRequest, resp rpcResponse)
if resp.Error != nil && strings.TrimSpace(resp.Error.Message) != "" {
detail = strings.TrimSpace(resp.Error.Message)
}
s.logPTYEvent(
"pty_channel_fault",
"session_id", sessionID,
"attachment_id", attachmentID,
"operation", req.Method,
"error_code", errorCode,
)
err := s.frameWriter.writeEvent(rpcEvent{
Event: "pty.error",
SessionID: sessionID,
@@ -2448,6 +2532,14 @@ func (s *rpcServer) handlePTYAttachContextWithReservation(
notifyReservationReady,
)
if err != nil {
s.logPTYEvent(
"pty_attach_failed",
"session_id", strings.TrimSpace(sessionID),
"attachment_id", attachmentID,
"require_existing", strconv.FormatBool(requireExisting),
"error_code", ptyAttachErrorCode(err, requireExisting),
"error_category", persistentDaemonErrorCategory(err),
)
return rpcResponse{
ID: req.ID,
OK: false,
@@ -2459,6 +2551,14 @@ func (s *rpcServer) handlePTYAttachContextWithReservation(
}
if err := operationCtx.Err(); err != nil {
hub.dropAttachment(attachment)
s.logPTYEvent(
"pty_attach_failed",
"session_id", strings.TrimSpace(sessionID),
"attachment_id", attachmentID,
"require_existing", strconv.FormatBool(requireExisting),
"error_code", ptyAttachErrorCode(err, requireExisting),
"error_category", persistentDaemonErrorCategory(err),
)
return rpcResponse{
ID: req.ID,
OK: false,
@@ -2470,6 +2570,14 @@ func (s *rpcServer) handlePTYAttachContextWithReservation(
}
if !s.trackPTYAttachment(attachment) {
hub.dropAttachment(attachment)
s.logPTYEvent(
"pty_attach_failed",
"session_id", strings.TrimSpace(sessionID),
"attachment_id", attachmentID,
"require_existing", strconv.FormatBool(requireExisting),
"error_code", ptyAttachErrorCode(nil, requireExisting),
"error_category", "connection_closed",
)
return rpcResponse{
ID: req.ID,
OK: false,
@@ -2717,6 +2825,12 @@ func (s *rpcServer) ptyAttachmentPump(ctx context.Context, attachment *wsPTYAtta
for {
select {
case <-ctx.Done():
s.logPTYEvent(
"pty_channel_closed",
"session_id", attachment.sessionKey.sessionID,
"attachment_id", attachment.id,
"reason", "connection_closed",
)
if s.ptyHub != nil {
s.ptyHub.dropAttachment(attachment)
}
@@ -2727,6 +2841,13 @@ func (s *rpcServer) ptyAttachmentPump(ctx context.Context, attachment *wsPTYAtta
select {
case frame := <-attachment.send:
if err := s.frameWriter.writeEvent(rpcPTYEventForFrame(attachment, frame)); err != nil {
s.logPTYEvent(
"pty_channel_fault",
"session_id", attachment.sessionKey.sessionID,
"attachment_id", attachment.id,
"direction", "daemon_to_client",
"error_category", persistentDaemonErrorCategory(err),
)
if s.ptyHub != nil {
s.ptyHub.dropAttachment(attachment)
}
@@ -2739,6 +2860,13 @@ func (s *rpcServer) ptyAttachmentPump(ctx context.Context, attachment *wsPTYAtta
}
case frame := <-attachment.send:
if err := s.frameWriter.writeEvent(rpcPTYEventForFrame(attachment, frame)); err != nil {
s.logPTYEvent(
"pty_channel_fault",
"session_id", attachment.sessionKey.sessionID,
"attachment_id", attachment.id,
"direction", "daemon_to_client",
"error_category", persistentDaemonErrorCategory(err),
)
if s.ptyHub != nil {
s.ptyHub.dropAttachment(attachment)
}
@@ -2748,6 +2876,13 @@ func (s *rpcServer) ptyAttachmentPump(ctx context.Context, attachment *wsPTYAtta
}
}
func (s *rpcServer) logPTYEvent(event string, fields ...string) {
if s == nil || s.ptyHub == nil {
return
}
logPersistentDaemonEvent(s.ptyHub.stderr, event, fields...)
}
func (s *rpcServer) trackPTYAttachment(attachment *wsPTYAttachment) bool {
if attachment == nil {
return false
@@ -0,0 +1,233 @@
package main
import (
"context"
"errors"
"fmt"
"io"
"net"
"os"
"strconv"
"strings"
"sync"
"syscall"
"time"
)
const (
persistentDaemonLogMaxBytes = int64(2 * 1024 * 1024)
persistentDaemonLogBackups = 2
)
type persistentDaemonLog struct {
mu sync.Mutex
path string
maxBytes int64
backups int
file *os.File
size int64
}
func openPersistentDaemonLog(path string) (*persistentDaemonLog, error) {
return openPersistentDaemonLogWithLimit(
path,
persistentDaemonLogMaxBytes,
persistentDaemonLogBackups,
)
}
func openPersistentDaemonLogWithLimit(path string, maxBytes int64, backups int) (*persistentDaemonLog, error) {
if maxBytes <= 0 {
return nil, errors.New("persistent daemon log size limit must be positive")
}
if backups < 0 {
return nil, errors.New("persistent daemon log backup count must not be negative")
}
logFile := &persistentDaemonLog{
path: path,
maxBytes: maxBytes,
backups: backups,
}
if err := logFile.openLocked(); err != nil {
return nil, err
}
return logFile, nil
}
func (l *persistentDaemonLog) Write(payload []byte) (int, error) {
l.mu.Lock()
defer l.mu.Unlock()
if l.file == nil {
return 0, os.ErrClosed
}
if len(payload) == 0 {
return 0, nil
}
originalLength := len(payload)
if int64(len(payload)) > l.maxBytes {
payload = payload[len(payload)-int(l.maxBytes):]
}
if l.size > 0 && l.size+int64(len(payload)) > l.maxBytes {
if err := l.rotateLocked(); err != nil {
return 0, err
}
}
written, err := l.file.Write(payload)
l.size += int64(written)
if err != nil {
return written, err
}
if written != len(payload) {
return written, io.ErrShortWrite
}
return originalLength, nil
}
func (l *persistentDaemonLog) Close() error {
l.mu.Lock()
defer l.mu.Unlock()
if l.file == nil {
return nil
}
err := l.file.Close()
l.file = nil
return err
}
func (l *persistentDaemonLog) openLocked() error {
file, err := os.OpenFile(l.path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
if err != nil {
return err
}
if err := file.Chmod(0o600); err != nil {
_ = file.Close()
return err
}
info, err := file.Stat()
if err != nil {
_ = file.Close()
return err
}
l.file = file
l.size = info.Size()
return nil
}
func (l *persistentDaemonLog) rotateLocked() error {
if err := l.file.Close(); err != nil {
return err
}
l.file = nil
if l.backups == 0 {
if err := os.Remove(l.path); err != nil && !errors.Is(err, os.ErrNotExist) {
return err
}
} else {
oldest := persistentDaemonLogBackupPath(l.path, l.backups)
if err := os.Remove(oldest); err != nil && !errors.Is(err, os.ErrNotExist) {
return err
}
for index := l.backups - 1; index >= 1; index-- {
from := persistentDaemonLogBackupPath(l.path, index)
to := persistentDaemonLogBackupPath(l.path, index+1)
if err := os.Rename(from, to); err != nil && !errors.Is(err, os.ErrNotExist) {
return err
}
}
if err := os.Rename(l.path, persistentDaemonLogBackupPath(l.path, 1)); err != nil &&
!errors.Is(err, os.ErrNotExist) {
return err
}
}
return l.openLocked()
}
func persistentDaemonLogBackupPath(path string, index int) string {
return path + "." + strconv.Itoa(index)
}
func logPersistentDaemonEvent(writer io.Writer, event string, fields ...string) {
if writer == nil {
return
}
var line strings.Builder
line.WriteString("time=")
line.WriteString(time.Now().UTC().Format(time.RFC3339Nano))
line.WriteString(" event=")
line.WriteString(event)
for index := 0; index+1 < len(fields); index += 2 {
line.WriteByte(' ')
line.WriteString(fields[index])
line.WriteByte('=')
line.WriteString(strconv.Quote(fields[index+1]))
}
line.WriteByte('\n')
_, _ = fmt.Fprint(writer, line.String())
}
func persistentDaemonErrorCategory(err error) string {
if err == nil {
return "none"
}
if errors.Is(err, context.Canceled) {
return "canceled"
}
if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, os.ErrDeadlineExceeded) {
return "timeout"
}
var netErr net.Error
if errors.As(err, &netErr) && netErr.Timeout() {
return "timeout"
}
if errors.Is(err, io.EOF) {
return "eof"
}
if errors.Is(err, net.ErrClosed) ||
errors.Is(err, os.ErrClosed) ||
errors.Is(err, io.ErrClosedPipe) ||
errors.Is(err, syscall.EPIPE) {
return "connection_closed"
}
if errors.Is(err, syscall.ECONNRESET) {
return "connection_reset"
}
if errors.Is(err, syscall.ECONNREFUSED) {
return "connection_refused"
}
if errors.Is(err, os.ErrNotExist) || errors.Is(err, syscall.ENOENT) {
return "not_found"
}
if errors.Is(err, os.ErrPermission) ||
errors.Is(err, syscall.EACCES) ||
errors.Is(err, syscall.EPERM) {
return "permission_denied"
}
if errors.Is(err, syscall.ENOSPC) ||
errors.Is(err, syscall.EMFILE) ||
errors.Is(err, syscall.ENFILE) ||
errors.Is(err, syscall.ENOMEM) {
return "resource_exhausted"
}
if errors.Is(err, syscall.EIO) {
return "io_error"
}
return "unexpected"
}
func persistentDaemonDiagnosticCode(value string) string {
value = strings.TrimSpace(value)
if value == "" || len(value) > 64 {
return "unknown"
}
for _, character := range value {
if (character >= 'a' && character <= 'z') ||
(character >= '0' && character <= '9') ||
character == '_' || character == '-' || character == '.' {
continue
}
return "unknown"
}
return value
}
@@ -0,0 +1,408 @@
package main
import (
"bytes"
"encoding/base64"
"errors"
"fmt"
"os"
"os/exec"
"path/filepath"
"strconv"
"strings"
"syscall"
"testing"
"time"
)
func TestPersistentDaemonLogsConnectionAndPTYLifecycle(t *testing.T) {
const firstAttachmentToken = "secret-token-must-not-be-logged"
const secondAttachmentToken = "second-secret-token-must-not-be-logged"
const firstCommand = "sleep 60"
const secondCommand = "exit 0"
const terminalInput = "terminal-input-must-not-be-logged"
const requestID = "request-id-must-not-be-logged"
logOutput := newNotifyingBuffer()
socketPath, stop := startPersistentDaemonWithVerifierAndLogForTest(
t,
persistentDaemonFixedTokenVerifier("lifecycle-log-token"),
logOutput,
)
defer stop()
conn, reader, writer := openPersistentTestClient(t, socketPath, "lifecycle-log-token")
defer conn.Close()
attach := persistentTestRPCCall(t, conn, reader, writer, rpcRequest{
ID: requestID + "-attach",
Method: "pty.attach",
Params: map[string]any{
"session_id": "logged-session",
"attachment_id": "logged-attachment",
"client_attachment_token": firstAttachmentToken,
"cols": 80,
"rows": 24,
"command": firstCommand,
},
})
if ok, _ := attach["ok"].(bool); !ok {
t.Fatalf("pty.attach failed: %v", attach)
}
readPersistentTestEvent(t, conn, reader, func(frame map[string]any) bool {
return frame["event"] == "pty.ready" && frame["attachment_id"] == "logged-attachment"
})
writeResponse := persistentTestRPCCall(t, conn, reader, writer, rpcRequest{
ID: requestID + "-write",
Method: "pty.write",
Params: map[string]any{
"session_id": "logged-session",
"attachment_id": "logged-attachment",
"client_attachment_token": firstAttachmentToken,
"data_base64": base64.StdEncoding.EncodeToString([]byte(terminalInput)),
},
})
if ok, _ := writeResponse["ok"].(bool); !ok {
t.Fatalf("pty.write failed: %v", writeResponse)
}
detach := persistentTestRPCCall(t, conn, reader, writer, rpcRequest{
ID: requestID + "-detach",
Method: "pty.detach",
Params: map[string]any{
"session_id": "logged-session",
"attachment_id": "logged-attachment",
"client_attachment_token": firstAttachmentToken,
},
})
if ok, _ := detach["ok"].(bool); !ok {
t.Fatalf("pty.detach failed: %v", detach)
}
closeResponse := persistentTestRPCCall(t, conn, reader, writer, rpcRequest{
ID: requestID + "-close",
Method: "pty.close",
Params: map[string]any{
"session_id": "logged-session",
},
})
if ok, _ := closeResponse["ok"].(bool); !ok {
t.Fatalf("pty.close failed: %v", closeResponse)
}
exitingAttach := persistentTestRPCCall(t, conn, reader, writer, rpcRequest{
ID: requestID + "-exit",
Method: "pty.attach",
Params: map[string]any{
"session_id": "logged-exit-session",
"attachment_id": "logged-exit-attachment",
"client_attachment_token": secondAttachmentToken,
"cols": 80,
"rows": 24,
"command": secondCommand,
},
})
if ok, _ := exitingAttach["ok"].(bool); !ok {
t.Fatalf("short-lived pty.attach failed: %v", exitingAttach)
}
readPersistentTestEvent(t, conn, reader, func(frame map[string]any) bool {
return frame["event"] == "pty.exit" && frame["session_id"] == "logged-exit-session"
})
logged := logOutput.String()
for _, event := range []string{
"event=connection_accepted",
"event=connection_authenticated",
"event=pty_attach",
"event=pty_detach",
"event=pty_close",
"event=pty_exit",
} {
if !strings.Contains(logged, event) {
t.Fatalf("persistent daemon log = %q, want %q", logged, event)
}
}
for _, secret := range []string{
firstAttachmentToken,
secondAttachmentToken,
firstCommand,
secondCommand,
terminalInput,
requestID,
} {
if strings.Contains(logged, secret) {
t.Fatalf("persistent daemon log exposed sensitive request data %q: %q", secret, logged)
}
}
}
func TestPersistentDaemonLogRotationIsSizeBounded(t *testing.T) {
const maxBytes = int64(220)
const backups = 2
logPath := filepath.Join(t.TempDir(), "daemon.log")
logOutput, err := openPersistentDaemonLogWithLimit(logPath, maxBytes, backups)
if err != nil {
t.Fatalf("open persistent daemon log: %v", err)
}
for index := 0; index < 12; index++ {
logPersistentDaemonEvent(
logOutput,
"rotation_probe",
"marker", fmt.Sprintf("event-%02d-%s", index, strings.Repeat("x", 32)),
)
}
if err := logOutput.Close(); err != nil {
t.Fatalf("close persistent daemon log: %v", err)
}
for index, path := range []string{
logPath,
persistentDaemonLogBackupPath(logPath, 1),
persistentDaemonLogBackupPath(logPath, 2),
} {
info, err := os.Stat(path)
if err != nil {
t.Fatalf("stat log generation %d: %v", index, err)
}
if info.Size() > maxBytes {
t.Fatalf("log generation %d size = %d, want <= %d", index, info.Size(), maxBytes)
}
if info.Mode().Perm() != 0o600 {
t.Fatalf("log generation %d mode = %o, want 600", index, info.Mode().Perm())
}
}
newest, err := os.ReadFile(logPath)
if err != nil {
t.Fatalf("read newest persistent daemon log: %v", err)
}
if !strings.Contains(string(newest), "event-11-") {
t.Fatalf("newest persistent daemon log lost the latest event: %q", string(newest))
}
}
func TestPersistentDaemonFaultLogsExcludeRawRequestDetails(t *testing.T) {
const attachmentToken = "fault-token-must-not-be-logged"
const command = "fault-command-must-not-be-logged"
const terminalInput = "fault-input-must-not-be-logged"
const requestID = "fault-request-id-must-not-be-logged"
const rawFailure = "raw-failure-detail-must-not-be-logged"
logOutput := newNotifyingBuffer()
hub := newWebSocketPTYHub(wsPTYServerConfig{Shell: "/bin/sh"}, logOutput)
t.Cleanup(hub.closeAll)
hub.openPTY = func() (*os.File, *os.File, error) {
return nil, nil, errors.New(rawFailure)
}
writer := &captureRPCFrameWriter{}
server := &rpcServer{ptyHub: hub, frameWriter: writer}
attachResponse := server.handleRequest(rpcRequest{
ID: requestID + "-attach",
Method: "pty.attach",
Params: map[string]any{
"session_id": "fault-session",
"attachment_id": "fault-attachment",
"client_attachment_token": attachmentToken,
"cols": 80,
"rows": 24,
"command": command,
},
})
if attachResponse.OK {
t.Fatalf("pty.attach unexpectedly succeeded: %+v", attachResponse)
}
notification := rpcRequest{
ID: requestID + "-write",
Method: "pty.write",
Params: map[string]any{
"session_id": "fault-session",
"attachment_id": "fault-attachment",
"client_attachment_token": attachmentToken,
"data_base64": base64.StdEncoding.EncodeToString([]byte(terminalInput)),
},
}
if err := server.handleNotificationResponse(notification, rpcResponse{
OK: false,
Error: &rpcError{
Code: "pty_input_queue_full",
Message: rawFailure,
},
}); err != nil {
t.Fatalf("handle notification failure: %v", err)
}
logged := logOutput.String()
for _, event := range []string{
"event=pty_start_fault",
"event=pty_attach_failed",
"event=pty_channel_fault",
} {
if !strings.Contains(logged, event) {
t.Fatalf("persistent daemon log = %q, want %q", logged, event)
}
}
for _, secret := range []string{
attachmentToken,
command,
terminalInput,
requestID,
rawFailure,
} {
if strings.Contains(logged, secret) {
t.Fatalf("persistent daemon fault log exposed request data %q: %q", secret, logged)
}
}
}
func TestPersistentDaemonProcessOutputUsesRotatingWriter(t *testing.T) {
const helperEnvironment = "CMUX_TEST_PERSISTENT_PROCESS_OUTPUT"
const pathEnvironment = "CMUX_TEST_PERSISTENT_PROCESS_LOG_PATH"
const stdoutSecret = "process-stdout-secret-must-not-be-logged"
const stderrSecret = "process-stderr-secret-must-not-be-logged"
if os.Getenv(helperEnvironment) == "1" {
logOutput, err := openPersistentDaemonLogWithLimit(os.Getenv(pathEnvironment), 512, 2)
if err != nil {
t.Fatalf("open helper process log: %v", err)
}
route, err := routePersistentDaemonProcessOutput(logOutput)
if err != nil {
t.Fatalf("route helper process output: %v", err)
}
_, _ = fmt.Fprintln(os.Stdout, stdoutSecret)
_, _ = fmt.Fprintln(os.Stderr, stderrSecret)
if err := route.Close(); err != nil {
t.Fatalf("close helper process output route: %v", err)
}
if err := logOutput.Close(); err != nil {
t.Fatalf("close helper process log: %v", err)
}
return
}
logPath := filepath.Join(t.TempDir(), "daemon.log")
command := exec.Command(os.Args[0], "-test.run=^TestPersistentDaemonProcessOutputUsesRotatingWriter$")
command.Env = append(
os.Environ(),
helperEnvironment+"=1",
pathEnvironment+"="+logPath,
)
var output bytes.Buffer
command.Stdout = &output
command.Stderr = &output
if err := command.Run(); err != nil {
t.Fatalf("process-output helper failed: %v; output=%q", err, output.String())
}
logged, err := os.ReadFile(logPath)
if err != nil {
t.Fatalf("read process-output log: %v", err)
}
for _, secret := range []string{stdoutSecret, stderrSecret} {
if strings.Contains(string(logged), secret) {
t.Fatalf("process output exposed sensitive content %q: %q", secret, string(logged))
}
}
processOutputBytes := map[string][]int64{}
for _, line := range strings.Split(string(logged), "\n") {
if !strings.Contains(line, "event=process_output") {
continue
}
fields := map[string]string{}
for _, field := range strings.Fields(line) {
key, value, ok := strings.Cut(field, "=")
if !ok {
continue
}
if unquoted, err := strconv.Unquote(value); err == nil {
fields[key] = unquoted
} else {
fields[key] = value
}
}
byteCount, err := strconv.ParseInt(fields["bytes"], 10, 64)
if err != nil || byteCount <= 0 {
t.Fatalf("process output byte count = %q, want a positive integer: %q", fields["bytes"], line)
}
processOutputBytes[fields["stream"]] = append(processOutputBytes[fields["stream"]], byteCount)
}
for _, stream := range []string{"stdout", "stderr"} {
if len(processOutputBytes[stream]) == 0 {
t.Fatalf("process output metadata missing stream %q: %q", stream, string(logged))
}
}
info, err := os.Stat(logPath)
if err != nil {
t.Fatalf("stat process-output log: %v", err)
}
if info.Mode().Perm() != 0o600 {
t.Fatalf("process-output log mode = %o, want 600", info.Mode().Perm())
}
}
func TestPersistentDaemonProcessOutputSummaryBatchesReads(t *testing.T) {
startedAt := time.Unix(100, 0)
summary := persistentDaemonProcessOutputSummary{minimumInterval: time.Second}
if byteCount, emit := summary.record(10, startedAt); !emit || byteCount != 10 {
t.Fatalf("first record = (%d, %t), want (10, true)", byteCount, emit)
}
if byteCount, emit := summary.record(20, startedAt.Add(100*time.Millisecond)); emit || byteCount != 0 {
t.Fatalf("second record = (%d, %t), want (0, false)", byteCount, emit)
}
if byteCount, emit := summary.record(30, startedAt.Add(999*time.Millisecond)); emit || byteCount != 0 {
t.Fatalf("third record = (%d, %t), want (0, false)", byteCount, emit)
}
if byteCount, emit := summary.record(40, startedAt.Add(time.Second)); !emit || byteCount != 90 {
t.Fatalf("interval record = (%d, %t), want (90, true)", byteCount, emit)
}
if byteCount, emit := summary.record(50, startedAt.Add(1100*time.Millisecond)); emit || byteCount != 0 {
t.Fatalf("pending record = (%d, %t), want (0, false)", byteCount, emit)
}
if byteCount, emit := summary.flush(startedAt.Add(1200 * time.Millisecond)); !emit || byteCount != 50 {
t.Fatalf("final flush = (%d, %t), want (50, true)", byteCount, emit)
}
if byteCount, emit := summary.flush(startedAt.Add(1300 * time.Millisecond)); emit || byteCount != 0 {
t.Fatalf("empty flush = (%d, %t), want (0, false)", byteCount, emit)
}
}
func TestPersistentDaemonProcessOutputRejectsClosedTargetDescriptor(t *testing.T) {
const helperEnvironment = "CMUX_TEST_PERSISTENT_PROCESS_OUTPUT_CLOSED_FD"
const targetFD = 1
if os.Getenv(helperEnvironment) == "1" {
if err := syscall.Close(targetFD); err != nil {
t.Fatalf("close target descriptor: %v", err)
}
stream, err := routePersistentDaemonProcessOutputStream(
"stdout",
targetFD,
&bytes.Buffer{},
)
if stream != nil {
_ = stream.restoreAndDrain()
t.Fatal("routing a closed target descriptor unexpectedly succeeded")
}
if !errors.Is(err, syscall.EBADF) {
t.Fatalf("route closed target descriptor error = %v, want EBADF", err)
}
duplicate, duplicateErr := syscall.Dup(targetFD)
if duplicateErr == nil {
_ = syscall.Close(duplicate)
t.Fatal("routing failure reopened the closed target descriptor")
}
if !errors.Is(duplicateErr, syscall.EBADF) {
t.Fatalf("duplicate closed target descriptor error = %v, want EBADF", duplicateErr)
}
return
}
command := exec.Command(
os.Args[0],
"-test.run=^TestPersistentDaemonProcessOutputRejectsClosedTargetDescriptor$",
)
command.Env = append(os.Environ(), helperEnvironment+"=1")
var output bytes.Buffer
command.Stdout = &output
command.Stderr = &output
if err := command.Run(); err != nil {
t.Fatalf("closed-target helper failed: %v; output=%q", err, output.String())
}
}
@@ -0,0 +1,209 @@
package main
import (
"errors"
"io"
"os"
"strconv"
"sync"
"syscall"
"time"
)
const persistentDaemonProcessOutputSummaryInterval = time.Second
// persistentDaemonProcessOutputRoute keeps process-level stdout and stderr on
// the same rotating writer as structured daemon diagnostics. The persistent
// server is a detached child, so routing must be owned by that child rather
// than by the short-lived stdio proxy that launched it.
type persistentDaemonProcessOutputRoute struct {
streams []*persistentDaemonProcessOutputStream
closeOnce sync.Once
closeErr error
}
type persistentDaemonProcessOutputStream struct {
name string
reader *os.File
savedFD int
targetFD int
drained chan struct{}
}
type persistentDaemonProcessOutputSummary struct {
minimumInterval time.Duration
pendingBytes int64
lastEmittedAt time.Time
}
func (s *persistentDaemonProcessOutputSummary) record(byteCount int, now time.Time) (int64, bool) {
if byteCount <= 0 {
return 0, false
}
s.pendingBytes += int64(byteCount)
if s.lastEmittedAt.IsZero() || now.Sub(s.lastEmittedAt) >= s.minimumInterval {
return s.flush(now)
}
return 0, false
}
func (s *persistentDaemonProcessOutputSummary) flush(now time.Time) (int64, bool) {
if s.pendingBytes <= 0 {
return 0, false
}
byteCount := s.pendingBytes
s.pendingBytes = 0
s.lastEmittedAt = now
return byteCount, true
}
func shouldRoutePersistentDaemonProcessOutput(stderr io.Writer) bool {
file, ok := stderr.(*os.File)
return ok &&
file.Fd() == os.Stderr.Fd() &&
os.Getenv(persistentDaemonReadyFDEnv) != ""
}
func routePersistentDaemonProcessOutput(writer io.Writer) (*persistentDaemonProcessOutputRoute, error) {
stdout, err := routePersistentDaemonProcessOutputStream(
"stdout",
int(os.Stdout.Fd()),
writer,
)
if err != nil {
return nil, err
}
stderr, err := routePersistentDaemonProcessOutputStream(
"stderr",
int(os.Stderr.Fd()),
writer,
)
if err != nil {
return nil, errors.Join(err, stdout.restoreAndDrain())
}
return &persistentDaemonProcessOutputRoute{streams: []*persistentDaemonProcessOutputStream{stdout, stderr}}, nil
}
func routePersistentDaemonProcessOutputStream(
name string,
targetFD int,
writer io.Writer,
) (*persistentDaemonProcessOutputStream, error) {
savedFD, err := syscall.Dup(targetFD)
if err != nil {
return nil, err
}
closeSaved := true
defer func() {
if closeSaved {
_ = syscall.Close(savedFD)
}
}()
reader, pipeWriter, err := os.Pipe()
if err != nil {
return nil, err
}
closePipe := true
defer func() {
if closePipe {
_ = reader.Close()
_ = pipeWriter.Close()
}
}()
if err := replacePersistentDaemonFD(int(pipeWriter.Fd()), targetFD); err != nil {
return nil, err
}
if err := pipeWriter.Close(); err != nil {
_ = replacePersistentDaemonFD(savedFD, targetFD)
return nil, err
}
stream := &persistentDaemonProcessOutputStream{
name: name,
reader: reader,
savedFD: savedFD,
targetFD: targetFD,
drained: make(chan struct{}),
}
go stream.drain(writer)
closePipe = false
closeSaved = false
return stream, nil
}
func (s *persistentDaemonProcessOutputStream) drain(writer io.Writer) {
summary := persistentDaemonProcessOutputSummary{
minimumInterval: persistentDaemonProcessOutputSummaryInterval,
}
emitSummary := func(byteCount int64) {
logPersistentDaemonEvent(
writer,
"process_output",
"stream", s.name,
"bytes", strconv.FormatInt(byteCount, 10),
)
}
defer func() {
if byteCount, emit := summary.flush(time.Now()); emit {
emitSummary(byteCount)
}
close(s.drained)
}()
buffer := make([]byte, 32*1024)
for {
count, err := s.reader.Read(buffer)
if count > 0 {
// Process output is arbitrary and may contain terminal input, commands,
// or credentials. Record rate-limited byte summaries and discard the bytes.
if byteCount, emit := summary.record(count, time.Now()); emit {
emitSummary(byteCount)
}
}
if err != nil {
return
}
}
}
func (r *persistentDaemonProcessOutputRoute) Close() error {
if r == nil {
return nil
}
r.closeOnce.Do(func() {
var restoreErrors []error
for _, stream := range r.streams {
if err := stream.restore(); err != nil {
restoreErrors = append(restoreErrors, err)
}
}
for _, stream := range r.streams {
stream.waitForDrain()
}
r.closeErr = errors.Join(restoreErrors...)
})
return r.closeErr
}
func (s *persistentDaemonProcessOutputStream) restore() error {
err := replacePersistentDaemonFD(s.savedFD, s.targetFD)
_ = syscall.Close(s.savedFD)
if err != nil {
// A failed restore may leave a pipe writer open. Closing the reader
// guarantees shutdown cannot deadlock waiting for the drain goroutine.
_ = s.reader.Close()
}
return err
}
func (s *persistentDaemonProcessOutputStream) waitForDrain() {
<-s.drained
_ = s.reader.Close()
}
func (s *persistentDaemonProcessOutputStream) restoreAndDrain() error {
err := s.restore()
s.waitForDrain()
return err
}
@@ -0,0 +1,9 @@
//go:build darwin
package main
import "syscall"
func replacePersistentDaemonFD(from int, to int) error {
return syscall.Dup2(from, to)
}
@@ -0,0 +1,12 @@
//go:build linux
package main
import "syscall"
func replacePersistentDaemonFD(from int, to int) error {
if from == to {
return nil
}
return syscall.Dup3(from, to, 0)
}
+46 -5
View File
@@ -18,6 +18,7 @@ import (
"os/exec"
"path/filepath"
"sort"
"strconv"
"strings"
"sync"
"syscall"
@@ -530,7 +531,11 @@ func handleWebSocketPTY(w http.ResponseWriter, r *http.Request, cfg wsPTYServerC
attachment, err := cfg.PTYHub.attach(r.Context(), conn, auth)
if err != nil {
_, _ = fmt.Fprintf(stderr, "ws pty attach failed: %v\n", err)
logPersistentDaemonEvent(
stderr,
"ws_pty_attach_failed",
"error_category", persistentDaemonErrorCategory(err),
)
_ = conn.Close(websocket.StatusInternalError, truncateWebSocketCloseReason(err.Error()))
return
}
@@ -1236,10 +1241,26 @@ func (h *wsPTYHub) prepareAttachmentWithReservation(
if superseded != nil {
superseded.closeNow()
logPersistentDaemonEvent(
h.stderr,
"pty_detach",
"session_id", sessionID,
"attachment_id", superseded.id,
"reason", "superseded",
)
}
if shouldApplySize {
h.applyCurrentPTYSize(session)
}
logPersistentDaemonEvent(
h.stderr,
"pty_attach",
"session_id", sessionID,
"attachment_id", attachment.id,
"persistent", strconv.FormatBool(persistent),
"require_existing", strconv.FormatBool(requireExisting),
"replay_bytes", strconv.Itoa(attachment.replayBytes),
)
return attachment, attachmentCtx, sessionDone, nil
}
@@ -1313,9 +1334,12 @@ func (h *wsPTYHub) startSession(sessionKey wsPTYSessionKey, sessionID string, co
if tmpScript != "" {
_ = os.Remove(tmpScript)
}
if h.stderr != nil {
_, _ = fmt.Fprintf(h.stderr, "pty session start failed session=%s: %v\n", sessionID, err)
}
logPersistentDaemonEvent(
h.stderr,
"pty_start_fault",
"session_id", sessionID,
"error_category", persistentDaemonErrorCategory(err),
)
return nil, err
}
session := &wsPTYSession{
@@ -1532,6 +1556,13 @@ func (h *wsPTYHub) detach(attachment *wsPTYAttachment) bool {
if shouldApplySize {
h.applyCurrentPTYSize(session)
}
logPersistentDaemonEvent(
h.stderr,
"pty_detach",
"session_id", session.id,
"attachment_id", attachment.id,
"reason", "attachment_removed",
)
return true
}
@@ -1634,6 +1665,7 @@ func (h *wsPTYHub) closeSessionByID(sessionID string) bool {
if start := h.startingSessions[sessionKey]; start != nil {
start.closeRequested = true
h.mu.Unlock()
logPersistentDaemonEvent(h.stderr, "pty_close", "session_id", sessionID, "phase", "starting")
return true
}
session := h.sessions[sessionKey]
@@ -1645,6 +1677,7 @@ func (h *wsPTYHub) closeSessionByID(sessionID string) bool {
h.cancelIdleReapLocked(session)
session.closed = true
h.mu.Unlock()
logPersistentDaemonEvent(h.stderr, "pty_close", "session_id", sessionID, "phase", "running")
session.terminateProcesses()
session.closePTYFiles()
@@ -1937,6 +1970,9 @@ func (h *wsPTYHub) pumpSession(session *wsPTYSession) {
}
func (h *wsPTYHub) finishSession(session *wsPTYSession) {
// Record the exit before closing session.done so an attachment cannot
// observe pty.exit ahead of the corresponding persistent diagnostic.
logPersistentDaemonEvent(h.stderr, "pty_exit", "session_id", session.id)
session.closePTYFiles()
h.mu.Lock()
@@ -2141,7 +2177,12 @@ func (h *wsPTYHub) applyPTYSizeWithWriteLock(session *wsPTYSession, cols int, ro
lastErr = fmt.Errorf("pty size remained %dx%d after resize to %dx%d", actual.Cols, actual.Rows, cols, rows)
}
if h.stderr != nil && lastErr != nil {
_, _ = fmt.Fprintf(h.stderr, "ws pty resize failed session=%s: %v\n", session.id, lastErr)
logPersistentDaemonEvent(
h.stderr,
"pty_resize_fault",
"session_id", session.id,
"error_category", persistentDaemonErrorCategory(lastErr),
)
}
return false
}
@@ -99,8 +99,8 @@ func newTestWebSocketPTYServer(t *testing.T, leasePath string) (*httptest.Server
// allocation failure (e.g. a hardened devpts mounted ptmxmode=000 where
// /dev/ptmx cannot be opened) is reported loudly: the error returned to the
// client names the failing device and explains the devpts cause, and the daemon
// records the failure instead of leaving a 0-byte log. This is the regression
// for https://github.com/manaflow-ai/cmux/issues/5185, where the failure
// records a safe failure category instead of leaving a 0-byte log. This is the
// regression for https://github.com/manaflow-ai/cmux/issues/5185, where the failure
// collapsed into a generic "remote PTY attach failed" with an empty daemon log.
func TestAttachRPCSurfacesPTYAllocationFailure(t *testing.T) {
stderr := &bytes.Buffer{}
@@ -140,8 +140,12 @@ func TestAttachRPCSurfacesPTYAllocationFailure(t *testing.T) {
if stderr.Len() == 0 {
t.Fatalf("PTY allocation failure must be logged to the daemon log, not swallowed")
}
if !strings.Contains(stderr.String(), "/dev/ptmx") {
t.Fatalf("daemon log should include the allocation failure detail: %q", stderr.String())
if !strings.Contains(stderr.String(), "event=pty_start_fault") ||
!strings.Contains(stderr.String(), `error_category="permission_denied"`) {
t.Fatalf("daemon log should classify the allocation failure: %q", stderr.String())
}
if strings.Contains(stderr.String(), denied.Error()) {
t.Fatalf("daemon log should not persist the raw allocation failure: %q", stderr.String())
}
}