From 64aa36cb01e3d942f2e7e263cbb2620dc8825ac8 Mon Sep 17 00:00:00 2001 From: Omry Yadan Date: Wed, 12 Aug 2026 20:48:15 +0800 Subject: [PATCH] Add embedded controlled-session broker Add the controller-side controlled-session client command to the monolithic Reploy binary. Translate the private host protocol into a strict versioned JSON Lines stream and a same-identity, single-use terminal socket with bounded attachment and backpressure behavior. Validate public requests against granted operations and lifecycle state, keep the broker alive through output finalization and terminal-result acknowledgement, and fail gracefully on unsupported platforms. Start terminal request and disconnect monitoring at attachment acceptance, preserving bounded pre-open input and the latest resize until host authorization arrives. Bound broker-to-host request writes so PTY backpressure cannot hold the control loop indefinitely, atomically discard in-flight PTY traffic once host termination latches, and preserve the acknowledgement write across clean post-result EOF. Stabilize watchdog-exit coverage by driving test input only after readiness. Cover protocol rejection, ordering, timeouts, socket loss, cleanup, and platform behavior. --- ...+controlled-session-controller-broker.yaml | 2 + docs/CONTROLLED_SESSION_DESIGN.md | 12 +- internal/cli/cli.go | 47 + internal/cli/controlled_session_test.go | 59 ++ internal/controlledsession/client.go | 10 +- internal/controlledsession/client_test.go | 85 ++ .../controlledsession/controller_broker.go | 630 ++++++++++++ .../controller_broker_linux_test.go | 925 ++++++++++++++++++ .../controlledsession/controller_stream.go | 281 ++++++ .../controller_stream_test.go | 101 ++ .../controlledsession/controller_terminal.go | 143 +++ .../controller_terminal_linux.go | 133 +++ .../controller_terminal_linux_test.go | 108 ++ .../controller_terminal_unsupported.go | 9 + .../controller_terminal_unsupported_test.go | 14 + internal/controlledsession/lifecycle.go | 38 +- internal/controlledsession/lifecycle_test.go | 20 + .../controlledsession/terminal_protocol.go | 152 +++ .../terminal_protocol_test.go | 65 ++ .../controlled_session_supervisor.go | 7 +- .../controlled_session_supervisor_test.go | 37 +- 21 files changed, 2864 insertions(+), 14 deletions(-) create mode 100644 .changes/unreleased/+controlled-session-controller-broker.yaml create mode 100644 internal/cli/controlled_session_test.go create mode 100644 internal/controlledsession/controller_broker.go create mode 100644 internal/controlledsession/controller_broker_linux_test.go create mode 100644 internal/controlledsession/controller_stream.go create mode 100644 internal/controlledsession/controller_stream_test.go create mode 100644 internal/controlledsession/controller_terminal.go create mode 100644 internal/controlledsession/controller_terminal_linux.go create mode 100644 internal/controlledsession/controller_terminal_linux_test.go create mode 100644 internal/controlledsession/controller_terminal_unsupported.go create mode 100644 internal/controlledsession/controller_terminal_unsupported_test.go create mode 100644 internal/controlledsession/terminal_protocol.go create mode 100644 internal/controlledsession/terminal_protocol_test.go diff --git a/.changes/unreleased/+controlled-session-controller-broker.yaml b/.changes/unreleased/+controlled-session-controller-broker.yaml new file mode 100644 index 00000000..ad9a3262 --- /dev/null +++ b/.changes/unreleased/+controlled-session-controller-broker.yaml @@ -0,0 +1,2 @@ +kind: Added +body: Add the embedded `reploy controlled-session client` broker with a strict versioned controller stream and a private, bounded terminal transport for OmegaFlow-controlled workloads. diff --git a/docs/CONTROLLED_SESSION_DESIGN.md b/docs/CONTROLLED_SESSION_DESIGN.md index e7e2196c..88122c0a 100644 --- a/docs/CONTROLLED_SESSION_DESIGN.md +++ b/docs/CONTROLLED_SESSION_DESIGN.md @@ -1248,6 +1248,10 @@ closes the private host connection, and therefore cannot be mistaken for successful completion. Repeated `terminate` remains idempotent and repeated valid resize requests remain ordinary operations. +Each broker-to-host request write is bounded to one second. Expiration is a +fatal transport failure, so host-side PTY backpressure cannot hold the broker's +control loop indefinitely. + The public `opened` projection contains only the operations granted to the controller, endpoint coordinates, terminal dimensions, and output-finalization timeout. It does not expose the host-internal authorization record or its @@ -1282,6 +1286,10 @@ content, or workload filesystem. The broker emits `broker-ready` after the listener exists and then claims the Host Reploy channel. It allows ten seconds for the one attachment to connect. +Request and disconnect monitoring begins as soon as that attachment is +accepted, including while the host claim is still waiting for `opened`; input +and the latest resize received in that interval join the same bounded +pre-`ready` buffer. An attachment EOF in that interval fails immediately. Before attachment, it may hold at most one complete private-protocol output frame and otherwise applies backpressure to Host Reploy. Attach timeout, unexpected attachment exit, malformed attachment traffic, or loss before the @@ -1293,7 +1301,9 @@ resize, and terminal-end records so terminal bytes cannot be interpreted as control. The attachment switches its asciinema-owned PTY to raw mode while running, restores it on exit, forwards ordinary Ctrl-C as byte `0x03`, and translates `SIGWINCH` into resize records. It writes received output bytes to -stdout unchanged and never writes diagnostics there. +stdout unchanged and never writes diagnostics there. Input or resize already +in flight when host termination latches is discarded without changing the +termination cause or reporting controller loss. For an activated session, the broker sends terminal-end only after it has forwarded every earlier output byte and then received the ordered diff --git a/internal/cli/cli.go b/internal/cli/cli.go index 398b8637..a59963ee 100644 --- a/internal/cli/cli.go +++ b/internal/cli/cli.go @@ -19,6 +19,7 @@ import ( reploy "github.com/omry/reploy" "github.com/omry/reploy/internal/blueprint" + "github.com/omry/reploy/internal/controlledsession" "github.com/omry/reploy/internal/deploy" "github.com/omry/reploy/internal/dockerdeploy" "github.com/omry/reploy/internal/overrideui" @@ -49,6 +50,7 @@ var dockerAppCommand = dockerdeploy.AppCommand var runOverrideEditor = overrideui.RunWithResult var runBuildProgress = overrideui.RunBuildProgress var inspectStagedOverrideValidation = dockerdeploy.InspectStagedOverrideValidation +var runControlledSessionBroker = controlledsession.RunControllerBrokerV1 func Main(args []string, stdout io.Writer, stderr io.Writer) int { if message := windowsWSLBoundaryError(runtime.GOOS, os.LookupEnv, os.Getwd); message != "" { @@ -85,6 +87,8 @@ func Main(args []string, stdout io.Writer, stderr io.Writer) int { return runBlueprintValidate(args[1:], stdout, stderr) case "services": return runServices(args[1:], stdout, stderr) + case "controlled-session": + return runControlledSession(args[1:], stdout, stderr) default: if isDeploymentCommand(args[0]) { return runDocker(args, stdout, stderr, globalOptions) @@ -99,6 +103,32 @@ func Main(args []string, stdout io.Writer, stderr io.Writer) int { } } +func runControlledSession(args []string, stdout io.Writer, stderr io.Writer) int { + if (len(args) == 1 && isHelpArg(args[0])) || (len(args) == 2 && args[0] == "client" && isHelpArg(args[1])) { + printControlledSessionHelp(stdout) + return 0 + } + if len(args) != 1 || args[0] != "client" { + fmt.Fprintln(stderr, "reploy controlled-session usage error: expected client") + printControlledSessionShortUsage(stderr) + return 2 + } + socket := os.Getenv("REPLOY_SESSION_SOCKET") + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + err := runControlledSessionBroker(ctx, controlledsession.ControllerBrokerOptionsV1{ + SessionSocket: socket, + TemporaryHome: controlledsession.ControllerTemporaryHomeV1, + Input: os.Stdin, + Output: stdout, + }) + if err != nil { + fmt.Fprintf(stderr, "reploy controlled-session client error: %v\n", err) + return 1 + } + return 0 +} + func runEmbeddedServiceContainer(args []string, stdout io.Writer, stderr io.Writer, globalOptions globalDeploymentOptions) int { dir := "" dockerPath := "" @@ -3174,6 +3204,8 @@ Commands: install Install or update a deployed host service uninstall Remove an installed host service and Docker resources services List Reploy-managed services + controlled-session + Run controller-side controlled-session integration commands index Manage the cached blueprint shorthand index version Print version information @@ -3209,6 +3241,21 @@ Run 'reploy COMMAND --help' for command-specific options. `, "\n")) } +func printControlledSessionShortUsage(output io.Writer) { + fmt.Fprintln(output, "Usage: reploy controlled-session client") +} + +func printControlledSessionHelp(output io.Writer) { + fmt.Fprint(output, strings.TrimLeft(` +Usage: reploy controlled-session client + +Run the controller-side controlled-session broker. The command consumes the +controller-private REPLOY_SESSION_SOCKET environment variable and exchanges +versioned JSON Lines with the controller orchestrator on stdin and stdout. +Human-readable diagnostics are written only to stderr. +`, "\n")) +} + func printPackIndexShortUsage(commandName string, output io.Writer) { fmt.Fprintf(output, "Usage: reploy %s COMMAND\n", commandName) fmt.Fprintln(output) diff --git a/internal/cli/controlled_session_test.go b/internal/cli/controlled_session_test.go new file mode 100644 index 00000000..bb78c336 --- /dev/null +++ b/internal/cli/controlled_session_test.go @@ -0,0 +1,59 @@ +package cli + +import ( + "context" + "errors" + "strings" + "testing" + + "github.com/omry/reploy/internal/controlledsession" +) + +func TestControlledSessionHelpAndUsage(t *testing.T) { + code, stdout, stderr := runCLI("controlled-session", "--help") + if code != 0 || stderr != "" || !strings.Contains(stdout, "Usage: reploy controlled-session client") || !strings.Contains(stdout, "REPLOY_SESSION_SOCKET") { + t.Fatalf("help code=%d stdout=%q stderr=%q", code, stdout, stderr) + } + code, stdout, stderr = runCLI("controlled-session", "client", "--help") + if code != 0 || stderr != "" || !strings.Contains(stdout, "Usage: reploy controlled-session client") { + t.Fatalf("client help code=%d stdout=%q stderr=%q", code, stdout, stderr) + } + code, stdout, stderr = runCLI("controlled-session") + if code != 2 || stdout != "" || !strings.Contains(stderr, "expected client") || !strings.Contains(stderr, "Usage: reploy controlled-session client") { + t.Fatalf("usage code=%d stdout=%q stderr=%q", code, stdout, stderr) + } + code, stdout, stderr = runCLI("--help") + if code != 0 || stderr != "" || !strings.Contains(stdout, "controlled-session") { + t.Fatalf("top-level help code=%d stdout=%q stderr=%q", code, stdout, stderr) + } +} + +func TestControlledSessionClientDispatchesEmbeddedBroker(t *testing.T) { + original := runControlledSessionBroker + t.Cleanup(func() { runControlledSessionBroker = original }) + t.Setenv("REPLOY_SESSION_SOCKET", "/run/reploy/session.sock") + called := false + runControlledSessionBroker = func(ctx context.Context, options controlledsession.ControllerBrokerOptionsV1) error { + called = true + if ctx == nil || ctx.Done() == nil || options.SessionSocket != "/run/reploy/session.sock" || options.TemporaryHome != controlledsession.ControllerTemporaryHomeV1 || options.Input == nil || options.Output == nil { + t.Fatalf("broker options = %#v", options) + } + return nil + } + code, stdout, stderr := runCLI("controlled-session", "client") + if code != 0 || stdout != "" || stderr != "" || !called { + t.Fatalf("client code=%d stdout=%q stderr=%q called=%t", code, stdout, stderr, called) + } +} + +func TestControlledSessionClientReportsRuntimeFailureOnlyOnStderr(t *testing.T) { + original := runControlledSessionBroker + t.Cleanup(func() { runControlledSessionBroker = original }) + runControlledSessionBroker = func(context.Context, controlledsession.ControllerBrokerOptionsV1) error { + return errors.New("controlled-session controller broker requires Linux") + } + code, stdout, stderr := runCLI("controlled-session", "client") + if code != 1 || stdout != "" || !strings.Contains(stderr, "requires Linux") { + t.Fatalf("failure code=%d stdout=%q stderr=%q", code, stdout, stderr) + } +} diff --git a/internal/controlledsession/client.go b/internal/controlledsession/client.go index 62987565..4b385e43 100644 --- a/internal/controlledsession/client.go +++ b/internal/controlledsession/client.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "io" "net" "sync" ) @@ -79,7 +80,14 @@ func (client *SessionClientV1) ReadEvent(ctx context.Context) (EventV1, error) { return readErr }) if err != nil { - return EventV1{}, errors.Join(fmt.Errorf("read controlled-session client event: %w", err), client.Close()) + readErr := fmt.Errorf("read controlled-session client event: %w", err) + client.stateMu.RLock() + terminated := client.terminated + client.stateMu.RUnlock() + if terminated && errors.Is(err, io.EOF) { + return EventV1{}, readErr + } + return EventV1{}, errors.Join(readErr, client.Close()) } if event.Kind == EventOpenedV1 { return EventV1{}, errors.Join(fmt.Errorf("read controlled-session client event: opened may appear only once"), client.Close()) diff --git a/internal/controlledsession/client_test.go b/internal/controlledsession/client_test.go index 1477f5e0..cb424256 100644 --- a/internal/controlledsession/client_test.go +++ b/internal/controlledsession/client_test.go @@ -1,7 +1,10 @@ package controlledsession import ( + "bytes" "context" + "errors" + "io" "net" "reflect" "strings" @@ -170,6 +173,51 @@ func TestSessionClientV1AcknowledgesStartupFailureBeforeReady(t *testing.T) { } } +func TestSessionClientV1KeepsPostTerminatedEOFOpenForAcknowledgement(t *testing.T) { + var events bytes.Buffer + opened := testOpenedV1() + terminated := EventV1{Kind: EventTerminatedV1, Terminated: &ResultV1{ + Cause: CauseStartupFailureV1, + WorkloadStatus: ProcessStatusV1{Kind: ProcessStatusUnknownV1}, + WorkloadOutputFinalizationStatus: WorkloadOutputFinalizationStatusV1{Kind: WorkloadOutputFinalizationDrainedV1}, + RuntimeObservationStatus: RuntimeObservationStatusV1{Kind: RuntimeObservationMaintainedV1}, + ControllerFinalizationStatus: ControllerFinalizationStatusV1{Kind: ControllerFinalizationStartupFailedV1}, + CleanupStatus: CleanupStatusV1{Kind: CleanupStatusSucceededV1}, + RecoveryAction: RecoveryNoneV1, + }} + if err := WriteEventV1(&events, EventV1{Kind: EventOpenedV1, Opened: &opened}); err != nil { + t.Fatal(err) + } + if err := WriteEventV1(&events, terminated); err != nil { + t.Fatal(err) + } + connection := &scriptedSessionClientConnectionV1{reader: bytes.NewReader(events.Bytes())} + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + client, err := newSessionClientV1(ctx, connection) + if err != nil { + t.Fatal(err) + } + defer client.Close() + if event, err := client.ReadEvent(ctx); err != nil || !reflect.DeepEqual(event, terminated) { + t.Fatalf("terminated event = %#v, error = %v", event, err) + } + if _, err := client.ReadEvent(ctx); !errors.Is(err, io.EOF) { + t.Fatalf("post-terminated read error = %v", err) + } + if connection.closed { + t.Fatal("post-terminated EOF closed the connection before acknowledgement") + } + wantRequest := RequestV1{Kind: RequestAcknowledgeTerminatedV1} + if err := client.WriteRequest(ctx, wantRequest); err != nil { + t.Fatal(err) + } + request, err := ReadRequestV1(bytes.NewReader(connection.writes.Bytes())) + if err != nil || !reflect.DeepEqual(request, wantRequest) { + t.Fatalf("acknowledgement = %#v, error = %v", request, err) + } +} + func TestSessionClientV1RequiresOpenedFirstAndRejectsRepeatedOpened(t *testing.T) { for _, test := range []struct { name string @@ -217,3 +265,40 @@ func testOpenedV1() OpenedV1 { } func pointerToOpenedV1(value OpenedV1) *OpenedV1 { return &value } + +type scriptedSessionClientConnectionV1 struct { + reader *bytes.Reader + writes bytes.Buffer + closed bool +} + +func (connection *scriptedSessionClientConnectionV1) Read(payload []byte) (int, error) { + if connection.closed { + return 0, net.ErrClosed + } + return connection.reader.Read(payload) +} + +func (connection *scriptedSessionClientConnectionV1) Write(payload []byte) (int, error) { + if connection.closed { + return 0, net.ErrClosed + } + return connection.writes.Write(payload) +} + +func (connection *scriptedSessionClientConnectionV1) Close() error { + connection.closed = true + return nil +} + +func (*scriptedSessionClientConnectionV1) LocalAddr() net.Addr { + return &net.UnixAddr{Name: "scripted-local", Net: "unix"} +} + +func (*scriptedSessionClientConnectionV1) RemoteAddr() net.Addr { + return &net.UnixAddr{Name: "scripted-remote", Net: "unix"} +} + +func (*scriptedSessionClientConnectionV1) SetDeadline(time.Time) error { return nil } +func (*scriptedSessionClientConnectionV1) SetReadDeadline(time.Time) error { return nil } +func (*scriptedSessionClientConnectionV1) SetWriteDeadline(time.Time) error { return nil } diff --git a/internal/controlledsession/controller_broker.go b/internal/controlledsession/controller_broker.go new file mode 100644 index 00000000..564995aa --- /dev/null +++ b/internal/controlledsession/controller_broker.go @@ -0,0 +1,630 @@ +package controlledsession + +import ( + "context" + "errors" + "fmt" + "io" + "path/filepath" + "strings" + "time" + "unicode/utf8" +) + +const ( + DefaultControllerAttachTimeoutV1 = 10 * time.Second + controllerRequestWriteTimeoutV1 = time.Second +) + +type ControllerBrokerOptionsV1 struct { + SessionSocket string + TemporaryHome string + Input io.Reader + Output io.Writer + AttachTimeout time.Duration +} + +type controllerBrokerRequestSourceV1 string + +const ( + controllerBrokerRequestPublicV1 controllerBrokerRequestSourceV1 = "public" + controllerBrokerRequestTerminalV1 controllerBrokerRequestSourceV1 = "terminal" +) + +type controllerBrokerRequestV1 struct { + source controllerBrokerRequestSourceV1 + request RequestV1 +} + +type controllerBrokerInputErrorV1 struct { + source controllerBrokerRequestSourceV1 + err error +} + +type controllerBrokerTerminalAcceptResultV1 struct { + connection *ControllerTerminalConnectionV1 + err error +} + +type controllerBrokerSessionResultV1 struct { + session *SessionClientV1 + err error +} + +type controllerBrokerTerminalOutputResultV1 struct { + err error +} + +type controllerBrokerPendingTerminalV1 struct { + input *RequestV1 + resize *RequestV1 +} + +type controllerBrokerStateV1 struct { + operations map[OperationV1]bool + requestWriteTimeout time.Duration + terminalWriteTimeout time.Duration + ready bool + workloadExited bool + terminating bool + outputsFinalized bool + terminalEnded bool + terminated bool + completeSent bool + acknowledged bool + pendingTerminalInput *RequestV1 + pendingTerminalResize *RequestV1 +} + +type controllerBrokerSessionV1 interface { + Opened() OpenedV1 + ReadEvent(context.Context) (EventV1, error) + WriteRequest(context.Context, RequestV1) error + Close() error +} + +func RunControllerBrokerV1(ctx context.Context, options ControllerBrokerOptionsV1) (resultErr error) { + if ctx == nil || ctx.Done() == nil { + return fmt.Errorf("run controlled-session controller broker: cancelable context is required") + } + if !filepath.IsAbs(options.SessionSocket) || filepath.Clean(options.SessionSocket) != options.SessionSocket { + return fmt.Errorf("run controlled-session controller broker: REPLOY_SESSION_SOCKET must contain an absolute clean path") + } + if options.TemporaryHome == "" { + options.TemporaryHome = ControllerTemporaryHomeV1 + } + if options.AttachTimeout == 0 { + options.AttachTimeout = DefaultControllerAttachTimeoutV1 + } + if options.AttachTimeout < 0 { + return fmt.Errorf("run controlled-session controller broker: attach timeout must be positive") + } + streamReader, err := NewControllerStreamReaderV1(options.Input) + if err != nil { + return err + } + streamWriter, err := NewControllerStreamWriterV1(options.Output) + if err != nil { + return err + } + terminalListener, err := PrepareControllerTerminalListenerV1(options.TemporaryHome) + if err != nil { + return err + } + defer func() { resultErr = errors.Join(resultErr, terminalListener.Close()) }() + if err := streamWriter.WriteEvent(ControllerStreamEventV1{Kind: ControllerStreamEventBrokerReadyV1, TerminalSocket: terminalListener.SocketPath()}); err != nil { + return err + } + attachCtx, cancelAttach := context.WithTimeout(ctx, options.AttachTimeout) + defer cancelAttach() + setupCtx, cancelSetup := context.WithCancel(ctx) + defer cancelSetup() + terminalReadCtx, cancelTerminalRead := context.WithCancel(ctx) + defer cancelTerminalRead() + requests := make(chan controllerBrokerRequestV1) + inputErrors := make(chan controllerBrokerInputErrorV1, 2) + var pendingTerminal controllerBrokerPendingTerminalV1 + terminalResult := make(chan controllerBrokerTerminalAcceptResultV1) + go func() { + terminal, acceptErr := terminalListener.Accept(attachCtx) + select { + case terminalResult <- controllerBrokerTerminalAcceptResultV1{connection: terminal, err: acceptErr}: + case <-setupCtx.Done(): + if terminal != nil { + _ = terminal.Close() + } + } + }() + sessionResult := make(chan controllerBrokerSessionResultV1) + go func() { + session, dialErr := DialSessionClientV1(setupCtx, options.SessionSocket) + select { + case sessionResult <- controllerBrokerSessionResultV1{session: session, err: dialErr}: + case <-setupCtx.Done(): + if session != nil { + _ = session.Close() + } + } + }() + var session *SessionClientV1 + var terminal *ControllerTerminalConnectionV1 + for session == nil || terminal == nil { + select { + case <-ctx.Done(): + return failControllerBrokerV1(streamWriter, "broker_canceled", ctx.Err()) + case terminalAccepted := <-terminalResult: + if terminalAccepted.err != nil { + code := "terminal_attachment_error" + if errors.Is(terminalAccepted.err, context.DeadlineExceeded) { + code = "attach_timeout" + } + return failControllerBrokerV1(streamWriter, code, terminalAccepted.err) + } + terminal = terminalAccepted.connection + cancelAttach() + go readControllerBrokerTerminalRequestsV1(terminalReadCtx, terminal, requests, inputErrors) + case sessionOpened := <-sessionResult: + if sessionOpened.err != nil { + return failControllerBrokerV1(streamWriter, "host_transport_error", sessionOpened.err) + } + session = sessionOpened.session + defer func() { resultErr = errors.Join(resultErr, session.Close()) }() + opened := session.Opened() + publicOpened := &ControllerStreamOpenedV1{ + Operations: append([]OperationV1{}, opened.Authorization.Operations...), + Endpoints: append([]EndpointV1{}, opened.Endpoints...), + Columns: opened.Columns, + Rows: opened.Rows, + OutputFinalizationTimeoutMilliseconds: opened.OutputFinalizationTimeoutMilliseconds, + } + if err := streamWriter.WriteEvent(ControllerStreamEventV1{Kind: ControllerStreamEventOpenedV1, Opened: publicOpened}); err != nil { + return err + } + case inputErr := <-inputErrors: + return failControllerBrokerV1(streamWriter, "terminal_attachment_error", inputErr.err) + case request := <-requests: + if err := pendingTerminal.add(request); err != nil { + return failControllerBrokerV1(streamWriter, "terminal_attachment_error", err) + } + } + } + cancelSetup() + opened := session.Opened() + terminalWriteTimeout := time.Duration(opened.OutputFinalizationTimeoutMilliseconds) * time.Millisecond + return runClaimedControllerBrokerV1(ctx, session, terminal, streamReader, streamWriter, requests, inputErrors, pendingTerminal, opened.Authorization.Operations, controllerRequestWriteTimeoutV1, terminalWriteTimeout) +} + +func runClaimedControllerBrokerV1( + ctx context.Context, + session controllerBrokerSessionV1, + terminal *ControllerTerminalConnectionV1, + streamReader *ControllerStreamReaderV1, + streamWriter *ControllerStreamWriterV1, + requests chan controllerBrokerRequestV1, + inputErrors chan controllerBrokerInputErrorV1, + pendingTerminal controllerBrokerPendingTerminalV1, + operations []OperationV1, + requestWriteTimeout time.Duration, + terminalWriteTimeout time.Duration, +) error { + runCtx, cancel := context.WithCancel(ctx) + defer cancel() + hostEvents := make(chan EventV1) + hostErrors := make(chan error, 1) + terminalOutputs := make(chan []byte) + terminalOutputResults := make(chan controllerBrokerTerminalOutputResultV1) + go readControllerBrokerPublicRequestsV1(runCtx, streamReader, requests, inputErrors) + go readControllerBrokerHostEventsV1(runCtx, session, hostEvents, hostErrors) + go writeControllerBrokerTerminalOutputV1(runCtx, terminalWriteTimeout, terminal, terminalOutputs, terminalOutputResults) + state := controllerBrokerStateV1{ + operations: make(map[OperationV1]bool, len(operations)), + requestWriteTimeout: requestWriteTimeout, + terminalWriteTimeout: terminalWriteTimeout, + } + for _, operation := range operations { + state.operations[operation] = true + } + for _, request := range pendingTerminal.requests() { + if err := state.applyRequest(ctx, session, request); err != nil { + return failControllerBrokerV1(streamWriter, "request_rejected", err) + } + } + var pendingOutput []byte + var outputPending bool + var outputWrite chan<- []byte + var outputResult <-chan controllerBrokerTerminalOutputResultV1 + for { + readHostEvents := hostEvents + if outputPending { + readHostEvents = nil + if outputResult == nil { + outputWrite = terminalOutputs + } + } + select { + case <-ctx.Done(): + return failControllerBrokerV1(streamWriter, "broker_canceled", ctx.Err()) + case inputErr := <-inputErrors: + if inputErr.source == controllerBrokerRequestTerminalV1 && state.terminalEnded && errors.Is(inputErr.err, io.EOF) { + continue + } + if inputErr.source == controllerBrokerRequestPublicV1 && state.acknowledged && errors.Is(inputErr.err, io.EOF) { + continue + } + code := "public_stream_error" + if inputErr.source == controllerBrokerRequestTerminalV1 { + code = "terminal_attachment_error" + } + return failControllerBrokerV1(streamWriter, code, inputErr.err) + case request := <-requests: + if err := state.applyRequest(ctx, session, request); err != nil { + return failControllerBrokerV1(streamWriter, "request_rejected", err) + } + case event := <-readHostEvents: + if event.Kind == EventOutputV1 { + if err := state.validateHostOutputEvent(event); err != nil { + return failControllerBrokerV1(streamWriter, "host_event_error", err) + } + pendingOutput = make([]byte, len(event.Bytes)) + copy(pendingOutput, event.Bytes) + outputPending = true + continue + } + if err := state.applyHostEvent(ctx, session, terminal, streamWriter, event); err != nil { + return failControllerBrokerV1(streamWriter, "host_event_error", err) + } + case hostErr := <-hostErrors: + if state.acknowledged && errors.Is(hostErr, io.EOF) { + return nil + } + return failControllerBrokerV1(streamWriter, "host_transport_error", hostErr) + case outputWrite <- pendingOutput: + outputWrite = nil + outputResult = terminalOutputResults + case result := <-outputResult: + if result.err != nil { + return failControllerBrokerV1(streamWriter, "terminal_attachment_error", result.err) + } + pendingOutput = nil + outputPending = false + outputResult = nil + } + } +} + +func (pending *controllerBrokerPendingTerminalV1) add(input controllerBrokerRequestV1) error { + if input.source != controllerBrokerRequestTerminalV1 { + return fmt.Errorf("setup request did not originate from the terminal attachment") + } + switch input.request.Kind { + case RequestInputV1: + pendingLength := len(input.request.Bytes) + if pending.input != nil { + pendingLength += len(pending.input.Bytes) + } + if pendingLength > MaxFramePayloadV1 { + return fmt.Errorf("terminal input buffered before host open exceeds %d bytes", MaxFramePayloadV1) + } + if pending.input == nil { + pending.input = &RequestV1{Kind: RequestInputV1, Bytes: make([]byte, 0, pendingLength)} + } + pending.input.Bytes = append(pending.input.Bytes, input.request.Bytes...) + case RequestResizeV1: + request := input.request + pending.resize = &request + default: + return fmt.Errorf("terminal attachment cannot carry %q", input.request.Kind) + } + return nil +} + +func (pending controllerBrokerPendingTerminalV1) requests() []controllerBrokerRequestV1 { + requests := make([]controllerBrokerRequestV1, 0, 2) + if pending.resize != nil { + requests = append(requests, controllerBrokerRequestV1{source: controllerBrokerRequestTerminalV1, request: *pending.resize}) + } + if pending.input != nil { + requests = append(requests, controllerBrokerRequestV1{ + source: controllerBrokerRequestTerminalV1, + request: RequestV1{Kind: RequestInputV1, Bytes: append([]byte{}, pending.input.Bytes...)}, + }) + } + return requests +} + +func readControllerBrokerPublicRequestsV1(ctx context.Context, reader *ControllerStreamReaderV1, requests chan<- controllerBrokerRequestV1, failures chan<- controllerBrokerInputErrorV1) { + for { + message, err := reader.ReadRequest() + if err != nil { + select { + case failures <- controllerBrokerInputErrorV1{source: controllerBrokerRequestPublicV1, err: err}: + case <-ctx.Done(): + } + return + } + request := RequestV1{Kind: RequestKindV1(message.Kind), Columns: message.Columns, Rows: message.Rows} + select { + case requests <- controllerBrokerRequestV1{source: controllerBrokerRequestPublicV1, request: request}: + case <-ctx.Done(): + return + } + } +} + +func readControllerBrokerTerminalRequestsV1(ctx context.Context, terminal *ControllerTerminalConnectionV1, requests chan<- controllerBrokerRequestV1, failures chan<- controllerBrokerInputErrorV1) { + for { + request, err := terminal.ReadRequest(ctx) + if err != nil { + select { + case failures <- controllerBrokerInputErrorV1{source: controllerBrokerRequestTerminalV1, err: err}: + case <-ctx.Done(): + } + return + } + select { + case requests <- controllerBrokerRequestV1{source: controllerBrokerRequestTerminalV1, request: request}: + case <-ctx.Done(): + return + } + } +} + +func readControllerBrokerHostEventsV1(ctx context.Context, session controllerBrokerSessionV1, events chan<- EventV1, failures chan<- error) { + for { + event, err := session.ReadEvent(ctx) + if err != nil { + select { + case failures <- err: + case <-ctx.Done(): + } + return + } + select { + case events <- event: + case <-ctx.Done(): + return + } + } +} + +func writeControllerBrokerTerminalOutputV1(ctx context.Context, timeout time.Duration, terminal *ControllerTerminalConnectionV1, outputs <-chan []byte, results chan<- controllerBrokerTerminalOutputResultV1) { + for { + select { + case output := <-outputs: + writeCtx, cancelWrite := context.WithTimeout(ctx, timeout) + err := terminal.WriteOutput(writeCtx, output) + cancelWrite() + select { + case results <- controllerBrokerTerminalOutputResultV1{err: err}: + case <-ctx.Done(): + return + } + if err != nil { + return + } + case <-ctx.Done(): + return + } + } +} + +func (state *controllerBrokerStateV1) applyRequest(ctx context.Context, session controllerBrokerSessionV1, input controllerBrokerRequestV1) error { + request := input.request + if input.source == controllerBrokerRequestPublicV1 && request.Kind == RequestInputV1 { + return fmt.Errorf("public controller stream cannot carry terminal input") + } + if input.source == controllerBrokerRequestTerminalV1 && request.Kind != RequestInputV1 && request.Kind != RequestResizeV1 { + return fmt.Errorf("terminal attachment cannot carry %q", request.Kind) + } + if input.source == controllerBrokerRequestTerminalV1 && (state.terminating || state.terminated) { + return nil + } + if request.Kind == RequestAcknowledgeTerminatedV1 { + if !state.terminated || state.acknowledged { + return fmt.Errorf("acknowledge-terminated is valid exactly once after terminated") + } + if err := state.writeHostRequest(ctx, session, request); err != nil { + return err + } + state.acknowledged = true + return nil + } + if !state.ready { + if input.source == controllerBrokerRequestTerminalV1 { + operation := operationForRequestV1(request.Kind) + if operation == "" || !state.operations[operation] { + return fmt.Errorf("operation %q was not granted", operation) + } + switch request.Kind { + case RequestInputV1: + pendingLength := len(request.Bytes) + if state.pendingTerminalInput != nil { + pendingLength += len(state.pendingTerminalInput.Bytes) + } + if pendingLength > MaxFramePayloadV1 { + return fmt.Errorf("terminal input buffered before ready exceeds %d bytes", MaxFramePayloadV1) + } + if state.pendingTerminalInput == nil { + state.pendingTerminalInput = &RequestV1{Kind: RequestInputV1, Bytes: make([]byte, 0, pendingLength)} + } + state.pendingTerminalInput.Bytes = append(state.pendingTerminalInput.Bytes, request.Bytes...) + return nil + case RequestResizeV1: + pending := request + state.pendingTerminalResize = &pending + return nil + } + } + return fmt.Errorf("%s is premature before ready", request.Kind) + } + if state.terminated { + return fmt.Errorf("%s is invalid after terminated", request.Kind) + } + operation := operationForRequestV1(request.Kind) + if operation == "" || !state.operations[operation] { + return fmt.Errorf("operation %q was not granted", operation) + } + switch request.Kind { + case RequestInputV1, RequestResizeV1: + if state.terminating { + return fmt.Errorf("%s is invalid while terminating", request.Kind) + } + case RequestTerminateV1: + // Repeated terminate requests remain idempotent at the host lifecycle. + case RequestCompleteV1: + if !state.terminating || !state.outputsFinalized || state.completeSent { + return fmt.Errorf("complete requires finalized workload output and may be sent once") + } + default: + return fmt.Errorf("request %q is unsupported", request.Kind) + } + if err := state.writeHostRequest(ctx, session, request); err != nil { + return err + } + if request.Kind == RequestCompleteV1 { + state.completeSent = true + } + return nil +} + +func (state *controllerBrokerStateV1) applyHostEvent(ctx context.Context, session controllerBrokerSessionV1, terminal *ControllerTerminalConnectionV1, writer *ControllerStreamWriterV1, event EventV1) error { + switch event.Kind { + case EventReadyV1: + if state.ready || state.terminating || state.terminated { + return fmt.Errorf("ready event is out of order") + } + if state.pendingTerminalResize != nil { + if err := state.writeHostRequest(ctx, session, *state.pendingTerminalResize); err != nil { + return err + } + state.pendingTerminalResize = nil + } + state.ready = true + if err := writer.WriteEvent(ControllerStreamEventV1{Kind: ControllerStreamEventReadyV1}); err != nil { + return err + } + if state.pendingTerminalInput != nil { + if err := state.writeHostRequest(ctx, session, *state.pendingTerminalInput); err != nil { + return err + } + state.pendingTerminalInput = nil + } + return nil + case EventOutputV1: + if err := state.validateHostOutputEvent(event); err != nil { + return err + } + return terminal.WriteOutput(ctx, event.Bytes) + case EventWorkloadExitV1: + if state.workloadExited || state.outputsFinalized || state.terminated || event.WorkloadExit == nil { + return fmt.Errorf("workload-exit event is out of order") + } + state.workloadExited = true + return writer.WriteEvent(ControllerStreamEventV1{Kind: ControllerStreamEventWorkloadExitV1, WorkloadExit: event.WorkloadExit}) + case EventTerminatingV1: + if state.terminating || state.terminated || event.Terminating == nil { + return fmt.Errorf("terminating event is out of order") + } + state.terminating = true + return writer.WriteEvent(ControllerStreamEventV1{Kind: ControllerStreamEventTerminatingV1, Terminating: event.Terminating}) + case EventDiagnosticV1: + if state.terminated || event.Diagnostic == nil { + return fmt.Errorf("diagnostic event is out of order") + } + return writer.WriteEvent(ControllerStreamEventV1{Kind: ControllerStreamEventDiagnosticV1, Diagnostic: event.Diagnostic}) + case EventWorkloadOutputsFinalizedV1: + if !state.terminating || state.outputsFinalized || state.terminated || event.WorkloadOutputsFinalized == nil { + return fmt.Errorf("workload-outputs-finalized event is out of order") + } + if state.ready && !state.workloadExited && event.WorkloadOutputsFinalized.Status == WorkloadOutputFinalizationDrainedV1 { + return fmt.Errorf("drained workload output requires a prior workload-exit event") + } + if err := writer.WriteEvent(ControllerStreamEventV1{Kind: ControllerStreamEventWorkloadOutputsFinalizedV1, WorkloadOutputsFinalized: event.WorkloadOutputsFinalized}); err != nil { + return err + } + status := WorkloadOutputFinalizationStatusV1{Kind: event.WorkloadOutputsFinalized.Status, Reason: event.WorkloadOutputsFinalized.Reason} + if err := state.writeTerminalEnd(ctx, terminal, status); err != nil { + return err + } + state.outputsFinalized = true + state.terminalEnded = true + return nil + case EventTerminatedV1: + if state.terminated || event.Terminated == nil || state.ready && !state.terminating { + return fmt.Errorf("terminated event is out of order") + } + if !state.terminalEnded { + if state.ready { + return fmt.Errorf("terminated event arrived before terminal output ended") + } + if err := state.writeTerminalEnd(ctx, terminal, event.Terminated.WorkloadOutputFinalizationStatus); err != nil { + return err + } + state.terminalEnded = true + state.outputsFinalized = true + } + if err := writer.WriteEvent(ControllerStreamEventV1{Kind: ControllerStreamEventTerminatedV1, Terminated: event.Terminated}); err != nil { + return err + } + state.terminated = true + return nil + case EventOpenedV1: + return fmt.Errorf("opened event may appear only once") + default: + return fmt.Errorf("host event kind %q is unsupported", event.Kind) + } +} + +func (state *controllerBrokerStateV1) writeHostRequest(ctx context.Context, session controllerBrokerSessionV1, request RequestV1) error { + timeout := state.requestWriteTimeout + if timeout == 0 { + timeout = controllerRequestWriteTimeoutV1 + } + writeCtx, cancelWrite := context.WithTimeout(ctx, timeout) + defer cancelWrite() + return session.WriteRequest(writeCtx, request) +} + +func (state *controllerBrokerStateV1) writeTerminalEnd(ctx context.Context, terminal *ControllerTerminalConnectionV1, status WorkloadOutputFinalizationStatusV1) error { + writeCtx, cancelWrite := context.WithTimeout(ctx, state.terminalWriteTimeout) + defer cancelWrite() + return terminal.WriteEnd(writeCtx, status) +} + +func (state *controllerBrokerStateV1) validateHostOutputEvent(event EventV1) error { + if state.outputsFinalized || state.terminated || event.Bytes == nil { + return fmt.Errorf("output event is out of order") + } + return nil +} + +func failControllerBrokerV1(writer *ControllerStreamWriterV1, code string, cause error) error { + message := controllerBrokerPublicErrorMessageV1(cause) + _ = writer.WriteEvent(ControllerStreamEventV1{ + Kind: ControllerStreamEventClientErrorV1, + ClientError: &DiagnosticV1{Code: code, Message: message}, + }) + return cause +} + +func controllerBrokerPublicErrorMessageV1(err error) string { + message := strings.TrimSpace(err.Error()) + message = strings.Map(func(character rune) rune { + if character < 0x20 || character == 0x7f { + return ' ' + } + return character + }, message) + message = strings.Join(strings.Fields(message), " ") + if message == "" { + message = "controlled-session client failed" + } + const limit = 512 + if len(message) > limit { + for len(message) > limit { + _, size := utf8.DecodeLastRuneInString(message) + message = message[:len(message)-size] + } + } + return message +} diff --git a/internal/controlledsession/controller_broker_linux_test.go b/internal/controlledsession/controller_broker_linux_test.go new file mode 100644 index 00000000..a9d67cd6 --- /dev/null +++ b/internal/controlledsession/controller_broker_linux_test.go @@ -0,0 +1,925 @@ +//go:build linux + +package controlledsession + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "io" + "net" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" +) + +func TestRunControllerBrokerV1CompletesAcknowledgedSession(t *testing.T) { + hostListener, hostSocket := newControllerBrokerHostListenerV1(t) + defer hostListener.Close() + publicInputReader, publicInputWriter := io.Pipe() + defer publicInputWriter.Close() + publicOutputReader, publicOutputWriter := io.Pipe() + defer publicOutputReader.Close() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + hostDone := make(chan error, 1) + go runControllerBrokerTestHostV1(hostListener, hostDone) + brokerDone := make(chan error, 1) + temporaryHome := shortControllerBrokerTempHomeV1(t) + go func() { + err := RunControllerBrokerV1(ctx, ControllerBrokerOptionsV1{ + SessionSocket: hostSocket, + TemporaryHome: temporaryHome, + Input: publicInputReader, + Output: publicOutputWriter, + }) + _ = publicOutputWriter.CloseWithError(err) + brokerDone <- err + }() + public := bufio.NewReader(publicOutputReader) + brokerReady := readControllerBrokerJSONLineV1(t, public) + if brokerReady["type"] != string(ControllerStreamEventBrokerReadyV1) { + t.Fatalf("first public event = %#v", brokerReady) + } + terminalSocket, ok := brokerReady["terminal_socket"].(string) + if !ok || filepath.Dir(filepath.Dir(terminalSocket)) != temporaryHome { + t.Fatalf("terminal socket = %#v", brokerReady["terminal_socket"]) + } + opened := readControllerBrokerJSONLineV1(t, public) + if opened["type"] != string(ControllerStreamEventOpenedV1) || opened["authorization"] != nil { + t.Fatalf("opened public event = %#v", opened) + } + terminal, err := net.DialUnix("unix", nil, &net.UnixAddr{Name: terminalSocket, Net: "unix"}) + if err != nil { + t.Fatal(err) + } + defer terminal.Close() + if event := readControllerBrokerJSONLineV1(t, public); event["type"] != string(ControllerStreamEventReadyV1) { + t.Fatalf("ready public event = %#v", event) + } + terminalOutput, err := ReadTerminalEventV1(terminal) + if err != nil || terminalOutput.Kind != TerminalEventOutputV1 || string(terminalOutput.Bytes) != "hello from workload" { + t.Fatalf("terminal output = %#v, %v", terminalOutput, err) + } + for _, kind := range []ControllerStreamEventKindV1{ControllerStreamEventWorkloadExitV1, ControllerStreamEventTerminatingV1, ControllerStreamEventWorkloadOutputsFinalizedV1} { + if event := readControllerBrokerJSONLineV1(t, public); event["type"] != string(kind) { + t.Fatalf("public event = %#v, want %q", event, kind) + } + } + terminalEnd, err := ReadTerminalEventV1(terminal) + if err != nil || terminalEnd.Kind != TerminalEventEndV1 || terminalEnd.Status == nil || terminalEnd.Status.Kind != WorkloadOutputFinalizationDrainedV1 { + t.Fatalf("terminal end = %#v, %v", terminalEnd, err) + } + writeControllerBrokerPublicRequestV1(t, publicInputWriter, ControllerStreamRequestCompleteV1, 0, 0) + terminated := readControllerBrokerJSONLineV1(t, public) + if terminated["type"] != string(ControllerStreamEventTerminatedV1) || terminated["result"] == nil { + t.Fatalf("terminated public event = %#v", terminated) + } + writeControllerBrokerPublicRequestV1(t, publicInputWriter, ControllerStreamRequestAcknowledgeTerminatedV1, 0, 0) + select { + case err := <-brokerDone: + if err != nil { + t.Fatal(err) + } + case <-time.After(15 * time.Second): + t.Fatal("broker did not exit after terminated acknowledgement") + } + if err := <-hostDone; err != nil { + t.Fatal(err) + } + entries, err := os.ReadDir(temporaryHome) + if err != nil || len(entries) != 0 { + t.Fatalf("temporary home after broker exit = %#v, %v", entries, err) + } +} + +func TestRunControllerBrokerV1TimesOutWaitingForAttachment(t *testing.T) { + hostListener, hostSocket := newControllerBrokerHostListenerV1(t) + defer hostListener.Close() + hostDone := make(chan error, 1) + go func() { + connection, err := hostListener.AcceptUnix() + if err != nil { + hostDone <- err + return + } + defer connection.Close() + if err := WriteEventV1(connection, EventV1{Kind: EventOpenedV1, Opened: pointerToOpenedV1(testOpenedV1())}); err != nil { + hostDone <- err + return + } + _, err = ReadRequestV1(connection) + hostDone <- err + }() + var output bytes.Buffer + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + temporaryHome := shortControllerBrokerTempHomeV1(t) + err := RunControllerBrokerV1(ctx, ControllerBrokerOptionsV1{ + SessionSocket: hostSocket, + TemporaryHome: temporaryHome, + Input: strings.NewReader(""), + Output: &output, + AttachTimeout: 20 * time.Millisecond, + }) + if err == nil || !strings.Contains(err.Error(), "deadline") { + t.Fatalf("attach timeout error = %v", err) + } + if !strings.Contains(output.String(), `"type":"client-error"`) || !strings.Contains(output.String(), `"code":"attach_timeout"`) { + t.Fatalf("attach timeout output = %q", output.String()) + } + if entries, readErr := os.ReadDir(temporaryHome); readErr != nil || len(entries) != 0 { + t.Fatalf("temporary home after timeout = %#v, %v", entries, readErr) + } + select { + case <-hostDone: + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } +} + +func TestRunControllerBrokerV1StartsAttachmentDeadlineAtBrokerReady(t *testing.T) { + hostListener, hostSocket := newControllerBrokerHostListenerV1(t) + defer hostListener.Close() + hostDone := make(chan struct{}) + go func() { + defer close(hostDone) + connection, err := hostListener.AcceptUnix() + if err != nil { + return + } + defer connection.Close() + _, _ = io.Copy(io.Discard, connection) + }() + var output bytes.Buffer + started := time.Now() + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + err := RunControllerBrokerV1(ctx, ControllerBrokerOptionsV1{ + SessionSocket: hostSocket, + TemporaryHome: shortControllerBrokerTempHomeV1(t), + Input: strings.NewReader(""), + Output: &output, + AttachTimeout: 50 * time.Millisecond, + }) + if err == nil || !strings.Contains(err.Error(), "deadline") { + t.Fatalf("attachment deadline error = %v", err) + } + if elapsed := time.Since(started); elapsed >= 200*time.Millisecond { + t.Fatalf("attachment deadline began too late: %s", elapsed) + } + if !strings.Contains(output.String(), `"code":"attach_timeout"`) { + t.Fatalf("attachment deadline output = %q", output.String()) + } + <-hostDone +} + +func TestRunControllerBrokerV1FailsWhenAttachmentIsLost(t *testing.T) { + hostListener, hostSocket := newControllerBrokerHostListenerV1(t) + defer hostListener.Close() + hostDone := make(chan struct{}) + go func() { + defer close(hostDone) + connection, err := hostListener.AcceptUnix() + if err != nil { + return + } + defer connection.Close() + _ = WriteEventV1(connection, EventV1{Kind: EventOpenedV1, Opened: pointerToOpenedV1(testOpenedV1())}) + _ = WriteEventV1(connection, EventV1{Kind: EventReadyV1}) + _, _ = ReadRequestV1(connection) + }() + publicInputReader, publicInputWriter := io.Pipe() + defer publicInputWriter.Close() + publicOutputReader, publicOutputWriter := io.Pipe() + defer publicOutputReader.Close() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + brokerDone := make(chan error, 1) + temporaryHome := shortControllerBrokerTempHomeV1(t) + go func() { + err := RunControllerBrokerV1(ctx, ControllerBrokerOptionsV1{ + SessionSocket: hostSocket, + TemporaryHome: temporaryHome, + Input: publicInputReader, + Output: publicOutputWriter, + }) + _ = publicOutputWriter.CloseWithError(err) + brokerDone <- err + }() + public := bufio.NewReader(publicOutputReader) + brokerReady := readControllerBrokerJSONLineV1(t, public) + _ = readControllerBrokerJSONLineV1(t, public) + terminal, err := net.Dial("unix", brokerReady["terminal_socket"].(string)) + if err != nil { + t.Fatal(err) + } + if err := terminal.Close(); err != nil { + t.Fatal(err) + } + for { + event := readControllerBrokerJSONLineV1(t, public) + if event["type"] == string(ControllerStreamEventClientErrorV1) { + if event["code"] != "terminal_attachment_error" { + t.Fatalf("socket-loss event = %#v", event) + } + break + } + } + select { + case err := <-brokerDone: + if err == nil { + t.Fatal("attachment loss returned nil") + } + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + <-hostDone +} + +func TestRunControllerBrokerV1FailsWhenAttachmentIsLostBeforeHostOpened(t *testing.T) { + hostListener, hostSocket := newControllerBrokerHostListenerV1(t) + defer hostListener.Close() + hostDone := make(chan struct{}) + go func() { + defer close(hostDone) + connection, err := hostListener.AcceptUnix() + if err != nil { + return + } + defer connection.Close() + _, _ = io.Copy(io.Discard, connection) + }() + publicInputReader, publicInputWriter := io.Pipe() + defer publicInputWriter.Close() + publicOutputReader, publicOutputWriter := io.Pipe() + defer publicOutputReader.Close() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + brokerDone := make(chan error, 1) + go func() { + err := RunControllerBrokerV1(ctx, ControllerBrokerOptionsV1{ + SessionSocket: hostSocket, + TemporaryHome: shortControllerBrokerTempHomeV1(t), + Input: publicInputReader, + Output: publicOutputWriter, + }) + _ = publicOutputWriter.CloseWithError(err) + brokerDone <- err + }() + public := bufio.NewReader(publicOutputReader) + brokerReady := readControllerBrokerJSONLineV1(t, public) + terminal, err := net.Dial("unix", brokerReady["terminal_socket"].(string)) + if err != nil { + t.Fatal(err) + } + if err := terminal.Close(); err != nil { + t.Fatal(err) + } + event := readControllerBrokerJSONLineV1(t, public) + if event["type"] != string(ControllerStreamEventClientErrorV1) || event["code"] != "terminal_attachment_error" { + t.Fatalf("pre-open attachment-loss event = %#v", event) + } + select { + case err := <-brokerDone: + if err == nil { + t.Fatal("pre-open attachment loss returned nil") + } + case <-ctx.Done(): + t.Fatal("broker did not detect attachment loss before host opened") + } + <-hostDone +} + +func TestRunControllerBrokerV1FailsWhenAttachmentIsLostAfterPreReadyInput(t *testing.T) { + hostListener, hostSocket := newControllerBrokerHostListenerV1(t) + defer hostListener.Close() + hostDone := make(chan struct{}) + go func() { + defer close(hostDone) + connection, err := hostListener.AcceptUnix() + if err != nil { + return + } + defer connection.Close() + _ = WriteEventV1(connection, EventV1{Kind: EventOpenedV1, Opened: pointerToOpenedV1(testOpenedV1())}) + _, _ = io.Copy(io.Discard, connection) + }() + publicInputReader, publicInputWriter := io.Pipe() + defer publicInputWriter.Close() + publicOutputReader, publicOutputWriter := io.Pipe() + defer publicOutputReader.Close() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + brokerDone := make(chan error, 1) + go func() { + err := RunControllerBrokerV1(ctx, ControllerBrokerOptionsV1{ + SessionSocket: hostSocket, + TemporaryHome: shortControllerBrokerTempHomeV1(t), + Input: publicInputReader, + Output: publicOutputWriter, + }) + _ = publicOutputWriter.CloseWithError(err) + brokerDone <- err + }() + public := bufio.NewReader(publicOutputReader) + brokerReady := readControllerBrokerJSONLineV1(t, public) + _ = readControllerBrokerJSONLineV1(t, public) + terminal, err := net.Dial("unix", brokerReady["terminal_socket"].(string)) + if err != nil { + t.Fatal(err) + } + if err := WriteTerminalRequestV1(terminal, RequestV1{Kind: RequestInputV1, Bytes: []byte("typed during startup")}); err != nil { + t.Fatal(err) + } + if err := terminal.Close(); err != nil { + t.Fatal(err) + } + for { + event := readControllerBrokerJSONLineV1(t, public) + if event["type"] == string(ControllerStreamEventClientErrorV1) { + if event["code"] != "terminal_attachment_error" { + t.Fatalf("socket-loss event = %#v", event) + } + break + } + } + select { + case err := <-brokerDone: + if err == nil { + t.Fatal("pre-ready attachment loss returned nil") + } + case <-ctx.Done(): + t.Fatal("broker did not detect pre-ready attachment loss") + } + <-hostDone +} + +func TestRunClaimedControllerBrokerV1ProcessesTerminateWhileTerminalOutputBlocks(t *testing.T) { + brokerConnection, attachmentConnection := net.Pipe() + defer brokerConnection.Close() + defer attachmentConnection.Close() + writeStarted := make(chan struct{}) + terminal := &ControllerTerminalConnectionV1{connection: &controllerBrokerSignalingWriteConnectionV1{Conn: brokerConnection, writeStarted: writeStarted}} + publicInputReader, publicInputWriter := io.Pipe() + defer publicInputWriter.Close() + publicOutputReader, publicOutputWriter := io.Pipe() + defer publicOutputReader.Close() + streamReader, err := NewControllerStreamReaderV1(publicInputReader) + if err != nil { + t.Fatal(err) + } + streamWriter, err := NewControllerStreamWriterV1(publicOutputWriter) + if err != nil { + t.Fatal(err) + } + session := newControllerBrokerChannelSessionV1() + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + done := make(chan error, 1) + requests := make(chan controllerBrokerRequestV1) + inputErrors := make(chan controllerBrokerInputErrorV1, 2) + go readControllerBrokerTerminalRequestsV1(ctx, terminal, requests, inputErrors) + go func() { + done <- runClaimedControllerBrokerV1(ctx, session, terminal, streamReader, streamWriter, requests, inputErrors, controllerBrokerPendingTerminalV1{}, []OperationV1{OperationTerminateV1}, 100*time.Millisecond, 100*time.Millisecond) + }() + public := bufio.NewReader(publicOutputReader) + session.events <- EventV1{Kind: EventReadyV1} + if event := readControllerBrokerJSONLineV1(t, public); event["type"] != string(ControllerStreamEventReadyV1) { + t.Fatalf("ready public event = %#v", event) + } + go func() { _, _ = io.Copy(io.Discard, public) }() + session.events <- EventV1{Kind: EventOutputV1, Bytes: []byte{}} + select { + case <-writeStarted: + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + writeControllerBrokerPublicRequestV1(t, publicInputWriter, ControllerStreamRequestTerminateV1, 0, 0) + select { + case request := <-session.requests: + if request.Kind != RequestTerminateV1 { + t.Fatalf("request while output blocked = %#v", request) + } + case <-ctx.Done(): + t.Fatal("terminate was not forwarded while terminal output was blocked") + } + select { + case err := <-done: + if err == nil { + t.Fatalf("blocked terminal output error = %v", err) + } + case <-time.After(time.Second): + t.Fatal("broker did not time out blocked terminal output") + } +} + +func TestControllerBrokerStateV1RejectsPrematureDuplicateAndUnauthorizedRequests(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + session := &controllerBrokerFakeSessionV1{} + state := controllerBrokerStateV1{operations: map[OperationV1]bool{OperationCompleteV1: true, OperationTerminateV1: true}} + publicComplete := controllerBrokerRequestV1{source: controllerBrokerRequestPublicV1, request: RequestV1{Kind: RequestCompleteV1}} + if err := state.applyRequest(ctx, session, publicComplete); err == nil || !strings.Contains(err.Error(), "before ready") { + t.Fatalf("premature complete error = %v", err) + } + state.ready = true + unauthorizedResize := controllerBrokerRequestV1{source: controllerBrokerRequestPublicV1, request: RequestV1{Kind: RequestResizeV1, Columns: 80, Rows: 24}} + if err := state.applyRequest(ctx, session, unauthorizedResize); err == nil || !strings.Contains(err.Error(), "not granted") { + t.Fatalf("unauthorized resize error = %v", err) + } + state.terminating = true + state.outputsFinalized = true + if err := state.applyRequest(ctx, session, publicComplete); err != nil { + t.Fatal(err) + } + if err := state.applyRequest(ctx, session, publicComplete); err == nil || !strings.Contains(err.Error(), "once") { + t.Fatalf("duplicate complete error = %v", err) + } + state.terminated = true + ack := controllerBrokerRequestV1{source: controllerBrokerRequestPublicV1, request: RequestV1{Kind: RequestAcknowledgeTerminatedV1}} + if err := state.applyRequest(ctx, session, ack); err != nil { + t.Fatal(err) + } + if err := state.applyRequest(ctx, session, ack); err == nil || !strings.Contains(err.Error(), "exactly once") { + t.Fatalf("duplicate acknowledgement error = %v", err) + } + if got := session.requestKinds(); strings.Join(got, ",") != "complete,acknowledge-terminated" { + t.Fatalf("forwarded requests = %#v", got) + } +} + +func TestControllerBrokerStateV1BuffersLatestTerminalResizeUntilReady(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + session := &controllerBrokerFakeSessionV1{} + state := controllerBrokerStateV1{operations: map[OperationV1]bool{OperationResizeV1: true}} + for _, request := range []RequestV1{ + {Kind: RequestResizeV1, Columns: 80, Rows: 24}, + {Kind: RequestResizeV1, Columns: 120, Rows: 40}, + } { + if err := state.applyRequest(ctx, session, controllerBrokerRequestV1{source: controllerBrokerRequestTerminalV1, request: request}); err != nil { + t.Fatal(err) + } + } + if got := session.requestKinds(); len(got) != 0 { + t.Fatalf("pre-ready forwarded requests = %#v", got) + } + var publicOutput bytes.Buffer + writer, err := NewControllerStreamWriterV1(&publicOutput) + if err != nil { + t.Fatal(err) + } + if err := state.applyHostEvent(ctx, session, nil, writer, EventV1{Kind: EventReadyV1}); err != nil { + t.Fatal(err) + } + session.mu.Lock() + requests := append([]RequestV1(nil), session.requests...) + session.mu.Unlock() + if len(requests) != 1 || requests[0].Kind != RequestResizeV1 || requests[0].Columns != 120 || requests[0].Rows != 40 { + t.Fatalf("ready forwarded requests = %#v", requests) + } + if state.pendingTerminalResize != nil || !state.ready || !strings.Contains(publicOutput.String(), `"type":"ready"`) { + t.Fatalf("ready state = %#v, public output = %q", state, publicOutput.String()) + } +} + +func TestControllerBrokerStateV1HoldsTerminalInputUntilReady(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + session := &controllerBrokerFakeSessionV1{} + state := controllerBrokerStateV1{operations: map[OperationV1]bool{OperationInputV1: true}} + input := controllerBrokerRequestV1{ + source: controllerBrokerRequestTerminalV1, + request: RequestV1{Kind: RequestInputV1, Bytes: []byte("typed during startup")}, + } + if err := state.applyRequest(ctx, session, input); err != nil { + t.Fatal(err) + } + if requests := session.requestKinds(); len(requests) != 0 { + t.Fatalf("input forwarded before ready: %#v", requests) + } + var publicOutput bytes.Buffer + writer, err := NewControllerStreamWriterV1(&publicOutput) + if err != nil { + t.Fatal(err) + } + if err := state.applyHostEvent(ctx, session, nil, writer, EventV1{Kind: EventReadyV1}); err != nil { + t.Fatal(err) + } + session.mu.Lock() + requests := append([]RequestV1(nil), session.requests...) + session.mu.Unlock() + if len(requests) != 1 || requests[0].Kind != RequestInputV1 || string(requests[0].Bytes) != "typed during startup" { + t.Fatalf("forwarded input = %#v", requests) + } + if state.pendingTerminalInput != nil || !state.ready || !strings.Contains(publicOutput.String(), `"type":"ready"`) { + t.Fatalf("ready state = %#v, public output = %q", state, publicOutput.String()) + } +} + +func TestControllerBrokerPendingTerminalV1PreservesEmptyInput(t *testing.T) { + var pending controllerBrokerPendingTerminalV1 + if err := pending.add(controllerBrokerRequestV1{ + source: controllerBrokerRequestTerminalV1, + request: RequestV1{Kind: RequestInputV1, Bytes: []byte{}}, + }); err != nil { + t.Fatal(err) + } + requests := pending.requests() + if len(requests) != 1 || requests[0].request.Kind != RequestInputV1 || requests[0].request.Bytes == nil || len(requests[0].request.Bytes) != 0 { + t.Fatalf("pending empty input = %#v", requests) + } +} + +func TestControllerBrokerStateV1BoundsHostRequestWrites(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + state := controllerBrokerStateV1{ + operations: map[OperationV1]bool{OperationInputV1: true}, + requestWriteTimeout: 20 * time.Millisecond, + ready: true, + } + input := controllerBrokerRequestV1{ + source: controllerBrokerRequestTerminalV1, + request: RequestV1{Kind: RequestInputV1, Bytes: []byte("blocked")}, + } + started := time.Now() + err := state.applyRequest(ctx, &controllerBrokerBlockingSessionV1{}, input) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("blocked host request error = %v", err) + } + if elapsed := time.Since(started); elapsed >= 200*time.Millisecond { + t.Fatalf("blocked host request exceeded bound: %s", elapsed) + } +} + +func TestControllerBrokerStateV1IgnoresPreReadyResizeAfterTerminationBegins(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + input := controllerBrokerRequestV1{ + source: controllerBrokerRequestTerminalV1, + request: RequestV1{Kind: RequestResizeV1, Columns: 120, Rows: 40}, + } + for _, state := range []*controllerBrokerStateV1{ + {operations: map[OperationV1]bool{OperationResizeV1: true}, terminating: true}, + {operations: map[OperationV1]bool{OperationResizeV1: true}, terminated: true}, + } { + session := &controllerBrokerFakeSessionV1{} + if err := state.applyRequest(ctx, session, input); err != nil { + t.Fatal(err) + } + if state.pendingTerminalResize != nil || len(session.requestKinds()) != 0 { + t.Fatalf("late resize changed broker state: %#v", state) + } + } +} + +func TestControllerBrokerStateV1IgnoresTerminalRequestsAfterTerminationBegins(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + requests := []RequestV1{ + {Kind: RequestInputV1, Bytes: []byte("late input")}, + {Kind: RequestResizeV1, Columns: 120, Rows: 40}, + } + for _, terminated := range []bool{false, true} { + state := controllerBrokerStateV1{ + operations: map[OperationV1]bool{OperationInputV1: true, OperationResizeV1: true}, + ready: true, + terminating: true, + terminated: terminated, + } + session := &controllerBrokerFakeSessionV1{} + for _, request := range requests { + input := controllerBrokerRequestV1{source: controllerBrokerRequestTerminalV1, request: request} + if err := state.applyRequest(ctx, session, input); err != nil { + t.Fatalf("terminated=%t request=%s: %v", terminated, request.Kind, err) + } + } + if got := session.requestKinds(); len(got) != 0 { + t.Fatalf("terminated=%t forwarded requests = %#v", terminated, got) + } + } +} + +func TestControllerBrokerStateV1RejectsPublicResizeWhileTerminating(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + state := controllerBrokerStateV1{ + operations: map[OperationV1]bool{OperationResizeV1: true}, + ready: true, + terminating: true, + } + input := controllerBrokerRequestV1{ + source: controllerBrokerRequestPublicV1, + request: RequestV1{Kind: RequestResizeV1, Columns: 120, Rows: 40}, + } + if err := state.applyRequest(ctx, &controllerBrokerFakeSessionV1{}, input); err == nil || !strings.Contains(err.Error(), "while terminating") { + t.Fatalf("public resize error = %v", err) + } +} + +func TestControllerBrokerStateV1AcceptsWorkloadExitBeforeReady(t *testing.T) { + code := 1 + state := controllerBrokerStateV1{terminating: true} + var publicOutput bytes.Buffer + writer, err := NewControllerStreamWriterV1(&publicOutput) + if err != nil { + t.Fatal(err) + } + event := EventV1{Kind: EventWorkloadExitV1, WorkloadExit: &WorkloadExitV1{Status: ProcessStatusV1{Kind: ProcessStatusExitedV1, Code: &code}}} + if err := state.applyHostEvent(context.Background(), nil, nil, writer, event); err != nil { + t.Fatal(err) + } + if !state.workloadExited || state.ready || !strings.Contains(publicOutput.String(), `"type":"workload-exit"`) { + t.Fatalf("pre-ready workload-exit state = %#v, public output = %q", state, publicOutput.String()) + } +} + +func TestControllerBrokerStateV1PreservesPreActivationOutputFinalizationFailure(t *testing.T) { + brokerConnection, attachmentConnection := net.Pipe() + defer brokerConnection.Close() + defer attachmentConnection.Close() + terminal := &ControllerTerminalConnectionV1{connection: brokerConnection} + var publicOutput bytes.Buffer + writer, err := NewControllerStreamWriterV1(&publicOutput) + if err != nil { + t.Fatal(err) + } + result := ResultV1{ + Cause: CauseStartupFailureV1, + WorkloadStatus: ProcessStatusV1{Kind: ProcessStatusUnknownV1}, + WorkloadOutputFinalizationStatus: WorkloadOutputFinalizationStatusV1{Kind: WorkloadOutputFinalizationFailedV1, Reason: "startup output unavailable"}, + RuntimeObservationStatus: RuntimeObservationStatusV1{Kind: RuntimeObservationMaintainedV1}, + ControllerFinalizationStatus: ControllerFinalizationStatusV1{Kind: ControllerFinalizationStartupFailedV1, Reason: "workload did not start"}, + CleanupStatus: CleanupStatusV1{Kind: CleanupStatusSucceededV1}, + RecoveryAction: RecoveryNoneV1, + } + state := controllerBrokerStateV1{terminalWriteTimeout: time.Second} + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + done := make(chan error, 1) + go func() { + done <- state.applyHostEvent(ctx, nil, terminal, writer, EventV1{Kind: EventTerminatedV1, Terminated: &result}) + }() + terminalEnd, err := ReadTerminalEventV1(attachmentConnection) + if err != nil { + t.Fatal(err) + } + if terminalEnd.Kind != TerminalEventEndV1 || terminalEnd.Status == nil || terminalEnd.Status.Kind != WorkloadOutputFinalizationFailedV1 || terminalEnd.Status.Reason != "startup output unavailable" { + t.Fatalf("pre-activation terminal end = %#v", terminalEnd) + } + if err := <-done; err != nil { + t.Fatal(err) + } + if !state.terminalEnded || !state.outputsFinalized || !state.terminated || !strings.Contains(publicOutput.String(), `"type":"terminated"`) { + t.Fatalf("pre-activation terminal state = %#v, public output = %q", state, publicOutput.String()) + } +} + +func TestControllerBrokerStateV1BoundsTerminalEndWrites(t *testing.T) { + result := ResultV1{ + Cause: CauseStartupFailureV1, + WorkloadStatus: ProcessStatusV1{Kind: ProcessStatusUnknownV1}, + WorkloadOutputFinalizationStatus: WorkloadOutputFinalizationStatusV1{Kind: WorkloadOutputFinalizationFailedV1, Reason: "startup output unavailable"}, + RuntimeObservationStatus: RuntimeObservationStatusV1{Kind: RuntimeObservationMaintainedV1}, + ControllerFinalizationStatus: ControllerFinalizationStatusV1{Kind: ControllerFinalizationStartupFailedV1, Reason: "workload did not start"}, + CleanupStatus: CleanupStatusV1{Kind: CleanupStatusSucceededV1}, + RecoveryAction: RecoveryNoneV1, + } + tests := []struct { + name string + state controllerBrokerStateV1 + event EventV1 + }{ + { + name: "outputs finalized", + state: controllerBrokerStateV1{ + ready: true, + workloadExited: true, + terminating: true, + terminalWriteTimeout: 20 * time.Millisecond, + }, + event: EventV1{ + Kind: EventWorkloadOutputsFinalizedV1, + WorkloadOutputsFinalized: &WorkloadOutputsFinalizedV1{ + Status: WorkloadOutputFinalizationDrainedV1, + }, + }, + }, + { + name: "pre-activation terminated", + state: controllerBrokerStateV1{ + terminalWriteTimeout: 20 * time.Millisecond, + }, + event: EventV1{Kind: EventTerminatedV1, Terminated: &result}, + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + brokerConnection, attachmentConnection := net.Pipe() + defer attachmentConnection.Close() + terminal := &ControllerTerminalConnectionV1{connection: brokerConnection} + var publicOutput bytes.Buffer + writer, err := NewControllerStreamWriterV1(&publicOutput) + if err != nil { + t.Fatal(err) + } + started := time.Now() + err = test.state.applyHostEvent(context.Background(), nil, terminal, writer, test.event) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("blocked terminal-end write error = %v", err) + } + if elapsed := time.Since(started); elapsed > time.Second { + t.Fatalf("blocked terminal-end write took %s", elapsed) + } + }) + } +} + +type controllerBrokerFakeSessionV1 struct { + mu sync.Mutex + requests []RequestV1 +} + +type controllerBrokerBlockingSessionV1 struct{} + +func (*controllerBrokerBlockingSessionV1) Opened() OpenedV1 { return testOpenedV1() } +func (*controllerBrokerBlockingSessionV1) ReadEvent(context.Context) (EventV1, error) { + return EventV1{}, io.EOF +} +func (*controllerBrokerBlockingSessionV1) WriteRequest(ctx context.Context, _ RequestV1) error { + <-ctx.Done() + return ctx.Err() +} +func (*controllerBrokerBlockingSessionV1) Close() error { return nil } + +type controllerBrokerChannelSessionV1 struct { + events chan EventV1 + requests chan RequestV1 +} + +func newControllerBrokerChannelSessionV1() *controllerBrokerChannelSessionV1 { + return &controllerBrokerChannelSessionV1{events: make(chan EventV1), requests: make(chan RequestV1, 1)} +} + +func (*controllerBrokerChannelSessionV1) Opened() OpenedV1 { return testOpenedV1() } +func (session *controllerBrokerChannelSessionV1) ReadEvent(ctx context.Context) (EventV1, error) { + select { + case event := <-session.events: + return event, nil + case <-ctx.Done(): + return EventV1{}, ctx.Err() + } +} +func (session *controllerBrokerChannelSessionV1) WriteRequest(ctx context.Context, request RequestV1) error { + select { + case session.requests <- request: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} +func (*controllerBrokerChannelSessionV1) Close() error { return nil } + +type controllerBrokerSignalingWriteConnectionV1 struct { + net.Conn + writeStarted chan struct{} + once sync.Once +} + +func (connection *controllerBrokerSignalingWriteConnectionV1) Write(content []byte) (int, error) { + connection.once.Do(func() { close(connection.writeStarted) }) + return connection.Conn.Write(content) +} + +func (*controllerBrokerFakeSessionV1) Opened() OpenedV1 { return testOpenedV1() } +func (*controllerBrokerFakeSessionV1) ReadEvent(context.Context) (EventV1, error) { + return EventV1{}, io.EOF +} +func (session *controllerBrokerFakeSessionV1) WriteRequest(_ context.Context, request RequestV1) error { + session.mu.Lock() + defer session.mu.Unlock() + session.requests = append(session.requests, request) + return nil +} +func (*controllerBrokerFakeSessionV1) Close() error { return nil } +func (session *controllerBrokerFakeSessionV1) requestKinds() []string { + session.mu.Lock() + defer session.mu.Unlock() + result := make([]string, len(session.requests)) + for index, request := range session.requests { + result[index] = string(request.Kind) + } + return result +} + +func newControllerBrokerHostListenerV1(t *testing.T) (*net.UnixListener, string) { + t.Helper() + directory, err := os.MkdirTemp("/tmp", "rph-") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = os.RemoveAll(directory) }) + socket := filepath.Join(directory, "host.sock") + listener, err := net.ListenUnix("unix", &net.UnixAddr{Name: socket, Net: "unix"}) + if err != nil { + t.Fatal(err) + } + return listener, socket +} + +func shortControllerBrokerTempHomeV1(t *testing.T) string { + t.Helper() + home, err := os.MkdirTemp("/tmp", "rpb-") + if err != nil { + t.Fatal(err) + } + if err := os.Chmod(home, 0o700); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = os.RemoveAll(home) }) + return home +} + +func runControllerBrokerTestHostV1(listener *net.UnixListener, done chan<- error) { + done <- runControllerBrokerTestHostConnectionV1(listener) +} + +func runControllerBrokerTestHostConnectionV1(listener *net.UnixListener) (resultErr error) { + connection, err := listener.AcceptUnix() + if err != nil { + return err + } + defer func() { resultErr = errors.Join(resultErr, connection.Close()) }() + code := 0 + result := ResultV1{ + Cause: CauseWorkloadExitV1, + WorkloadStatus: ProcessStatusV1{Kind: ProcessStatusExitedV1, Code: &code}, + WorkloadOutputFinalizationStatus: WorkloadOutputFinalizationStatusV1{Kind: WorkloadOutputFinalizationDrainedV1}, + RuntimeObservationStatus: RuntimeObservationStatusV1{Kind: RuntimeObservationMaintainedV1}, + ControllerFinalizationStatus: ControllerFinalizationStatusV1{Kind: ControllerFinalizationCompletedV1}, + CleanupStatus: CleanupStatusV1{Kind: CleanupStatusSucceededV1}, + RecoveryAction: RecoveryNoneV1, + } + events := []EventV1{ + {Kind: EventOpenedV1, Opened: pointerToOpenedV1(testOpenedV1())}, + {Kind: EventOutputV1, Bytes: []byte("hello from workload")}, + {Kind: EventReadyV1}, + {Kind: EventWorkloadExitV1, WorkloadExit: &WorkloadExitV1{Status: result.WorkloadStatus}}, + {Kind: EventTerminatingV1, Terminating: &TerminatingV1{Cause: CauseWorkloadExitV1}}, + {Kind: EventWorkloadOutputsFinalizedV1, WorkloadOutputsFinalized: &WorkloadOutputsFinalizedV1{Status: WorkloadOutputFinalizationDrainedV1}}, + } + for _, event := range events { + if err := WriteEventV1(connection, event); err != nil { + return err + } + } + request, err := ReadRequestV1(connection) + if err != nil || request.Kind != RequestCompleteV1 { + return errors.Join(err, errors.New("host expected complete request")) + } + if err := WriteEventV1(connection, EventV1{Kind: EventTerminatedV1, Terminated: &result}); err != nil { + return err + } + request, err = ReadRequestV1(connection) + if err != nil || request.Kind != RequestAcknowledgeTerminatedV1 { + return errors.Join(err, errors.New("host expected acknowledge-terminated request")) + } + return nil +} + +func readControllerBrokerJSONLineV1(t *testing.T, reader *bufio.Reader) map[string]any { + t.Helper() + line, err := reader.ReadBytes('\n') + if err != nil { + t.Fatal(err) + } + var value map[string]any + if err := json.Unmarshal(line, &value); err != nil { + t.Fatal(err) + } + return value +} + +func writeControllerBrokerPublicRequestV1(t *testing.T, writer io.Writer, kind ControllerStreamRequestKindV1, columns uint32, rows uint32) { + t.Helper() + message := map[string]any{"schema": ControllerStreamSchemaV1, "type": kind} + if kind == ControllerStreamRequestResizeV1 { + message["columns"] = columns + message["rows"] = rows + } + payload, err := json.Marshal(message) + if err != nil { + t.Fatal(err) + } + payload = append(payload, '\n') + if _, err := writer.Write(payload); err != nil { + t.Fatal(err) + } +} diff --git a/internal/controlledsession/controller_stream.go b/internal/controlledsession/controller_stream.go new file mode 100644 index 00000000..3724b7fa --- /dev/null +++ b/internal/controlledsession/controller_stream.go @@ -0,0 +1,281 @@ +package controlledsession + +import ( + "bufio" + "bytes" + "encoding/json" + "fmt" + "io" + "sync" + "unicode/utf8" +) + +const ( + ControllerStreamSchemaV1 = "reploy-controlled-session-client-v1" + MaxControllerStreamLineV1 = 1 << 20 +) + +type ControllerStreamRequestKindV1 string + +const ( + ControllerStreamRequestResizeV1 ControllerStreamRequestKindV1 = "resize" + ControllerStreamRequestTerminateV1 ControllerStreamRequestKindV1 = "terminate" + ControllerStreamRequestCompleteV1 ControllerStreamRequestKindV1 = "complete" + ControllerStreamRequestAcknowledgeTerminatedV1 ControllerStreamRequestKindV1 = "acknowledge-terminated" +) + +type ControllerStreamRequestV1 struct { + Kind ControllerStreamRequestKindV1 + Columns uint32 + Rows uint32 +} + +type ControllerStreamEventKindV1 string + +const ( + ControllerStreamEventBrokerReadyV1 ControllerStreamEventKindV1 = "broker-ready" + ControllerStreamEventOpenedV1 ControllerStreamEventKindV1 = "opened" + ControllerStreamEventReadyV1 ControllerStreamEventKindV1 = "ready" + ControllerStreamEventWorkloadExitV1 ControllerStreamEventKindV1 = "workload-exit" + ControllerStreamEventTerminatingV1 ControllerStreamEventKindV1 = "terminating" + ControllerStreamEventDiagnosticV1 ControllerStreamEventKindV1 = "diagnostic" + ControllerStreamEventWorkloadOutputsFinalizedV1 ControllerStreamEventKindV1 = "workload-outputs-finalized" + ControllerStreamEventTerminatedV1 ControllerStreamEventKindV1 = "terminated" + ControllerStreamEventClientErrorV1 ControllerStreamEventKindV1 = "client-error" +) + +type ControllerStreamOpenedV1 struct { + Operations []OperationV1 `json:"operations"` + Endpoints []EndpointV1 `json:"endpoints"` + Columns uint32 `json:"columns"` + Rows uint32 `json:"rows"` + OutputFinalizationTimeoutMilliseconds uint32 `json:"output_finalization_timeout_milliseconds"` +} + +type ControllerStreamEventV1 struct { + Kind ControllerStreamEventKindV1 + TerminalSocket string + Opened *ControllerStreamOpenedV1 + WorkloadExit *WorkloadExitV1 + Terminating *TerminatingV1 + Diagnostic *DiagnosticV1 + WorkloadOutputsFinalized *WorkloadOutputsFinalizedV1 + Terminated *ResultV1 + ClientError *DiagnosticV1 +} + +type ControllerStreamReaderV1 struct { + reader *bufio.Reader +} + +type ControllerStreamWriterV1 struct { + writer io.Writer + mu sync.Mutex +} + +func NewControllerStreamReaderV1(reader io.Reader) (*ControllerStreamReaderV1, error) { + if reader == nil { + return nil, fmt.Errorf("create controlled-session controller stream reader: input is required") + } + return &ControllerStreamReaderV1{reader: bufio.NewReaderSize(reader, MaxControllerStreamLineV1+1)}, nil +} + +func NewControllerStreamWriterV1(writer io.Writer) (*ControllerStreamWriterV1, error) { + if writer == nil { + return nil, fmt.Errorf("create controlled-session controller stream writer: output is required") + } + return &ControllerStreamWriterV1{writer: writer}, nil +} + +func (reader *ControllerStreamReaderV1) ReadRequest() (ControllerStreamRequestV1, error) { + line, err := reader.reader.ReadSlice('\n') + if err != nil { + if err == bufio.ErrBufferFull || len(line) > MaxControllerStreamLineV1 { + return ControllerStreamRequestV1{}, fmt.Errorf("controlled-session controller message exceeds %d bytes", MaxControllerStreamLineV1) + } + if err == io.EOF && len(line) != 0 { + return ControllerStreamRequestV1{}, fmt.Errorf("controlled-session controller message is not newline terminated") + } + return ControllerStreamRequestV1{}, err + } + if len(line) > MaxControllerStreamLineV1 { + return ControllerStreamRequestV1{}, fmt.Errorf("controlled-session controller message exceeds %d bytes", MaxControllerStreamLineV1) + } + payload := line[:len(line)-1] + if len(payload) == 0 { + return ControllerStreamRequestV1{}, fmt.Errorf("controlled-session controller message must not be empty") + } + if !utf8.Valid(payload) { + return ControllerStreamRequestV1{}, fmt.Errorf("controlled-session controller message is not valid UTF-8 JSON") + } + if !bytes.Equal(bytes.TrimSpace(payload), payload) { + return ControllerStreamRequestV1{}, fmt.Errorf("controlled-session controller message must contain exactly one JSON object") + } + var envelope struct { + Schema string `json:"schema"` + Type ControllerStreamRequestKindV1 `json:"type"` + } + if err := rejectDuplicateJSONFieldsV1(payload); err != nil { + return ControllerStreamRequestV1{}, fmt.Errorf("decode controlled-session controller message envelope: %w", err) + } + if err := json.Unmarshal(payload, &envelope); err != nil { + return ControllerStreamRequestV1{}, fmt.Errorf("decode controlled-session controller message envelope: %w", err) + } + if envelope.Schema != ControllerStreamSchemaV1 { + return ControllerStreamRequestV1{}, fmt.Errorf("controlled-session controller message schema must be %q", ControllerStreamSchemaV1) + } + switch envelope.Type { + case ControllerStreamRequestResizeV1: + var message struct { + Schema string `json:"schema"` + Type ControllerStreamRequestKindV1 `json:"type"` + Columns uint32 `json:"columns"` + Rows uint32 `json:"rows"` + } + if err := decodeStrictJSONV1("controller resize message", payload, &message); err != nil { + return ControllerStreamRequestV1{}, err + } + request := ControllerStreamRequestV1{Kind: message.Type, Columns: message.Columns, Rows: message.Rows} + if !validDimensionsV1(request.Columns, request.Rows) { + return ControllerStreamRequestV1{}, fmt.Errorf("controlled-session controller resize requires dimensions between 1 and 65535") + } + return request, nil + case ControllerStreamRequestTerminateV1, ControllerStreamRequestCompleteV1, ControllerStreamRequestAcknowledgeTerminatedV1: + var message struct { + Schema string `json:"schema"` + Type ControllerStreamRequestKindV1 `json:"type"` + } + if err := decodeStrictJSONV1("controller request message", payload, &message); err != nil { + return ControllerStreamRequestV1{}, err + } + return ControllerStreamRequestV1{Kind: message.Type}, nil + default: + return ControllerStreamRequestV1{}, fmt.Errorf("controlled-session controller message type %q is unsupported", envelope.Type) + } +} + +func (writer *ControllerStreamWriterV1) WriteEvent(event ControllerStreamEventV1) error { + value, err := controllerStreamEventWireV1(event) + if err != nil { + return err + } + payload, err := json.Marshal(value) + if err != nil { + return fmt.Errorf("encode controlled-session controller event: %w", err) + } + payload = append(payload, '\n') + if len(payload) > MaxControllerStreamLineV1 { + return fmt.Errorf("controlled-session controller event exceeds %d bytes", MaxControllerStreamLineV1) + } + writer.mu.Lock() + defer writer.mu.Unlock() + if err := writeAllV1(writer.writer, payload); err != nil { + return fmt.Errorf("write controlled-session controller event: %w", err) + } + return nil +} + +func controllerStreamEventWireV1(event ControllerStreamEventV1) (any, error) { + base := struct { + Schema string `json:"schema"` + Type ControllerStreamEventKindV1 `json:"type"` + }{Schema: ControllerStreamSchemaV1, Type: event.Kind} + switch event.Kind { + case ControllerStreamEventBrokerReadyV1: + if event.TerminalSocket == "" || controllerStreamEventPayloadCountV1(event) != 1 { + return nil, fmt.Errorf("controlled-session broker-ready event requires exactly one terminal socket") + } + return struct { + Schema string `json:"schema"` + Type ControllerStreamEventKindV1 `json:"type"` + TerminalSocket string `json:"terminal_socket"` + }{base.Schema, base.Type, event.TerminalSocket}, nil + case ControllerStreamEventOpenedV1: + if event.Opened == nil || controllerStreamEventPayloadCountV1(event) != 1 || event.Opened.Operations == nil || event.Opened.Endpoints == nil || !validDimensionsV1(event.Opened.Columns, event.Opened.Rows) || event.Opened.OutputFinalizationTimeoutMilliseconds == 0 { + return nil, fmt.Errorf("controlled-session opened controller event is invalid") + } + return struct { + Schema string `json:"schema"` + Type ControllerStreamEventKindV1 `json:"type"` + Operations []OperationV1 `json:"operations"` + Endpoints []EndpointV1 `json:"endpoints"` + Columns uint32 `json:"columns"` + Rows uint32 `json:"rows"` + OutputFinalizationTimeoutMilliseconds uint32 `json:"output_finalization_timeout_milliseconds"` + }{base.Schema, base.Type, event.Opened.Operations, event.Opened.Endpoints, event.Opened.Columns, event.Opened.Rows, event.Opened.OutputFinalizationTimeoutMilliseconds}, nil + case ControllerStreamEventReadyV1: + if controllerStreamEventPayloadCountV1(event) != 0 { + return nil, fmt.Errorf("controlled-session ready controller event must not contain a payload") + } + return base, nil + case ControllerStreamEventWorkloadExitV1: + if event.WorkloadExit == nil || controllerStreamEventPayloadCountV1(event) != 1 { + return nil, fmt.Errorf("controlled-session workload-exit controller event is invalid") + } + return struct { + Schema string `json:"schema"` + Type ControllerStreamEventKindV1 `json:"type"` + Status ProcessStatusV1 `json:"status"` + }{base.Schema, base.Type, event.WorkloadExit.Status}, nil + case ControllerStreamEventTerminatingV1: + if event.Terminating == nil || controllerStreamEventPayloadCountV1(event) != 1 { + return nil, fmt.Errorf("controlled-session terminating controller event is invalid") + } + return struct { + Schema string `json:"schema"` + Type ControllerStreamEventKindV1 `json:"type"` + Cause TerminationCauseV1 `json:"cause"` + }{base.Schema, base.Type, event.Terminating.Cause}, nil + case ControllerStreamEventDiagnosticV1, ControllerStreamEventClientErrorV1: + diagnostic := event.Diagnostic + if event.Kind == ControllerStreamEventClientErrorV1 { + diagnostic = event.ClientError + } + if diagnostic == nil || controllerStreamEventPayloadCountV1(event) != 1 || validateProtocolCodeV1("controller event code", diagnostic.Code) != nil || validateRequiredSafeTextV1("controller event message", diagnostic.Message) != nil { + return nil, fmt.Errorf("controlled-session %s controller event is invalid", event.Kind) + } + return struct { + Schema string `json:"schema"` + Type ControllerStreamEventKindV1 `json:"type"` + Code string `json:"code"` + Message string `json:"message"` + }{base.Schema, base.Type, diagnostic.Code, diagnostic.Message}, nil + case ControllerStreamEventWorkloadOutputsFinalizedV1: + if event.WorkloadOutputsFinalized == nil || controllerStreamEventPayloadCountV1(event) != 1 { + return nil, fmt.Errorf("controlled-session workload-outputs-finalized controller event is invalid") + } + if err := validateWorkloadOutputFinalizationStatusV1(WorkloadOutputFinalizationStatusV1{Kind: event.WorkloadOutputsFinalized.Status, Reason: event.WorkloadOutputsFinalized.Reason}); err != nil { + return nil, err + } + return struct { + Schema string `json:"schema"` + Type ControllerStreamEventKindV1 `json:"type"` + Status WorkloadOutputFinalizationStatusKindV1 `json:"status"` + Reason string `json:"reason,omitempty"` + }{base.Schema, base.Type, event.WorkloadOutputsFinalized.Status, event.WorkloadOutputsFinalized.Reason}, nil + case ControllerStreamEventTerminatedV1: + if event.Terminated == nil || controllerStreamEventPayloadCountV1(event) != 1 { + return nil, fmt.Errorf("controlled-session terminated controller event is invalid") + } + if err := ValidateResultV1(*event.Terminated); err != nil { + return nil, err + } + return struct { + Schema string `json:"schema"` + Type ControllerStreamEventKindV1 `json:"type"` + Result ResultV1 `json:"result"` + }{base.Schema, base.Type, *event.Terminated}, nil + default: + return nil, fmt.Errorf("controlled-session controller event kind %q is unsupported", event.Kind) + } +} + +func controllerStreamEventPayloadCountV1(event ControllerStreamEventV1) int { + count := 0 + for _, present := range []bool{event.TerminalSocket != "", event.Opened != nil, event.WorkloadExit != nil, event.Terminating != nil, event.Diagnostic != nil, event.WorkloadOutputsFinalized != nil, event.Terminated != nil, event.ClientError != nil} { + if present { + count++ + } + } + return count +} diff --git a/internal/controlledsession/controller_stream_test.go b/internal/controlledsession/controller_stream_test.go new file mode 100644 index 00000000..5c9d7ca4 --- /dev/null +++ b/internal/controlledsession/controller_stream_test.go @@ -0,0 +1,101 @@ +package controlledsession + +import ( + "bytes" + "io" + "strings" + "testing" +) + +func TestControllerStreamReaderV1AcceptsExactRequests(t *testing.T) { + input := strings.Join([]string{ + `{"schema":"reploy-controlled-session-client-v1","type":"resize","columns":120,"rows":40}`, + `{"schema":"reploy-controlled-session-client-v1","type":"terminate"}`, + `{"schema":"reploy-controlled-session-client-v1","type":"complete"}`, + `{"schema":"reploy-controlled-session-client-v1","type":"acknowledge-terminated"}`, + }, "\n") + "\n" + reader, err := NewControllerStreamReaderV1(strings.NewReader(input)) + if err != nil { + t.Fatal(err) + } + want := []ControllerStreamRequestV1{ + {Kind: ControllerStreamRequestResizeV1, Columns: 120, Rows: 40}, + {Kind: ControllerStreamRequestTerminateV1}, + {Kind: ControllerStreamRequestCompleteV1}, + {Kind: ControllerStreamRequestAcknowledgeTerminatedV1}, + } + for index, expected := range want { + request, err := reader.ReadRequest() + if err != nil || request != expected { + t.Fatalf("request %d = %#v, %v; want %#v", index, request, err, expected) + } + } + if _, err := reader.ReadRequest(); err != io.EOF { + t.Fatalf("terminal read error = %v, want EOF", err) + } +} + +func TestControllerStreamReaderV1RejectsNonCanonicalMessages(t *testing.T) { + tests := []struct { + name string + message string + want string + }{ + {name: "empty", message: "\n", want: "must not be empty"}, + {name: "missing newline", message: `{}`, want: "not newline terminated"}, + {name: "leading whitespace", message: ` {"schema":"reploy-controlled-session-client-v1","type":"terminate"}` + "\n", want: "exactly one JSON object"}, + {name: "duplicate", message: `{"schema":"reploy-controlled-session-client-v1","type":"terminate","type":"terminate"}` + "\n", want: "repeats field"}, + {name: "unknown field", message: `{"schema":"reploy-controlled-session-client-v1","type":"terminate","extra":true}` + "\n", want: "unknown field"}, + {name: "unknown schema", message: `{"schema":"other","type":"terminate"}` + "\n", want: "schema must be"}, + {name: "unknown type", message: `{"schema":"reploy-controlled-session-client-v1","type":"input"}` + "\n", want: "unsupported"}, + {name: "bad dimensions", message: `{"schema":"reploy-controlled-session-client-v1","type":"resize","columns":0,"rows":24}` + "\n", want: "between 1 and 65535"}, + {name: "payload on terminate", message: `{"schema":"reploy-controlled-session-client-v1","type":"terminate","columns":80}` + "\n", want: "unknown field"}, + {name: "invalid utf8", message: string([]byte{0xff, '\n'}), want: "valid UTF-8"}, + {name: "too large", message: strings.Repeat("x", MaxControllerStreamLineV1) + "\n", want: "exceeds"}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + reader, err := NewControllerStreamReaderV1(strings.NewReader(test.message)) + if err != nil { + t.Fatal(err) + } + if _, err := reader.ReadRequest(); err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("error = %v, want containing %q", err, test.want) + } + }) + } +} + +func TestControllerStreamWriterV1UsesExactJSONLines(t *testing.T) { + var output bytes.Buffer + writer, err := NewControllerStreamWriterV1(&output) + if err != nil { + t.Fatal(err) + } + if err := writer.WriteEvent(ControllerStreamEventV1{Kind: ControllerStreamEventReadyV1}); err != nil { + t.Fatal(err) + } + if got, want := output.String(), `{"schema":"reploy-controlled-session-client-v1","type":"ready"}`+"\n"; got != want { + t.Fatalf("ready line = %q, want %q", got, want) + } + output.Reset() + opened := testOpenedV1() + if err := writer.WriteEvent(ControllerStreamEventV1{Kind: ControllerStreamEventOpenedV1, Opened: &ControllerStreamOpenedV1{ + Operations: opened.Authorization.Operations, Endpoints: opened.Endpoints, Columns: opened.Columns, Rows: opened.Rows, + OutputFinalizationTimeoutMilliseconds: opened.OutputFinalizationTimeoutMilliseconds, + }}); err != nil { + t.Fatal(err) + } + line := output.String() + for _, fragment := range []string{`"type":"opened"`, `"operations":["complete","input","resize","terminate"]`, `"authorization"`} { + if fragment == `"authorization"` { + if strings.Contains(line, fragment) { + t.Fatalf("opened projection leaked authorization: %s", line) + } + continue + } + if !strings.Contains(line, fragment) { + t.Fatalf("opened projection missing %q: %s", fragment, line) + } + } +} diff --git a/internal/controlledsession/controller_terminal.go b/internal/controlledsession/controller_terminal.go new file mode 100644 index 00000000..653024b5 --- /dev/null +++ b/internal/controlledsession/controller_terminal.go @@ -0,0 +1,143 @@ +package controlledsession + +import ( + "context" + "crypto/rand" + "errors" + "fmt" + "io" + "net" + "path/filepath" + "sync" +) + +const ( + ControllerTemporaryHomeV1 = "/mnt/reploy-home" + ControllerTerminalSocketNameV1 = "terminal.sock" +) + +type controllerTerminalTransportV1 interface { + Accept(context.Context) (net.Conn, error) + Close() error +} + +type ControllerTerminalListenerV1 struct { + directory string + socket string + transport controllerTerminalTransportV1 + + mu sync.Mutex + acceptDone bool + connection *ControllerTerminalConnectionV1 + closeOnce sync.Once + closeErr error +} + +type ControllerTerminalConnectionV1 struct { + connection net.Conn + readMu sync.Mutex + writeMu sync.Mutex + closeOnce sync.Once + closeErr error +} + +func PrepareControllerTerminalListenerV1(temporaryHome string) (*ControllerTerminalListenerV1, error) { + if !filepath.IsAbs(temporaryHome) || filepath.Clean(temporaryHome) != temporaryHome { + return nil, fmt.Errorf("prepare controlled-session terminal listener requires an absolute clean temporary home") + } + var random [16]byte + if _, err := io.ReadFull(rand.Reader, random[:]); err != nil { + return nil, fmt.Errorf("prepare controlled-session terminal listener randomness: %w", err) + } + directory := filepath.Join(temporaryHome, fmt.Sprintf("reploy-controlled-session-%x", random)) + socket := filepath.Join(directory, ControllerTerminalSocketNameV1) + transport, err := prepareControllerTerminalTransportV1(temporaryHome, directory, socket) + if err != nil { + return nil, err + } + return &ControllerTerminalListenerV1{directory: directory, socket: socket, transport: transport}, nil +} + +func (listener *ControllerTerminalListenerV1) SocketPath() string { return listener.socket } + +func (listener *ControllerTerminalListenerV1) Accept(ctx context.Context) (*ControllerTerminalConnectionV1, error) { + if ctx == nil || ctx.Done() == nil { + return nil, fmt.Errorf("accept controlled-session terminal attachment: cancelable context is required") + } + listener.mu.Lock() + if listener.acceptDone { + listener.mu.Unlock() + return nil, fmt.Errorf("accept controlled-session terminal attachment: listener has already accepted or failed") + } + listener.acceptDone = true + listener.mu.Unlock() + connection, err := listener.transport.Accept(ctx) + if err != nil { + return nil, errors.Join(fmt.Errorf("accept controlled-session terminal attachment: %w", err), listener.Close()) + } + terminal := &ControllerTerminalConnectionV1{connection: connection} + listener.mu.Lock() + listener.connection = terminal + listener.mu.Unlock() + return terminal, nil +} + +func (listener *ControllerTerminalListenerV1) Close() error { + listener.closeOnce.Do(func() { + listener.mu.Lock() + connection := listener.connection + listener.mu.Unlock() + var connectionErr error + if connection != nil { + connectionErr = connection.Close() + } + listener.closeErr = errors.Join(connectionErr, listener.transport.Close()) + }) + return listener.closeErr +} + +func (connection *ControllerTerminalConnectionV1) ReadRequest(ctx context.Context) (RequestV1, error) { + if ctx == nil || ctx.Done() == nil { + return RequestV1{}, fmt.Errorf("read controlled-session terminal request: cancelable context is required") + } + connection.readMu.Lock() + defer connection.readMu.Unlock() + var request RequestV1 + err := withConnectionDeadlineV1(ctx, connection.connection.SetReadDeadline, func() error { + var readErr error + request, readErr = ReadTerminalRequestV1(connection.connection) + return readErr + }) + if err != nil { + return RequestV1{}, fmt.Errorf("read controlled-session terminal request: %w", err) + } + return request, nil +} + +func (connection *ControllerTerminalConnectionV1) WriteOutput(ctx context.Context, content []byte) error { + if content == nil { + return fmt.Errorf("write controlled-session terminal output: byte sequence is required") + } + return connection.write(ctx, func() error { return WriteTerminalOutputV1(connection.connection, content) }) +} + +func (connection *ControllerTerminalConnectionV1) WriteEnd(ctx context.Context, status WorkloadOutputFinalizationStatusV1) error { + return connection.write(ctx, func() error { return WriteTerminalEndV1(connection.connection, status) }) +} + +func (connection *ControllerTerminalConnectionV1) write(ctx context.Context, write func() error) error { + if ctx == nil || ctx.Done() == nil { + return fmt.Errorf("write controlled-session terminal event: cancelable context is required") + } + connection.writeMu.Lock() + defer connection.writeMu.Unlock() + if err := withConnectionDeadlineV1(ctx, connection.connection.SetWriteDeadline, write); err != nil { + return errors.Join(fmt.Errorf("write controlled-session terminal event: %w", err), connection.Close()) + } + return nil +} + +func (connection *ControllerTerminalConnectionV1) Close() error { + connection.closeOnce.Do(func() { connection.closeErr = connection.connection.Close() }) + return connection.closeErr +} diff --git a/internal/controlledsession/controller_terminal_linux.go b/internal/controlledsession/controller_terminal_linux.go new file mode 100644 index 00000000..930ff744 --- /dev/null +++ b/internal/controlledsession/controller_terminal_linux.go @@ -0,0 +1,133 @@ +//go:build linux + +package controlledsession + +import ( + "context" + "errors" + "fmt" + "net" + "os" + "path/filepath" + "sync" + "syscall" + "time" + + "golang.org/x/sys/unix" +) + +type controllerUnixTerminalTransportV1 struct { + directory string + socket string + listener *net.UnixListener + stopOnce sync.Once + stopErr error + closeOnce sync.Once + closeErr error +} + +func prepareControllerTerminalTransportV1(temporaryHome string, directory string, socket string) (controllerTerminalTransportV1, error) { + info, err := os.Lstat(temporaryHome) + if err != nil { + return nil, fmt.Errorf("inspect controlled-session temporary home: %w", err) + } + if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() { + return nil, fmt.Errorf("controlled-session temporary home must be a real directory, not a symlink") + } + identity, ok := info.Sys().(*syscall.Stat_t) + if !ok { + return nil, fmt.Errorf("controlled-session temporary home ownership is unavailable") + } + if identity.Uid != uint32(os.Geteuid()) || identity.Gid != uint32(os.Getegid()) { + return nil, fmt.Errorf("controlled-session temporary home is owned by %d:%d, expected %d:%d", identity.Uid, identity.Gid, os.Geteuid(), os.Getegid()) + } + if info.Mode().Perm() != 0o700 { + return nil, fmt.Errorf("controlled-session temporary home mode is %04o, expected 0700", info.Mode().Perm()) + } + if filepath.Dir(directory) != temporaryHome || filepath.Base(socket) != ControllerTerminalSocketNameV1 || filepath.Dir(socket) != directory { + return nil, fmt.Errorf("controlled-session terminal path escaped the private temporary home") + } + if len(socket) > len(unix.RawSockaddrUnix{}.Path)-1 { + return nil, fmt.Errorf("controlled-session terminal socket path exceeds the Linux AF_UNIX maximum") + } + if err := os.Mkdir(directory, 0o700); err != nil { + return nil, fmt.Errorf("create controlled-session terminal directory: %w", err) + } + if err := os.Chmod(directory, 0o700); err != nil { + return nil, fmt.Errorf("set controlled-session terminal directory mode: %w", err) + } + var listener *net.UnixListener + cleanup := true + defer func() { + if cleanup { + if listener != nil { + _ = listener.Close() + } + _ = os.Remove(socket) + _ = os.Remove(directory) + } + }() + listener, err = net.ListenUnix("unix", &net.UnixAddr{Name: socket, Net: "unix"}) + if err != nil { + return nil, fmt.Errorf("listen on controlled-session terminal socket: %w", err) + } + listener.SetUnlinkOnClose(true) + if err := os.Chmod(socket, 0o600); err != nil { + return nil, fmt.Errorf("set controlled-session terminal socket mode: %w", err) + } + cleanup = false + return &controllerUnixTerminalTransportV1{directory: directory, socket: socket, listener: listener}, nil +} + +func (transport *controllerUnixTerminalTransportV1) Accept(ctx context.Context) (net.Conn, error) { + deadline := time.Time{} + if value, ok := ctx.Deadline(); ok { + deadline = value + } + if err := transport.listener.SetDeadline(deadline); err != nil { + return nil, err + } + stop := context.AfterFunc(ctx, func() { _ = transport.listener.SetDeadline(time.Now()) }) + connection, err := transport.listener.AcceptUnix() + stop() + if ctxErr := ctx.Err(); ctxErr != nil { + if connection != nil { + _ = connection.Close() + } + return nil, ctxErr + } + if err != nil { + if !deadline.IsZero() && !time.Now().Before(deadline) { + var networkError net.Error + if errors.As(err, &networkError) && networkError.Timeout() { + return nil, context.DeadlineExceeded + } + } + return nil, err + } + uid, gid, err := unixPeerIdentityV1(connection) + if err != nil { + return nil, errors.Join(err, connection.Close()) + } + if uid != uint32(os.Geteuid()) || gid != uint32(os.Getegid()) { + return nil, errors.Join(fmt.Errorf("terminal attachment identity is %d:%d, expected %d:%d", uid, gid, os.Geteuid(), os.Getegid()), connection.Close()) + } + if err := transport.stopAccepting(); err != nil { + return nil, errors.Join(err, connection.Close()) + } + return connection, nil +} + +func (transport *controllerUnixTerminalTransportV1) stopAccepting() error { + transport.stopOnce.Do(func() { + transport.stopErr = errors.Join(transport.listener.Close(), removeIfExistsV1(transport.socket)) + }) + return transport.stopErr +} + +func (transport *controllerUnixTerminalTransportV1) Close() error { + transport.closeOnce.Do(func() { + transport.closeErr = errors.Join(transport.stopAccepting(), removeIfExistsV1(transport.socket), os.Remove(transport.directory)) + }) + return transport.closeErr +} diff --git a/internal/controlledsession/controller_terminal_linux_test.go b/internal/controlledsession/controller_terminal_linux_test.go new file mode 100644 index 00000000..040e588d --- /dev/null +++ b/internal/controlledsession/controller_terminal_linux_test.go @@ -0,0 +1,108 @@ +//go:build linux + +package controlledsession + +import ( + "context" + "errors" + "net" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestControllerTerminalListenerV1IsPrivateSingleUseAndCleansUp(t *testing.T) { + home := shortControllerBrokerTempHomeV1(t) + listener, err := PrepareControllerTerminalListenerV1(home) + if err != nil { + t.Fatal(err) + } + directory := filepath.Dir(listener.SocketPath()) + if filepath.Dir(directory) != home || filepath.Base(directory)[:len("reploy-controlled-session-")] != "reploy-controlled-session-" { + t.Fatalf("terminal directory escaped expected grammar: %q", directory) + } + if info, err := os.Stat(directory); err != nil || info.Mode().Perm() != 0o700 { + t.Fatalf("terminal directory mode = %v, %v; want 0700", info, err) + } + if info, err := os.Stat(listener.SocketPath()); err != nil || info.Mode().Perm() != 0o600 { + t.Fatalf("terminal socket mode = %v, %v; want 0600", info, err) + } + + accepted := make(chan *ControllerTerminalConnectionV1, 1) + failures := make(chan error, 1) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + go func() { + connection, err := listener.Accept(ctx) + if err != nil { + failures <- err + return + } + accepted <- connection + }() + peer, err := net.DialUnix("unix", nil, &net.UnixAddr{Name: listener.SocketPath(), Net: "unix"}) + if err != nil { + t.Fatal(err) + } + defer peer.Close() + select { + case connection := <-accepted: + if connection == nil { + t.Fatal("accepted nil terminal connection") + } + case err := <-failures: + t.Fatal(err) + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + if _, err := os.Lstat(listener.SocketPath()); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("claimed terminal socket still exists: %v", err) + } + if _, err := listener.Accept(ctx); err == nil { + t.Fatal("terminal listener accepted a second claim") + } + if err := listener.Close(); err != nil { + t.Fatal(err) + } + if _, err := os.Lstat(directory); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("terminal directory survived close: %v", err) + } +} + +func TestControllerTerminalListenerV1RejectsSymlinkHome(t *testing.T) { + root := t.TempDir() + realHome := filepath.Join(root, "real") + if err := os.Mkdir(realHome, 0o700); err != nil { + t.Fatal(err) + } + link := filepath.Join(root, "link") + if err := os.Symlink(realHome, link); err != nil { + t.Fatal(err) + } + if _, err := PrepareControllerTerminalListenerV1(link); err == nil { + t.Fatal("symlink temporary home was accepted") + } +} + +func TestControllerTerminalListenerV1RejectsNonPrivateHome(t *testing.T) { + home := shortControllerBrokerTempHomeV1(t) + if err := os.Chmod(home, 0o750); err != nil { + t.Fatal(err) + } + if _, err := PrepareControllerTerminalListenerV1(home); err == nil || !strings.Contains(err.Error(), "expected 0700") { + t.Fatalf("non-private home error = %v", err) + } +} + +func TestControllerTerminalConnectionV1AppliesBackpressure(t *testing.T) { + broker, peer := net.Pipe() + defer peer.Close() + connection := &ControllerTerminalConnectionV1{connection: broker} + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + if err := connection.WriteOutput(ctx, make([]byte, MaxFramePayloadV1)); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("blocked output error = %v, want deadline exceeded", err) + } +} diff --git a/internal/controlledsession/controller_terminal_unsupported.go b/internal/controlledsession/controller_terminal_unsupported.go new file mode 100644 index 00000000..40ce1bc6 --- /dev/null +++ b/internal/controlledsession/controller_terminal_unsupported.go @@ -0,0 +1,9 @@ +//go:build !linux + +package controlledsession + +import "fmt" + +func prepareControllerTerminalTransportV1(string, string, string) (controllerTerminalTransportV1, error) { + return nil, fmt.Errorf("controlled-session controller broker requires Linux") +} diff --git a/internal/controlledsession/controller_terminal_unsupported_test.go b/internal/controlledsession/controller_terminal_unsupported_test.go new file mode 100644 index 00000000..7c9537e5 --- /dev/null +++ b/internal/controlledsession/controller_terminal_unsupported_test.go @@ -0,0 +1,14 @@ +//go:build !linux + +package controlledsession + +import ( + "strings" + "testing" +) + +func TestControllerTerminalListenerV1FailsGracefullyOffLinux(t *testing.T) { + if _, err := PrepareControllerTerminalListenerV1(t.TempDir()); err == nil || !strings.Contains(err.Error(), "requires Linux") { + t.Fatalf("non-Linux terminal listener error = %v", err) + } +} diff --git a/internal/controlledsession/lifecycle.go b/internal/controlledsession/lifecycle.go index 80667b85..80e4c293 100644 --- a/internal/controlledsession/lifecycle.go +++ b/internal/controlledsession/lifecycle.go @@ -347,28 +347,41 @@ func (machine *MachineV1) validateControllerFinishLocked(status ControllerFinali } func (machine *MachineV1) ApplyRequest(request RequestV1) (TransitionV1, error) { + transition, _, err := machine.applyRequestV1(request, false) + return transition, err +} + +// ApplyRequestOrDiscardTerminatingPTYV1 applies a request under the lifecycle +// lock, but treats authorized input and resize requests that lose a race with +// authoritative termination as discarded. Structured controller requests and +// unauthorized terminal requests retain the strict ApplyRequest behavior. +func (machine *MachineV1) ApplyRequestOrDiscardTerminatingPTYV1(request RequestV1) (TransitionV1, bool, error) { + return machine.applyRequestV1(request, true) +} + +func (machine *MachineV1) applyRequestV1(request RequestV1, discardTerminatingPTY bool) (TransitionV1, bool, error) { machine.mu.Lock() defer machine.mu.Unlock() transition := TransitionV1{Before: machine.state, After: machine.state, Cause: machine.cause} if err := ValidateRequestV1(request); err != nil { - return transition, fmt.Errorf("%w: %v", ErrRequestRejected, err) + return transition, false, fmt.Errorf("%w: %v", ErrRequestRejected, err) } // A terminal acknowledgement is protocol flow control, not a granted // controller capability. Its lifecycle position is the authorization. if request.Kind == RequestAcknowledgeTerminatedV1 { if machine.state != StateTerminatedV1 || machine.result == nil || !machine.resultDelivered { - return transition, fmt.Errorf("%w: terminal result has not been delivered", ErrRequestRejected) + return transition, false, fmt.Errorf("%w: terminal result has not been delivered", ErrRequestRejected) } machine.resultAcknowledged = true transition.RequestAccepted = true transition.AwaitingResultAcknowledgement = false transition.ResultAcknowledged = true transition.Result = cloneResultV1(machine.result) - return transition, nil + return transition, false, nil } operation := operationForRequestV1(request.Kind) if !containsOperationV1(machine.authorization.Operations, operation) { - return transition, fmt.Errorf("%w: operation %q was not granted", ErrRequestRejected, operation) + return transition, false, fmt.Errorf("%w: operation %q was not granted", ErrRequestRejected, operation) } switch machine.state { @@ -378,26 +391,31 @@ func (machine *MachineV1) ApplyRequest(request RequestV1) (TransitionV1, error) case RequestTerminateV1: machine.latchLocked(CauseControllerTerminateV1, &transition) case RequestCompleteV1: - return transition, fmt.Errorf("%w: completion is valid only after workload outputs are finalized", ErrRequestRejected) + return transition, false, fmt.Errorf("%w: completion is valid only after workload outputs are finalized", ErrRequestRejected) } case StateTerminatingV1: switch request.Kind { + case RequestInputV1, RequestResizeV1: + if discardTerminatingPTY { + return transition, true, nil + } + return transition, false, fmt.Errorf("%w: %s is not accepted while terminating", ErrRequestRejected, request.Kind) case RequestTerminateV1: // Repeated graceful termination is idempotent and does not alter // output or controller finalization already in progress. case RequestCompleteV1: if machine.waitingOutputs || !machine.waitingFinalize || machine.controller.Kind != ControllerFinalizationActiveV1 { - return transition, fmt.Errorf("%w: completion requires finalized workload outputs and a pending controller finalization", ErrRequestRejected) + return transition, false, fmt.Errorf("%w: completion requires finalized workload outputs and a pending controller finalization", ErrRequestRejected) } machine.waitingFinalize = false machine.controller = ControllerFinalizationStatusV1{Kind: ControllerFinalizationCompletedV1} default: - return transition, fmt.Errorf("%w: %s is not accepted while terminating", ErrRequestRejected, request.Kind) + return transition, false, fmt.Errorf("%w: %s is not accepted while terminating", ErrRequestRejected, request.Kind) } case StatePreparingV1, StateTerminatedV1: - return transition, fmt.Errorf("%w: requests are not accepted while %s", ErrRequestRejected, machine.state) + return transition, false, fmt.Errorf("%w: requests are not accepted while %s", ErrRequestRejected, machine.state) default: - return transition, fmt.Errorf("%w: lifecycle state %q is invalid", ErrRequestRejected, machine.state) + return transition, false, fmt.Errorf("%w: lifecycle state %q is invalid", ErrRequestRejected, machine.state) } transition.After = machine.state @@ -409,7 +427,7 @@ func (machine *MachineV1) ApplyRequest(request RequestV1) (TransitionV1, error) transition.AwaitingResultAcknowledgement = machine.resultDelivered && !machine.resultAcknowledged transition.RequestAccepted = true transition.ResultAcknowledged = machine.resultAcknowledged - return transition, nil + return transition, false, nil } func (machine *MachineV1) latchLocked(cause TerminationCauseV1, transition *TransitionV1) { diff --git a/internal/controlledsession/lifecycle_test.go b/internal/controlledsession/lifecycle_test.go index a8f5b0c5..8cd8231f 100644 --- a/internal/controlledsession/lifecycle_test.go +++ b/internal/controlledsession/lifecycle_test.go @@ -117,6 +117,26 @@ func TestLifecycleOutputBarrierPrecedesControllerFinalization(t *testing.T) { } } +func TestLifecycleAtomicallyDiscardsAuthorizedPTYDuringTermination(t *testing.T) { + machine := activatedMachineV1(t) + if _, err := machine.Observe(ObservationV1{Kind: ObservationHostCancelV1, Reason: "test termination"}); err != nil { + t.Fatal(err) + } + + for _, request := range []RequestV1{ + {Kind: RequestInputV1, Bytes: []byte("late input")}, + {Kind: RequestResizeV1, Columns: 120, Rows: 40}, + } { + transition, discarded, err := machine.ApplyRequestOrDiscardTerminatingPTYV1(request) + if err != nil || !discarded { + t.Fatalf("terminating %s request = %#v, discarded %t, %v", request.Kind, transition, discarded, err) + } + if transition.Before != StateTerminatingV1 || transition.After != StateTerminatingV1 || transition.RequestAccepted { + t.Fatalf("discarded %s transition = %#v", request.Kind, transition) + } + } +} + func TestLifecycleRejectsCompletionUntilOutputFinalizationIsPublished(t *testing.T) { machine := activatedMachineV1(t) code := 0 diff --git a/internal/controlledsession/terminal_protocol.go b/internal/controlledsession/terminal_protocol.go new file mode 100644 index 00000000..7f2f5421 --- /dev/null +++ b/internal/controlledsession/terminal_protocol.go @@ -0,0 +1,152 @@ +package controlledsession + +import ( + "bytes" + "encoding/binary" + "encoding/json" + "fmt" + "io" +) + +const terminalFrameHeaderSizeV1 = 10 + +var terminalFrameMagicV1 = [4]byte{'R', 'P', 'T', 'M'} + +type terminalWireKindV1 byte + +const ( + terminalWireInputV1 terminalWireKindV1 = 0x01 + terminalWireResizeV1 terminalWireKindV1 = 0x02 + terminalWireOutputV1 terminalWireKindV1 = 0x81 + terminalWireEndV1 terminalWireKindV1 = 0x82 +) + +type TerminalEventKindV1 string + +const ( + TerminalEventOutputV1 TerminalEventKindV1 = "output" + TerminalEventEndV1 TerminalEventKindV1 = "terminal-end" +) + +type TerminalEventV1 struct { + Kind TerminalEventKindV1 + Bytes []byte + Status *WorkloadOutputFinalizationStatusV1 +} + +func WriteTerminalRequestV1(writer io.Writer, request RequestV1) error { + if err := ValidateRequestV1(request); err != nil { + return err + } + switch request.Kind { + case RequestInputV1: + return writeTerminalFrameV1(writer, terminalWireInputV1, request.Bytes) + case RequestResizeV1: + payload := make([]byte, 8) + binary.BigEndian.PutUint32(payload[:4], request.Columns) + binary.BigEndian.PutUint32(payload[4:], request.Rows) + return writeTerminalFrameV1(writer, terminalWireResizeV1, payload) + default: + return fmt.Errorf("controlled-session terminal protocol does not carry %q requests", request.Kind) + } +} + +func ReadTerminalRequestV1(reader io.Reader) (RequestV1, error) { + kind, payload, err := readTerminalFrameV1(reader) + if err != nil { + return RequestV1{}, err + } + switch kind { + case terminalWireInputV1: + request := RequestV1{Kind: RequestInputV1, Bytes: payload} + return request, ValidateRequestV1(request) + case terminalWireResizeV1: + if len(payload) != 8 { + return RequestV1{}, fmt.Errorf("controlled-session terminal resize payload must contain 8 bytes") + } + request := RequestV1{Kind: RequestResizeV1, Columns: binary.BigEndian.Uint32(payload[:4]), Rows: binary.BigEndian.Uint32(payload[4:])} + return request, ValidateRequestV1(request) + default: + return RequestV1{}, fmt.Errorf("controlled-session terminal frame kind 0x%02x is not a request", byte(kind)) + } +} + +func WriteTerminalOutputV1(writer io.Writer, content []byte) error { + if content == nil { + return fmt.Errorf("controlled-session terminal output requires a byte sequence") + } + return writeTerminalFrameV1(writer, terminalWireOutputV1, content) +} + +func WriteTerminalEndV1(writer io.Writer, status WorkloadOutputFinalizationStatusV1) error { + if err := validateWorkloadOutputFinalizationStatusV1(status); err != nil { + return err + } + payload, err := json.Marshal(status) + if err != nil { + return fmt.Errorf("encode controlled-session terminal end: %w", err) + } + return writeTerminalFrameV1(writer, terminalWireEndV1, payload) +} + +func ReadTerminalEventV1(reader io.Reader) (TerminalEventV1, error) { + kind, payload, err := readTerminalFrameV1(reader) + if err != nil { + return TerminalEventV1{}, err + } + switch kind { + case terminalWireOutputV1: + return TerminalEventV1{Kind: TerminalEventOutputV1, Bytes: payload}, nil + case terminalWireEndV1: + var status WorkloadOutputFinalizationStatusV1 + if err := decodeStrictJSONV1("terminal-end event", payload, &status); err != nil { + return TerminalEventV1{}, err + } + if err := validateWorkloadOutputFinalizationStatusV1(status); err != nil { + return TerminalEventV1{}, err + } + return TerminalEventV1{Kind: TerminalEventEndV1, Status: &status}, nil + default: + return TerminalEventV1{}, fmt.Errorf("controlled-session terminal frame kind 0x%02x is not an event", byte(kind)) + } +} + +func writeTerminalFrameV1(writer io.Writer, kind terminalWireKindV1, payload []byte) error { + if len(payload) > MaxFramePayloadV1 { + return fmt.Errorf("controlled-session terminal frame payload exceeds %d bytes", MaxFramePayloadV1) + } + header := make([]byte, terminalFrameHeaderSizeV1) + copy(header[:4], terminalFrameMagicV1[:]) + header[4] = ProtocolVersionV1 + header[5] = byte(kind) + binary.BigEndian.PutUint32(header[6:], uint32(len(payload))) + if err := writeAllV1(writer, header); err != nil { + return fmt.Errorf("write controlled-session terminal frame header: %w", err) + } + if err := writeAllV1(writer, payload); err != nil { + return fmt.Errorf("write controlled-session terminal frame payload: %w", err) + } + return nil +} + +func readTerminalFrameV1(reader io.Reader) (terminalWireKindV1, []byte, error) { + header := make([]byte, terminalFrameHeaderSizeV1) + if _, err := io.ReadFull(reader, header); err != nil { + return 0, nil, fmt.Errorf("read controlled-session terminal frame header: %w", err) + } + if !bytes.Equal(header[:4], terminalFrameMagicV1[:]) { + return 0, nil, fmt.Errorf("controlled-session terminal frame magic is invalid") + } + if header[4] != ProtocolVersionV1 { + return 0, nil, fmt.Errorf("controlled-session terminal protocol version %d is unsupported", header[4]) + } + length := binary.BigEndian.Uint32(header[6:]) + if length > MaxFramePayloadV1 { + return 0, nil, fmt.Errorf("controlled-session terminal frame payload length %d exceeds %d bytes", length, MaxFramePayloadV1) + } + payload := make([]byte, int(length)) + if _, err := io.ReadFull(reader, payload); err != nil { + return 0, nil, fmt.Errorf("read controlled-session terminal frame payload: %w", err) + } + return terminalWireKindV1(header[5]), payload, nil +} diff --git a/internal/controlledsession/terminal_protocol_test.go b/internal/controlledsession/terminal_protocol_test.go new file mode 100644 index 00000000..e31fc4bb --- /dev/null +++ b/internal/controlledsession/terminal_protocol_test.go @@ -0,0 +1,65 @@ +package controlledsession + +import ( + "bytes" + "encoding/binary" + "strings" + "testing" +) + +func TestTerminalProtocolV1RoundTripsRequestsAndEvents(t *testing.T) { + requests := []RequestV1{ + {Kind: RequestInputV1, Bytes: []byte{0, 3, 255}}, + {Kind: RequestResizeV1, Columns: 132, Rows: 43}, + } + for _, want := range requests { + var wire bytes.Buffer + if err := WriteTerminalRequestV1(&wire, want); err != nil { + t.Fatal(err) + } + got, err := ReadTerminalRequestV1(&wire) + if err != nil || got.Kind != want.Kind || !bytes.Equal(got.Bytes, want.Bytes) || got.Columns != want.Columns || got.Rows != want.Rows { + t.Fatalf("terminal request = %#v, %v; want %#v", got, err, want) + } + } + + var wire bytes.Buffer + if err := WriteTerminalOutputV1(&wire, []byte("hello\x00world")); err != nil { + t.Fatal(err) + } + if err := WriteTerminalEndV1(&wire, WorkloadOutputFinalizationStatusV1{Kind: WorkloadOutputFinalizationDrainedV1}); err != nil { + t.Fatal(err) + } + output, err := ReadTerminalEventV1(&wire) + if err != nil || output.Kind != TerminalEventOutputV1 || string(output.Bytes) != "hello\x00world" { + t.Fatalf("terminal output = %#v, %v", output, err) + } + end, err := ReadTerminalEventV1(&wire) + if err != nil || end.Kind != TerminalEventEndV1 || end.Status == nil || end.Status.Kind != WorkloadOutputFinalizationDrainedV1 { + t.Fatalf("terminal end = %#v, %v", end, err) + } +} + +func TestTerminalProtocolV1RejectsMalformedOrWrongDirectionFrames(t *testing.T) { + var wire bytes.Buffer + header := make([]byte, terminalFrameHeaderSizeV1) + copy(header, terminalFrameMagicV1[:]) + header[4] = ProtocolVersionV1 + header[5] = byte(terminalWireResizeV1) + binary.BigEndian.PutUint32(header[6:], 1) + wire.Write(header) + wire.WriteByte(0) + if _, err := ReadTerminalRequestV1(&wire); err == nil || !strings.Contains(err.Error(), "8 bytes") { + t.Fatalf("malformed resize error = %v", err) + } + wire.Reset() + if err := WriteTerminalOutputV1(&wire, []byte("output")); err != nil { + t.Fatal(err) + } + if _, err := ReadTerminalRequestV1(&wire); err == nil || !strings.Contains(err.Error(), "not a request") { + t.Fatalf("wrong-direction error = %v", err) + } + if err := WriteTerminalRequestV1(&wire, RequestV1{Kind: RequestTerminateV1}); err == nil || !strings.Contains(err.Error(), "does not carry") { + t.Fatalf("terminal terminate error = %v", err) + } +} diff --git a/internal/dockerdeploy/controlled_session_supervisor.go b/internal/dockerdeploy/controlled_session_supervisor.go index c9ee57fc..130e651c 100644 --- a/internal/dockerdeploy/controlled_session_supervisor.go +++ b/internal/dockerdeploy/controlled_session_supervisor.go @@ -910,10 +910,15 @@ func (supervisor *controlledSessionSupervisorV1) handleRequest(ctx context.Conte default: } } - transition, err := supervisor.machine.ApplyRequest(request) + transition, discarded, err := supervisor.machine.ApplyRequestOrDiscardTerminatingPTYV1(request) if err != nil { return err } + if discarded { + // Termination latches before its lifecycle event can reach the controller. + // Discard PTY traffic already in flight across that publication gap. + return nil + } supervisor.recordTransition(transition) if _, err := controlledsession.ApplyAcceptedWorkloadPTYRequestV1(ctx, supervisor.workload, request); err != nil { return err diff --git a/internal/dockerdeploy/controlled_session_supervisor_test.go b/internal/dockerdeploy/controlled_session_supervisor_test.go index 0c6ea8d6..3915163a 100644 --- a/internal/dockerdeploy/controlled_session_supervisor_test.go +++ b/internal/dockerdeploy/controlled_session_supervisor_test.go @@ -145,6 +145,40 @@ func TestControlledSessionSupervisorRejectsRequestBeforeReady(t *testing.T) { } } +func TestControlledSessionSupervisorIgnoresPTYRequestsRacingTerminatingEvent(t *testing.T) { + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + machine, err := controlledsession.NewMachineV1(plan.Authorization) + if err != nil { + t.Fatal(err) + } + if _, err := machine.Observe(controlledsession.ObservationV1{Kind: controlledsession.ObservationActivatedV1}); err != nil { + t.Fatal(err) + } + if _, err := machine.Observe(controlledsession.ObservationV1{ + Kind: controlledsession.ObservationHostCancelV1, Reason: "test termination", + }); err != nil { + t.Fatal(err) + } + workload := newFakeControlledSessionWorkloadV1(nil, 0) + supervisor := &controlledSessionSupervisorV1{ + machine: machine, workload: workload, controllerRequestsReady: true, + } + for _, request := range []controlledsession.RequestV1{ + {Kind: controlledsession.RequestInputV1, Bytes: []byte("late input")}, + {Kind: controlledsession.RequestResizeV1, Columns: 120, Rows: 40}, + } { + if err := supervisor.handleRequest(t.Context(), request); err != nil { + t.Fatalf("late %s request error = %v", request.Kind, err) + } + } + if input := workload.snapshotInput(); len(input) != 0 || workload.columns != 0 || workload.rows != 0 { + t.Fatalf("late PTY effects = input %q, dimensions %dx%d", input, workload.columns, workload.rows) + } + if snapshot := machine.Snapshot(); snapshot.State != controlledsession.StateTerminatingV1 || snapshot.Cause != controlledsession.CauseHostCancelV1 { + t.Fatalf("lifecycle changed by late PTY request: %#v", snapshot) + } +} + func TestRunControlledSessionV1PreparesAttachesAndCleansLeaseNetwork(t *testing.T) { plan := controlledSessionNetworkPlanFixtureV1(t) requests := make(chan controlledsession.RequestV1, 8) @@ -313,11 +347,12 @@ func TestBindControlledSessionDockerEndpointV1SelectsOnceForBothContainers(t *te func TestRunControlledSessionV1FailsClosedAfterWatchdogExit(t *testing.T) { plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) requests := make(chan controlledsession.RequestV1, 8) - requests <- controlledsession.RequestV1{Kind: controlledsession.RequestInputV1, Bytes: []byte("ready")} controller := newFakeControlledSessionProcessV1() transport := &fakeControlledSessionTransportV1{requests: requests} transport.onEvent = func(event controlledsession.EventV1) { switch event.Kind { + case controlledsession.EventReadyV1: + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestInputV1, Bytes: []byte("ready")} case controlledsession.EventWorkloadOutputsFinalizedV1: requests <- controlledsession.RequestV1{Kind: controlledsession.RequestCompleteV1} case controlledsession.EventTerminatedV1: