diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml index 53af737b..b63c9fae 100644 --- a/.github/workflows/docker.yml +++ b/.github/workflows/docker.yml @@ -59,14 +59,15 @@ jobs: platforms: linux/amd64,linux/arm64 provenance: false context: . - push: ${{ github.event.pull_request.head.repo.fork != true }} + # Our repository is itself a fork; publish trusted same-repository PRs. + push: ${{ github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository }} tags: ${{ steps.meta.outputs.tags }} labels: ${{ steps.meta.outputs.labels }} cache-from: type=gha cache-to: type=gha,mode=max - name: Generate artifact attestation - if: ${{ github.event.pull_request.head.repo.fork != true }} + if: ${{ github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository }} uses: actions/attest@508db95dd578ae2727ebd6217d5ba78e4fbda05d # v4.2.1 with: subject-name: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} @@ -74,7 +75,7 @@ jobs: push-to-registry: true - name: Job summary - if: ${{ github.event.pull_request.head.repo.fork != true }} + if: ${{ github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository }} env: DIGEST: ${{ steps.push.outputs.digest }} REF_NAME: ${{ github.ref_name }} diff --git a/internal/kubernetes/kubernetesClientImpl.go b/internal/kubernetes/kubernetesClientImpl.go index 1bb12224..661465bd 100644 --- a/internal/kubernetes/kubernetesClientImpl.go +++ b/internal/kubernetes/kubernetesClientImpl.go @@ -151,7 +151,17 @@ func (k *kubernetesClientImpl) attemptPodCreate( wg: &sync.WaitGroup{}, removeLock: &sync.Mutex{}, } - return createdPod.wait(ctx) + readyPod, waitErr := createdPod.wait(ctx) + if waitErr != nil && k.config.Pod.Mode == containerSSHConfig.KubernetesExecutionModeConnection { + // Startup may have exhausted ctx. Roll back this creation before the + // retry loop loses its identity, with a separate, bounded stop budget. + // remove logs failures; preserve the original readiness error. + cleanupCtx, cancel := context.WithTimeout(context.Background(), k.config.Timeouts.PodStop) + _ = createdPod.remove(cleanupCtx) + cancel() + return nil, waitErr + } + return readyPod, waitErr } k.backendFailuresMetric.Increment() logger.Debug( diff --git a/internal/kubernetes/kubernetesExecutionImpl.go b/internal/kubernetes/kubernetesExecutionImpl.go index 35b958d9..411005e1 100644 --- a/internal/kubernetes/kubernetesExecutionImpl.go +++ b/internal/kubernetes/kubernetesExecutionImpl.go @@ -214,7 +214,7 @@ type stdoutProxyWriter struct { pidRead bool lock *sync.Mutex buf *bytes.Buffer - tty bool + waitEcho bool } func (s *stdoutProxyWriter) Write(p []byte) (n int, err error) { @@ -224,11 +224,13 @@ func (s *stdoutProxyWriter) Write(p []byte) (n int, err error) { return s.backend.Write(p) } // Read the pid. See https://github.com/containerssh/agent for details. - // We need 6 bytes because the first 2 bytes will be \r\n from the --wait function injected in the reader above. + // Session-mode PTYs also need the two echoed --wait bytes. Connection + // and persistent execs do not use --wait: their four-byte PID must unblock + // initial resizing before a terminal-dependent application emits output. if n, err := s.buf.Write(p); err != nil { return n, err } - if (s.tty && s.buf.Len() >= 6) || (!s.tty && s.buf.Len() >= 4) { + if (s.waitEcho && s.buf.Len() >= 6) || (!s.waitEcho && s.buf.Len() >= 4) { bufferBytes := s.buf.Bytes() s.pidRead = true s.buf = nil @@ -264,7 +266,7 @@ func (k *kubernetesExecutionImpl) run( } } stdout = &stdoutProxyWriter{ - tty: k.tty, + waitEcho: k.tty && k.pod.config.Pod.Mode == config.KubernetesExecutionModeSession, backend: stdout, lock: &sync.Mutex{}, pidChannel: pidChannel, diff --git a/internal/kubernetes/pushSizeQueueImpl.go b/internal/kubernetes/pushSizeQueueImpl.go index e942ecec..8b73dcfa 100644 --- a/internal/kubernetes/pushSizeQueueImpl.go +++ b/internal/kubernetes/pushSizeQueueImpl.go @@ -2,31 +2,70 @@ package kubernetes import ( "context" + "errors" + "sync" "k8s.io/client-go/tools/remotecommand" ) type pushSizeQueueImpl struct { resizeChan chan remotecommand.TerminalSize + lock sync.Mutex + stopped bool + stopChan chan struct{} } func (s *pushSizeQueueImpl) Push(ctx context.Context, size remotecommand.TerminalSize) error { + s.lock.Lock() + stopChan := s.getStopChanLocked() + stopped := s.stopped + s.lock.Unlock() + if stopped { + return errors.New("terminal size queue is stopped") + } + select { case <-ctx.Done(): return ctx.Err() + case <-stopChan: + return errors.New("terminal size queue is stopped") case s.resizeChan <- size: return nil } } func (s *pushSizeQueueImpl) Next() *remotecommand.TerminalSize { - size, ok := <-s.resizeChan - if !ok { + s.lock.Lock() + stopChan := s.getStopChanLocked() + stopped := s.stopped + s.lock.Unlock() + if stopped { + return nil + } + + select { + case <-stopChan: return nil + case size := <-s.resizeChan: + return &size } - return &size } func (s *pushSizeQueueImpl) Stop() { - close(s.resizeChan) + s.lock.Lock() + defer s.lock.Unlock() + if s.stopped { + return + } + s.stopped = true + // Window requests may still be sending when execution finishes. Signal + // shutdown without closing their data channel or blocking behind a send. + close(s.getStopChanLocked()) +} + +func (s *pushSizeQueueImpl) getStopChanLocked() chan struct{} { + if s.stopChan == nil { + s.stopChan = make(chan struct{}) + } + return s.stopChan } diff --git a/internal/kubernetes/resize_shutdown_test.go b/internal/kubernetes/resize_shutdown_test.go new file mode 100644 index 00000000..678e4dcc --- /dev/null +++ b/internal/kubernetes/resize_shutdown_test.go @@ -0,0 +1,247 @@ +package kubernetes //nolint:testpackage // Regression exercises the unexported execution and queue lifecycle. + +import ( + "context" + "errors" + "fmt" + "io" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.containerssh.io/containerssh/config" + "go.containerssh.io/containerssh/internal/metrics" + "go.containerssh.io/containerssh/log" + "k8s.io/client-go/tools/remotecommand" + "k8s.io/client-go/util/exec" +) + +func TestResizeQueueShutdown(t *testing.T) { + for _, mode := range []config.KubernetesExecutionMode{ + config.KubernetesExecutionModeConnection, + config.KubernetesExecutionModePersistent, + config.KubernetesExecutionModeSession, + } { + t.Run(string(mode), func(t *testing.T) { + tty := true + pod := &kubernetesPodImpl{ + config: config.KubernetesConfig{}, + tty: &tty, + wg: &sync.WaitGroup{}, + } + pod.config.Pod.Mode = mode + // PID framing is independent of resize shutdown. Bypass the agent so + // this regression also runs against versions with different framing. + pod.config.Pod.DisableAgent = true + pod.config.Timeouts.PodStop = time.Second + if mode == config.KubernetesExecutionModeConnection { + pod.wg.Add(1) + } + + collector := metrics.New(nil) + stream := &resizeExecutor{ + entered: make(chan struct{}), + received: make(chan remotecommand.TerminalSize, 1), + finish: make(chan struct{}), + } + execution := &kubernetesExecutionImpl{ + pod: pod, + exec: stream, + terminalSizeQueue: &pushSizeQueueImpl{resizeChan: make(chan remotecommand.TerminalSize)}, + logger: log.NewTestLogger(t), + doneChan: make(chan struct{}), + backendRequestsMetric: collector.MustCreateCounter( + "resize-shutdown-requests", "", "test"), + backendFailuresMetric: collector.MustCreateCounter( + "resize-shutdown-failures", "", "test"), + } + network := &networkHandler{ + mutex: &sync.Mutex{}, + config: config.KubernetesConfig{}, + logger: log.NewTestLogger(t), + } + network.config.Timeouts.Window = 2 * time.Second + handler := &channelHandler{networkHandler: network, exec: execution} + + exitStatus := make(chan int, 1) + streamDone := make(chan struct{}) + go func() { + execution.handleStream(nil, io.Discard, io.Discard, func() error { return nil }, func(code int) { + exitStatus <- code + }) + close(streamDone) + }() + select { + case <-stream.entered: + case <-time.After(time.Second): + t.Fatal("remotecommand executor did not enter StreamWithContext") + } + + // A live window request must reach the real execution's stream queue. + active := captureWindow(handler, 80, 24) + if active.panicValue != nil { + t.Fatalf("active OnWindow panicked: %v", active.panicValue) + } + require.NoError(t, active.err) + select { + case size := <-stream.received: + require.Equal(t, remotecommand.TerminalSize{Width: 80, Height: 24}, size) + case <-time.After(time.Second): + t.Fatal("active resize was not delivered to the remotecommand executor") + } + + // The executor stops reading sizes. Leave an OnWindow push pending + // while actual stream completion stops its queue. + const concurrentResizes = 4 + results := make(chan resizeCall, concurrentResizes) + started := make(chan struct{}, concurrentResizes) + for i := 0; i < concurrentResizes; i++ { + go func() { + started <- struct{}{} + results <- captureWindow(handler, 100, 40) + }() + } + for i := 0; i < concurrentResizes; i++ { + select { + case <-started: + case <-time.After(time.Second): + t.Fatal("concurrent resize caller did not start") + } + } + select { + case result := <-results: + t.Fatalf("resize completed before the stream stopped: %+v", result) + case <-time.After(25 * time.Millisecond): + } + + close(stream.finish) + select { + case <-streamDone: + case <-time.After(time.Second): + t.Fatal("execution did not finish after the remote stream returned") + } + require.Equal(t, 7, <-exitStatus) + + var failures []string + for i := 0; i < concurrentResizes; i++ { + select { + case result := <-results: + if result.panicValue != nil { + failures = append(failures, fmt.Sprintf("concurrent OnWindow panicked: %v", result.panicValue)) + } else if result.err == nil { + failures = append(failures, "concurrent resize succeeded after stream shutdown") + } + case <-time.After(time.Second): + failures = append(failures, "concurrent OnWindow remained blocked after stream shutdown") + } + } + + // A delayed window notification must return an error, never panic. + delayed := captureWindow(handler, 120, 50) + if delayed.panicValue != nil { + failures = append(failures, fmt.Sprintf("post-exit OnWindow panicked: %v", delayed.panicValue)) + } else if delayed.err == nil { + failures = append(failures, "post-exit OnWindow unexpectedly succeeded") + } + require.Empty(t, failures, "%s execution resize shutdown failures", mode) + }) + } + + t.Run("queue shutdown contract", func(t *testing.T) { + const prompt = 250 * time.Millisecond + + readerQueue := &pushSizeQueueImpl{resizeChan: make(chan remotecommand.TerminalSize)} + readResult := make(chan *remotecommand.TerminalSize, 1) + go func() { readResult <- readerQueue.Next() }() + select { + case size := <-readResult: + t.Fatalf("Next returned before shutdown: %v", size) + case <-time.After(25 * time.Millisecond): + } + readerQueue.Stop() + readerQueue.Stop() + select { + case size := <-readResult: + require.Nil(t, size, "Next must return nil after shutdown") + case <-time.After(prompt): + t.Fatal("Stop did not release a blocked Next") + } + + writerQueue := &pushSizeQueueImpl{resizeChan: make(chan remotecommand.TerminalSize)} + writeResult := make(chan resizePushResult, 1) + go func() { + panicValue, err := captureResizePush(writerQueue, context.Background()) + writeResult <- resizePushResult{err: err, panicValue: panicValue} + }() + select { + case result := <-writeResult: + t.Fatalf("Push returned before shutdown: %+v", result) + case <-time.After(25 * time.Millisecond): + } + writerQueue.Stop() + writerQueue.Stop() + select { + case result := <-writeResult: + require.Nil(t, result.panicValue, "Stop must not panic a blocked Push") + require.Error(t, result.err, "shutdown must release a pending Push with an error") + case <-time.After(prompt): + t.Fatal("Stop did not release a blocked Push") + } + + futurePanic, futureErr := captureResizePush(writerQueue, context.Background()) + require.Nil(t, futurePanic, "Push after Stop must not panic") + require.Error(t, futureErr, "Push after Stop must return an error") + + cancelledQueue := &pushSizeQueueImpl{resizeChan: make(chan remotecommand.TerminalSize)} + ctx, cancel := context.WithCancel(context.Background()) + cancel() + cancelledPanic, cancelledErr := captureResizePush(cancelledQueue, ctx) + require.Nil(t, cancelledPanic) + require.ErrorIs(t, cancelledErr, context.Canceled, "caller cancellation must be preserved") + cancelledQueue.Stop() + }) +} + +type resizeCall struct { + err error + panicValue interface{} +} + +func captureWindow(handler *channelHandler, columns, rows uint32) (result resizeCall) { + defer func() { result.panicValue = recover() }() + result.err = handler.OnWindow(1, columns, rows, 0, 0) + return result +} + +type resizePushResult struct { + err error + panicValue interface{} +} + +func captureResizePush(queue pushSizeQueue, ctx context.Context) (panicValue interface{}, err error) { + defer func() { panicValue = recover() }() + err = queue.Push(ctx, remotecommand.TerminalSize{Width: 80, Height: 24}) + return panicValue, err +} + +type resizeExecutor struct { + entered chan struct{} + received chan remotecommand.TerminalSize + finish chan struct{} +} + +func (s *resizeExecutor) Stream(options remotecommand.StreamOptions) error { + return s.StreamWithContext(context.Background(), options) +} + +func (s *resizeExecutor) StreamWithContext(_ context.Context, options remotecommand.StreamOptions) error { + close(s.entered) + size := options.TerminalSizeQueue.Next() + if size == nil { + return fmt.Errorf("terminal size queue stopped before active resize") + } + s.received <- *size + <-s.finish + return exec.CodeExitError{Err: errors.New("resize fixture exit"), Code: 7} +} diff --git a/internal/kubernetes/settings_pid_test.go b/internal/kubernetes/settings_pid_test.go new file mode 100644 index 00000000..d012a6ef --- /dev/null +++ b/internal/kubernetes/settings_pid_test.go @@ -0,0 +1,146 @@ +package kubernetes //nolint:testpackage // Exercises the unexported execution at the Kubernetes stream boundary. + +import ( + "bytes" + "context" + "fmt" + "io" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.containerssh.io/containerssh/config" + "go.containerssh.io/containerssh/internal/metrics" + "go.containerssh.io/containerssh/log" + "k8s.io/client-go/tools/remotecommand" + "k8s.io/client-go/util/exec" +) + +// Follow channelHandler's ordering: run must receive the agent PID before the +// initial resize. The stream emits no application output to rescue PID framing. +func TestSettingsPIDFramingAndResize(t *testing.T) { + for _, tc := range []struct { + mode config.KubernetesExecutionMode + tty bool + }{ + {config.KubernetesExecutionModeConnection, true}, + {config.KubernetesExecutionModePersistent, true}, + {config.KubernetesExecutionModeSession, true}, + {config.KubernetesExecutionModeConnection, false}, + {config.KubernetesExecutionModeSession, false}, + } { + t.Run(fmt.Sprintf("%s/tty=%t", tc.mode, tc.tty), func(t *testing.T) { + waitEcho := tc.mode == config.KubernetesExecutionModeSession && tc.tty + stream := &settingsPIDStream{pidWritten: make(chan struct{}), echo: make(chan bool), + size: make(chan remotecommand.TerminalSize, 1), finish: make(chan struct{}), waitEcho: waitEcho} + collector := metrics.New(nil) + pod := &kubernetesPodImpl{config: config.KubernetesConfig{}, tty: &tc.tty, wg: &sync.WaitGroup{}} + pod.config.Pod.Mode = tc.mode + if tc.mode == config.KubernetesExecutionModeConnection { + pod.wg.Add(1) + } + execution := &kubernetesExecutionImpl{pod: pod, exec: stream, tty: tc.tty, + terminalSizeQueue: &pushSizeQueueImpl{resizeChan: make(chan remotecommand.TerminalSize)}, + logger: log.NewTestLogger(t), doneChan: make(chan struct{}), lock: &sync.Mutex{}, + backendRequestsMetric: collector.MustCreateCounter("requests", "", "test"), + backendFailuresMetric: collector.MustCreateCounter("failures", "", "test")} + var stdout bytes.Buffer + runReturned, exited := make(chan struct{}), make(chan int, 1) + resizeResult := make(chan error, 1) + go func() { + execution.run(bytes.NewReader(nil), &stdout, io.Discard, func() error { return nil }, func(code int) { exited <- code }) + close(runReturned) + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + resizeResult <- execution.resize(ctx, 24, 80) + }() + select { + case <-stream.pidWritten: + case <-time.After(time.Second): + t.Fatal("stream did not write PID") + } + stalled := false + if waitEcho { + select { + case <-runReturned: + t.Error("session PTY must retain its six-byte PID/wait-echo framing") + case <-time.After(50 * time.Millisecond): + } + } else { + select { + case <-runReturned: + case <-time.After(500 * time.Millisecond): + t.Error("four-byte PID must unblock run and initial resize without application output") + stalled = true + } + } + // On red, supply the two missing bytes only to drain stock's blocked + // goroutines; they are never part of the successful exec-mode fixture. + stream.echo <- waitEcho || stalled + select { + case size := <-stream.size: + require.Equal(t, remotecommand.TerminalSize{Width: 80, Height: 24}, size) + case <-time.After(2 * time.Second): + t.Fatal("initial resize never reached the stream") + } + require.NoError(t, <-resizeResult) + close(stream.finish) + require.Equal(t, 9, <-exited) + require.Equal(t, 0x1234, execution.pid) + if !stalled { + if waitEcho { + require.Equal(t, "\r\n", stdout.String()) + } else { + require.Empty(t, stdout.String()) + } + } + }) + } +} + +// The remotecommand.Executor boundary supplies agent bytes, then requires a +// terminal size before producing anything else, as a terminal-dependent app does. +type settingsPIDStream struct { + pidWritten chan struct{} + echo chan bool + size chan remotecommand.TerminalSize + finish chan struct{} + waitEcho bool +} + +func (s *settingsPIDStream) Stream(options remotecommand.StreamOptions) error { + return s.StreamWithContext(context.Background(), options) +} + +func (s *settingsPIDStream) StreamWithContext(_ context.Context, options remotecommand.StreamOptions) error { + if s.waitEcho { + var start [1]byte + if _, err := io.ReadFull(options.Stdin, start[:]); err != nil { + return err + } + if start[0] != '\n' { + return fmt.Errorf("missing session --wait start newline") + } + } + // Fragment the four-byte little-endian PID at the transport boundary. + if _, err := options.Stdout.Write([]byte{0x34}); err != nil { + return err + } + if _, err := options.Stdout.Write([]byte{0x12, 0, 0}); err != nil { + return err + } + close(s.pidWritten) + if <-s.echo { + if _, err := options.Stdout.Write([]byte("\r\n")); err != nil { + return err + } + } + size := options.TerminalSizeQueue.Next() + if size == nil { + return fmt.Errorf("size queue closed before initial resize") + } + s.size <- *size + <-s.finish + return exec.CodeExitError{Err: fmt.Errorf("fixture exit"), Code: 9} +} diff --git a/internal/kubernetes/settings_rollback_test.go b/internal/kubernetes/settings_rollback_test.go new file mode 100644 index 00000000..c900af14 --- /dev/null +++ b/internal/kubernetes/settings_rollback_test.go @@ -0,0 +1,114 @@ +package kubernetes //nolint:testpackage // Exercises the real client/readiness lifecycle against an HTTP API fixture. + +import ( + "context" + "encoding/json" + "net" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.containerssh.io/containerssh/config" + "go.containerssh.io/containerssh/internal/metrics" + "go.containerssh.io/containerssh/internal/structutils" + "go.containerssh.io/containerssh/log" + "go.containerssh.io/containerssh/metadata" + core "k8s.io/api/core/v1" + meta "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// Use the real client, readiness watch and network-handler lifecycle against an +// HTTP API seam. No cluster/scheduler is needed to reproduce a pending pod leak. +func TestSettingsReadinessFailureRollback(t *testing.T) { + for _, tc := range []struct { + name string + mode config.KubernetesExecutionMode + deleteFails bool + wantDelete bool + }{ + {"connection healthy API", config.KubernetesExecutionModeConnection, false, true}, + {"connection unavailable delete", config.KubernetesExecutionModeConnection, true, true}, + {"persistent unchanged", config.KubernetesExecutionModePersistent, false, false}, + } { + t.Run(tc.name, func(t *testing.T) { + var lock sync.Mutex + var created, deleted int + var deleteAfterStartTimeout bool + var started time.Time + pod := core.Pod{TypeMeta: meta.TypeMeta{APIVersion: "v1", Kind: "Pod"}, + ObjectMeta: meta.ObjectMeta{Name: "settings-pending", Namespace: "sessions", UID: "created-uid", ResourceVersion: "1"}, + Status: core.PodStatus{Phase: core.PodPending}} + api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + switch { + case r.Method == http.MethodPost: + lock.Lock() + created++ + lock.Unlock() + w.WriteHeader(http.StatusCreated) + _ = json.NewEncoder(w).Encode(pod) + case r.Method == http.MethodDelete: + lock.Lock() + deleted++ + deleteAfterStartTimeout = time.Since(started) >= 300*time.Millisecond + lock.Unlock() + if tc.deleteFails { + w.WriteHeader(http.StatusServiceUnavailable) + _ = json.NewEncoder(w).Encode(meta.Status{Status: "Failure", Reason: meta.StatusReasonServiceUnavailable, Code: 503}) + } else { + // Kubernetes can acknowledge deletion while finalizers retain the object. + _ = json.NewEncoder(w).Encode(meta.Status{Status: "Success", Code: 200}) + } + case r.URL.Query().Get("watch") == "true": + _ = json.NewEncoder(w).Encode(map[string]interface{}{"type": "ADDED", "object": pod}) + w.(http.Flusher).Flush() + <-r.Context().Done() + default: + _ = json.NewEncoder(w).Encode(core.PodList{TypeMeta: meta.TypeMeta{APIVersion: "v1", Kind: "PodList"}, + ListMeta: meta.ListMeta{ResourceVersion: "1"}, Items: []core.Pod{pod}}) + } + })) + defer api.Close() + cfg := config.KubernetesConfig{} + structutils.Defaults(&cfg) + cfg.Pod.Mode = tc.mode + cfg.Connection.Host = api.URL + cfg.Timeouts.PodStart = 300 * time.Millisecond + cfg.Timeouts.PodStop = 200 * time.Millisecond + logger := log.NewTestLogger(t) + collector := metrics.New(nil) + factory := &kubernetesClientFactoryImpl{ + backendRequestsMetric: collector.MustCreateCounter("requests", "", "test"), + backendFailuresMetric: collector.MustCreateCounter("failures", "", "test"), + } + cli, err := factory.get(context.Background(), cfg, logger) + require.NoError(t, err) + network := &networkHandler{mutex: &sync.Mutex{}, cli: cli, config: cfg, logger: logger, + client: net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)}, connectionID: "settings", done: make(chan struct{})} + // Persistent mode uses findPod first; exercise creation directly to avoid its stock retry sleeps. + started = time.Now() + if tc.mode == config.KubernetesExecutionModePersistent { + ctx, cancel := context.WithTimeout(context.Background(), cfg.Timeouts.PodStart) + defer cancel() + _, err = cli.createPod(ctx, nil, nil, nil, nil, nil) + } else { + _, _, err = network.OnHandshakeSuccess(metadata.ConnectionAuthenticatedMetadata{AuthenticatedUsername: "owner"}) + } + require.Error(t, err, "pending pod must fail startup") + require.Less(t, time.Since(started), 2*time.Second, "rollback must not reuse a background context without a deadline") + network.OnDisconnect() + lock.Lock() + defer lock.Unlock() + require.Equal(t, 1, created) + if tc.wantDelete { + require.Equal(t, 1, deleted, "failed readiness must not orphan the created connection pod") + require.True(t, deleteAfterStartTimeout, "rollback uses a fresh context after startup expires") + } else { + require.Zero(t, deleted, "persistent pods retain their existing lifecycle") + } + }) + } +} diff --git a/internal/security/handler_session.go b/internal/security/handler_session.go index 33d50240..0abac0e5 100644 --- a/internal/security/handler_session.go +++ b/internal/security/handler_session.go @@ -3,10 +3,10 @@ package security import ( "context" - config2 "go.containerssh.io/containerssh/config" - "go.containerssh.io/containerssh/internal/sshserver" - "go.containerssh.io/containerssh/log" - "go.containerssh.io/containerssh/message" + config2 "go.containerssh.io/containerssh/config" + "go.containerssh.io/containerssh/internal/sshserver" + "go.containerssh.io/containerssh/log" + "go.containerssh.io/containerssh/message" ) type sessionHandler struct { @@ -262,7 +262,7 @@ func (s *sessionHandler) OnSubsystem( } func (s *sessionHandler) OnSignal(requestID uint64, signal string) error { - mode := s.getPolicy(s.config.Shell.Mode) + mode := s.getPolicy(s.config.Signal.Mode) switch mode { case config2.ExecutionPolicyDisable: err := message.UserMessage( diff --git a/internal/security/settings_signal_test.go b/internal/security/settings_signal_test.go new file mode 100644 index 00000000..742388e2 --- /dev/null +++ b/internal/security/settings_signal_test.go @@ -0,0 +1,51 @@ +package security //nolint:testpackage // Exercises internal policy routing, consistent with existing security tests. + +import ( + "testing" + + "github.com/stretchr/testify/require" + "go.containerssh.io/containerssh/config" + "go.containerssh.io/containerssh/log" +) + +func TestSettingsSignalPolicyIndependentOfShell(t *testing.T) { + for _, tc := range []struct { + name string + shell, signal config.SecurityExecutionPolicy + allow, deny []string + wantForward bool + }{ + {"disabled with shell enabled", config.ExecutionPolicyEnable, config.ExecutionPolicyDisable, nil, nil, false}, + {"enabled with shell disabled", config.ExecutionPolicyDisable, config.ExecutionPolicyEnable, nil, nil, true}, + {"filter allowed", config.ExecutionPolicyDisable, config.ExecutionPolicyFilter, []string{"TERM"}, nil, true}, + {"filter unlisted", config.ExecutionPolicyEnable, config.ExecutionPolicyFilter, []string{"INT"}, nil, false}, + {"enabled denied", config.ExecutionPolicyEnable, config.ExecutionPolicyEnable, nil, []string{"TERM"}, false}, + } { + t.Run(tc.name, func(t *testing.T) { + backend := &settingsSignalBackend{} + session := &sessionHandler{backend: backend, logger: log.NewTestLogger(t)} + session.config.Shell.Mode = tc.shell + session.config.Signal.Mode = tc.signal + session.config.Signal.Allow = tc.allow + session.config.Signal.Deny = tc.deny + err := session.OnSignal(7, "TERM") + if tc.wantForward { + require.NoError(t, err) + require.Equal(t, []string{"TERM"}, backend.signals) + } else { + require.Error(t, err) + require.Empty(t, backend.signals) + } + }) + } +} + +type settingsSignalBackend struct { + dummyBackend + signals []string +} + +func (b *settingsSignalBackend) OnSignal(_ uint64, signal string) error { + b.signals = append(b.signals, signal) + return nil +} diff --git a/internal/sshserver/serverImpl.go b/internal/sshserver/serverImpl.go index 17a169d1..43dab5a2 100644 --- a/internal/sshserver/serverImpl.go +++ b/internal/sshserver/serverImpl.go @@ -858,6 +858,7 @@ func (s *serverImpl) handleSessionChannel( channelMetadata.Connection.ConnectionID, channelMetadata.ChannelID, ) + handlerChannel = &startOnceSessionChannelHandler{SessionChannelHandler: handlerChannel} s.shutdownHandlers.Register(shutdownHandlerID, handlerChannel) channel, requests, err := newChannel.Accept() if err != nil { diff --git a/internal/sshserver/settings_start_test.go b/internal/sshserver/settings_start_test.go new file mode 100644 index 00000000..3f564da8 --- /dev/null +++ b/internal/sshserver/settings_start_test.go @@ -0,0 +1,126 @@ +package sshserver_test + +import ( + "fmt" + "net" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.containerssh.io/containerssh/internal/sshserver" + "go.containerssh.io/containerssh/log" + "go.containerssh.io/containerssh/metadata" + "golang.org/x/crypto/ssh" +) + +// Exercise the real request/reply loop, not a backend that already refuses restarts. +func TestSettingsSingleAcceptedStart(t *testing.T) { + logger := log.NewTestLogger(t) + user := sshserver.NewTestUser("settings") + user.SetPassword("test") + handler := &settingsStartHandler{} + srv := sshserver.NewTestServer(t, sshserver.NewTestAuthenticationHandler(handler, user), logger, nil) + srv.Start() + defer srv.Stop(5 * time.Second) + conn, err := net.DialTimeout("tcp", srv.GetListen(), 5*time.Second) + require.NoError(t, err) + defer func() { _ = conn.Close() }() + require.NoError(t, conn.SetDeadline(time.Now().Add(20*time.Second))) + hostKey, err := ssh.ParsePrivateKey([]byte(srv.GetHostKey())) + require.NoError(t, err) + sshConn, channels, requests, err := ssh.NewClientConn(conn, srv.GetListen(), &ssh.ClientConfig{ + User: "settings", Auth: []ssh.AuthMethod{ssh.Password("test")}, + HostKeyCallback: ssh.FixedHostKey(hostKey.PublicKey()), + }) + require.NoError(t, err) + client := ssh.NewClient(sshConn, channels, requests) + defer func() { _ = client.Close() }() + starts := []struct { + name string + payload []byte + }{ + {"shell", nil}, + {"exec", ssh.Marshal(struct{ Command string }{"/app"})}, + {"subsystem", ssh.Marshal(struct{ Name string }{"sftp"})}, + } + for _, first := range starts { + t.Run(first.name, func(t *testing.T) { + channel, replies, err := client.OpenChannel("session", nil) + require.NoError(t, err) + defer func() { _ = channel.Close() }() + go ssh.DiscardRequests(replies) + before := handler.attempts.Load() + // A backend-rejected start must not consume the channel's opportunity. + ok, err := channel.SendRequest(first.name, true, first.payload) + require.NoError(t, err) + require.False(t, ok) + ok, err = channel.SendRequest(first.name, true, first.payload) + require.NoError(t, err) + require.True(t, ok) + for _, next := range starts { + ok, err = channel.SendRequest(next.name, true, next.payload) + require.NoError(t, err) + require.False(t, ok, "second %s after %s must be rejected", next.name, first.name) + } + // Exit status doesn't make the same channel reusable; non-start requests still work. + ok, err = channel.SendRequest("env", true, ssh.Marshal(struct{ Name, Value string }{"exit", "0"})) + require.NoError(t, err) + require.True(t, ok) + ok, err = channel.SendRequest("shell", true, nil) + require.NoError(t, err) + require.False(t, ok) + require.Equal(t, before+2, handler.attempts.Load(), "only failed start and first accepted start reach backend") + }) + } +} + +type settingsStartHandler struct { + sshserver.AbstractHandler + attempts atomic.Int32 +} + +func (h *settingsStartHandler) OnNetworkConnection(meta metadata.ConnectionMetadata) (sshserver.NetworkConnectionHandler, metadata.ConnectionMetadata, error) { + return &settingsStartNetwork{handler: h}, meta, nil +} + +type settingsStartNetwork struct { + sshserver.AbstractNetworkConnectionHandler + handler *settingsStartHandler +} + +func (n *settingsStartNetwork) OnHandshakeSuccess(meta metadata.ConnectionAuthenticatedMetadata) (sshserver.SSHConnectionHandler, metadata.ConnectionAuthenticatedMetadata, error) { + return &settingsStartConnection{handler: n.handler}, meta, nil +} + +type settingsStartConnection struct { + fullSSHConnectionHandler + handler *settingsStartHandler +} + +func (c *settingsStartConnection) OnSessionChannel(_ metadata.ChannelMetadata, _ []byte, channel sshserver.SessionChannel) (sshserver.SessionChannelHandler, sshserver.ChannelRejection) { + return &settingsStartSession{handler: c.handler, channel: channel, rejectFirst: true}, nil +} + +type settingsStartSession struct { + sshserver.AbstractSessionChannelHandler + handler *settingsStartHandler + channel sshserver.SessionChannel + rejectFirst bool +} + +func (s *settingsStartSession) start() error { + s.handler.attempts.Add(1) + if s.rejectFirst { + s.rejectFirst = false + return fmt.Errorf("start failed") + } + return nil +} +func (s *settingsStartSession) OnShell(_ uint64) error { return s.start() } +func (s *settingsStartSession) OnSubsystem(_ uint64, _ string) error { return s.start() } +func (s *settingsStartSession) OnExecRequest(_ uint64, _ string) error { return s.start() } +func (s *settingsStartSession) OnEnvRequest(_ uint64, _, _ string) error { + s.channel.ExitStatus(0) + return nil +} diff --git a/internal/sshserver/startOnceSessionChannelHandler.go b/internal/sshserver/startOnceSessionChannelHandler.go new file mode 100644 index 00000000..3130eec6 --- /dev/null +++ b/internal/sshserver/startOnceSessionChannelHandler.go @@ -0,0 +1,42 @@ +// Package sshserver implements SSH transport and dispatches requests to backend handlers. +package sshserver + +import ( + "fmt" + "sync" +) + +// startOnceSessionChannelHandler permits one successful program start over the +// channel's lifetime. Failed starts remain retryable, as in the stock server. +// The lock is separate from channelWrapper's IO lock: a backend start may send +// exit status or close the channel synchronously. +type startOnceSessionChannelHandler struct { + SessionChannelHandler + startLock sync.Mutex + started bool +} + +func (s *startOnceSessionChannelHandler) start(start func() error) error { + s.startLock.Lock() + defer s.startLock.Unlock() + if s.started { + return fmt.Errorf("a program has already been started on this channel") + } + if err := start(); err != nil { + return err + } + s.started = true + return nil +} + +func (s *startOnceSessionChannelHandler) OnShell(requestID uint64) error { + return s.start(func() error { return s.SessionChannelHandler.OnShell(requestID) }) +} + +func (s *startOnceSessionChannelHandler) OnExecRequest(requestID uint64, command string) error { + return s.start(func() error { return s.SessionChannelHandler.OnExecRequest(requestID, command) }) +} + +func (s *startOnceSessionChannelHandler) OnSubsystem(requestID uint64, subsystem string) error { + return s.start(func() error { return s.SessionChannelHandler.OnSubsystem(requestID, subsystem) }) +}