From 0b47dd778a7635bc2b38881fa9ea570ad29215f0 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Thu, 18 Jun 2026 17:11:04 +0800 Subject: [PATCH 01/14] fix(cli): harden ssh fallback paths --- cmd/crabbox-ssh-gateway/main.go | 80 +++++++++++---- cmd/crabbox-ssh-gateway/main_test.go | 110 +++++++++++++++++++- cmd/crabfleet/main.go | 145 ++++++++++++++++----------- cmd/crabfleet/main_test.go | 133 +++++++++++++++++++++++- internal/fleetapi/client.go | 4 +- internal/terminalws/client.go | 20 +++- internal/terminalws/client_test.go | 98 ++++++++++++++++-- 7 files changed, 497 insertions(+), 93 deletions(-) diff --git a/cmd/crabbox-ssh-gateway/main.go b/cmd/crabbox-ssh-gateway/main.go index dcc42255..1b4f5b42 100644 --- a/cmd/crabbox-ssh-gateway/main.go +++ b/cmd/crabbox-ssh-gateway/main.go @@ -22,6 +22,11 @@ import ( "golang.org/x/crypto/ssh" ) +var ( + sshHandshakeTimeout = 15 * time.Second + sshAuthTimeout = 15 * time.Second +) + type apiClient struct { baseURL string token string @@ -86,8 +91,10 @@ func main() { ServerVersion: "SSH-2.0-Crabfleet", PublicKeyCallback: func(meta ssh.ConnMetadata, key ssh.PublicKey) (*ssh.Permissions, error) { linkMode := meta.User() == "link" || meta.User() == "onboard" + authCtx, cancel := context.WithTimeout(context.Background(), sshAuthTimeout) + defer cancel() auth, err := client.auth( - context.Background(), + authCtx, key, meta.User(), remoteHost(meta.RemoteAddr()), @@ -131,11 +138,17 @@ func main() { func handleConn(raw net.Conn, config *ssh.ServerConfig, client *apiClient) { defer raw.Close() + if err := raw.SetDeadline(time.Now().Add(sshHandshakeTimeout)); err != nil { + log.Printf("handshake deadline %s: %v", raw.RemoteAddr(), err) + } conn, chans, reqs, err := ssh.NewServerConn(raw, config) if err != nil { log.Printf("handshake %s: %v", raw.RemoteAddr(), err) return } + if err := raw.SetDeadline(time.Time{}); err != nil { + log.Printf("clear handshake deadline %s: %v", raw.RemoteAddr(), err) + } defer conn.Close() go ssh.DiscardRequests(reqs) @@ -304,7 +317,11 @@ func runCommand(ctx context.Context, out io.ReadWriter, perms *ssh.Permissions, printList(out, state) return 0 case "new": - create := parseCreate(args[1:], api) + create, err := parseCreate(args[1:], api) + if err != nil { + fmt.Fprintf(out, "usage: new [--repo owner/repo] [--branch main] [--runtime crabbox|container] [--profile name] [prompt]\nerror: %v\n", err) + return 2 + } session, err := api.CreateSession(ctx, create.request) if err != nil { fmt.Fprintf(out, "error: %v\n", err) @@ -403,7 +420,11 @@ func runCommand(ctx context.Context, out io.ReadWriter, perms *ssh.Permissions, fmt.Fprintln(out, "usage: message SESSION_ID [--no-enter] TEXT") return 2 } - message := parseMessage(args[2:]) + message, err := parseMessage(args[2:]) + if err != nil { + fmt.Fprintf(out, "usage: message SESSION_ID [--no-enter] TEXT\nerror: %v\n", err) + return 2 + } if message.text == "" { fmt.Fprintln(out, "usage: message SESSION_ID [--no-enter] TEXT") return 2 @@ -419,7 +440,11 @@ func runCommand(ctx context.Context, out io.ReadWriter, perms *ssh.Permissions, fmt.Fprintln(out, "usage: summary SESSION_ID [--purpose text] [summary text]") return 2 } - update := parseSummary(args[2:]) + update, err := parseSummary(args[2:]) + if err != nil { + fmt.Fprintf(out, "usage: summary SESSION_ID [--purpose text] [summary text]\nerror: %v\n", err) + return 2 + } if update.summary == "" && update.purpose == "" { state, err := api.State(ctx) if err != nil { @@ -569,7 +594,7 @@ func splitCommand(command string) ([]string, error) { return args, nil } -func parseCreate(args []string, api *fleetapi.Client) createArgs { +func parseCreate(args []string, api *fleetapi.Client) (createArgs, error) { fs := flag.NewFlagSet("new", flag.ContinueOnError) fs.SetOutput(io.Discard) var req fleetapi.CreateSessionRequest @@ -586,14 +611,16 @@ func parseCreate(args []string, api *fleetapi.Client) createArgs { fs.StringVar(&req.Summary, "summary", "", "summary") fs.BoolVar(&detach, "detach", false, "do not attach after creating") fs.BoolVar(&vnc, "vnc", false, "print vnc URL without attaching") - _ = fs.Parse(args) + if err := fs.Parse(args); err != nil { + return createArgs{}, err + } req.Prompt = strings.Join(fs.Args(), " ") if req.Repo == "" && api != nil { if state, err := api.State(context.Background()); err == nil && len(state.Repos) > 0 { req.Repo = state.Repos[0] } } - return createArgs{request: req, detach: detach, vnc: vnc} + return createArgs{request: req, detach: detach, vnc: vnc}, nil } type summaryUpdate struct { @@ -606,24 +633,35 @@ type messageInput struct { noEnter bool } -func parseMessage(args []string) messageInput { - fs := flag.NewFlagSet("message", flag.ContinueOnError) - fs.SetOutput(io.Discard) +func parseMessage(args []string) (messageInput, error) { var input messageInput - fs.BoolVar(&input.noEnter, "no-enter", false, "do not append enter") - _ = fs.Parse(args) - input.text = strings.Join(fs.Args(), " ") - return input + remaining := args + if len(remaining) > 0 && remaining[0] == "--no-enter" { + input.noEnter = true + remaining = remaining[1:] + } + if len(remaining) > 0 && remaining[0] == "--" { + remaining = remaining[1:] + } + input.text = strings.Join(remaining, " ") + return input, nil } -func parseSummary(args []string) summaryUpdate { - fs := flag.NewFlagSet("summary", flag.ContinueOnError) - fs.SetOutput(io.Discard) +func parseSummary(args []string) (summaryUpdate, error) { var update summaryUpdate - fs.StringVar(&update.purpose, "purpose", "", "purpose") - _ = fs.Parse(args) - update.summary = strings.Join(fs.Args(), " ") - return update + remaining := args + if len(remaining) > 0 && remaining[0] == "--purpose" { + if len(remaining) < 2 { + return summaryUpdate{}, errors.New("flag needs an argument: -purpose") + } + update.purpose = remaining[1] + remaining = remaining[2:] + } + if len(remaining) > 0 && remaining[0] == "--" { + remaining = remaining[1:] + } + update.summary = strings.Join(remaining, " ") + return update, nil } func (c *apiClient) auth(ctx context.Context, key ssh.PublicKey, sshUser string, remote string, createLink bool) (keyAuth, error) { diff --git a/cmd/crabbox-ssh-gateway/main_test.go b/cmd/crabbox-ssh-gateway/main_test.go index f201d1f1..b24ef6de 100644 --- a/cmd/crabbox-ssh-gateway/main_test.go +++ b/cmd/crabbox-ssh-gateway/main_test.go @@ -4,11 +4,13 @@ import ( "bytes" "context" "encoding/json" + "net" "net/http" "net/http/httptest" "reflect" "strings" "testing" + "time" "github.com/openclaw/crabfleet/internal/fleetapi" "golang.org/x/crypto/ssh" @@ -64,7 +66,7 @@ func TestSplitCommandPreservesBackslashesInSingleQuotes(t *testing.T) { } func TestParseCreateKeepsLineageAndSummaryFlags(t *testing.T) { - create := parseCreate( + create, err := parseCreate( []string{ "--repo", "openclaw/crabfleet", "--parent", "IS-1", @@ -75,6 +77,9 @@ func TestParseCreateKeepsLineageAndSummaryFlags(t *testing.T) { }, nil, ) + if err != nil { + t.Fatal(err) + } if got, want := create.request.ParentSessionID, "IS-1"; got != want { t.Fatalf("parent = %q, want %q", got, want) } @@ -93,35 +98,76 @@ func TestParseCreateKeepsLineageAndSummaryFlags(t *testing.T) { } func TestParseMessageKeepsNoEnterAndText(t *testing.T) { - message := parseMessage([]string{"--no-enter", "hello", "child"}) + message, err := parseMessage([]string{"--no-enter", "hello", "child"}) + if err != nil { + t.Fatal(err) + } if !message.noEnter { t.Fatal("expected no-enter") } if got, want := message.text, "hello child"; got != want { t.Fatalf("text = %q, want %q", got, want) } + + message, err = parseMessage([]string{"--help", "is", "terminal", "text"}) + if err != nil { + t.Fatal(err) + } + if got, want := message.text, "--help is terminal text"; got != want { + t.Fatalf("text = %q, want %q", got, want) + } +} + +func TestParseSummaryKeepsDashPrefixedText(t *testing.T) { + summary, err := parseSummary([]string{"--looks-like-flag", "but", "is", "text"}) + if err != nil { + t.Fatal(err) + } + if got, want := summary.summary, "--looks-like-flag but is text"; got != want { + t.Fatalf("summary = %q, want %q", got, want) + } + + summary, err = parseSummary([]string{"--purpose", "handoff", "--", "--summary-start"}) + if err != nil { + t.Fatal(err) + } + if got, want := summary.purpose, "handoff"; got != want { + t.Fatalf("purpose = %q, want %q", got, want) + } + if got, want := summary.summary, "--summary-start"; got != want { + t.Fatalf("summary = %q, want %q", got, want) + } } func TestParseCreateLeavesRuntimeToDeploymentDefault(t *testing.T) { - create := parseCreate([]string{"--repo", "openclaw/crabfleet", "fix it"}, nil) + create, err := parseCreate([]string{"--repo", "openclaw/crabfleet", "fix it"}, nil) + if err != nil { + t.Fatal(err) + } if create.request.Runtime != "" { t.Fatalf("runtime = %q, want deployment default", create.request.Runtime) } - create = parseCreate( + create, err = parseCreate( []string{"--repo", "openclaw/crabfleet", "--runtime", "container", "fix it"}, nil, ) + if err != nil { + t.Fatal(err) + } if create.request.Runtime != "container" { t.Fatalf("runtime = %q, want explicit override", create.request.Runtime) } } func TestParseCreateAcceptsProfileOverride(t *testing.T) { - create := parseCreate( + create, err := parseCreate( []string{"--repo", "openclaw/crabfleet", "--profile", "desktop-a", "fix it"}, nil, ) + if err != nil { + t.Fatal(err) + } if create.request.Profile != "desktop-a" { t.Fatalf("profile = %q, want explicit override", create.request.Profile) } @@ -194,6 +240,60 @@ func TestDeleteCommandUsesWorkspaceStopAction(t *testing.T) { } } +func TestInvalidSubcommandFlagsDoNotCallControlPlane(t *testing.T) { + calls := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls++ + w.WriteHeader(http.StatusInternalServerError) + })) + defer server.Close() + + client := &apiClient{baseURL: server.URL, token: "gateway-token", client: server.Client()} + permissions := &ssh.Permissions{Extensions: map[string]string{ + "authorized": "true", + "fingerprint": "SHA256:test", + "login": "operator", + "role": "owner", + }} + for _, command := range []string{ + "new --repo", + "new --bogus value", + "message IS-7 --no-enter", + "summary IS-7 --purpose", + } { + var output bytes.Buffer + if exit := runCommand(context.Background(), &output, permissions, client, command, sessionPTY{}); exit != 2 { + t.Fatalf("command=%q exit=%d output=%q", command, exit, output.String()) + } + if !strings.Contains(output.String(), "usage:") { + t.Fatalf("command=%q output=%q", command, output.String()) + } + } + if calls != 0 { + t.Fatalf("control plane calls = %d, want 0", calls) + } +} + +func TestHandleConnClosesStalledHandshake(t *testing.T) { + previous := sshHandshakeTimeout + sshHandshakeTimeout = 20 * time.Millisecond + defer func() { sshHandshakeTimeout = previous }() + + server, client := net.Pipe() + defer client.Close() + done := make(chan struct{}) + go func() { + handleConn(server, &ssh.ServerConfig{}, nil) + close(done) + }() + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("stalled handshake was not closed") + } +} + func TestHelpNamesDeleteAsCanonicalCommand(t *testing.T) { var output bytes.Buffer printHelp(&output, fleetapi.User{Login: "operator", Role: "owner"}) diff --git a/cmd/crabfleet/main.go b/cmd/crabfleet/main.go index e235b374..0b429390 100644 --- a/cmd/crabfleet/main.go +++ b/cmd/crabfleet/main.go @@ -195,29 +195,35 @@ func (cmd newCmd) Run(app *cli, api *fleetapi.Client) error { req := cmd.sessionRequest(app) session, err := api.CreateSession(context.Background(), req) if err != nil { - if app.NoInput || app.JSON { - return err - } - args := cmd.sshCreateArgs(req) - if cmd.VNC { - output, captureErr := runSSHCommandOutput(app, args...) - if output != "" { - fmt.Fprint(os.Stdout, output) - } - if captureErr != nil { - return captureErr + if canFallbackToSSH(app, err) { + args := cmd.sshCreateArgs(req) + if cmd.VNC { + output, captureErr := runSSHCommandOutput(app, args...) + if output != "" { + fmt.Fprint(os.Stdout, output) + } + if captureErr != nil { + return captureErr + } + if url := vncURLFromOutput(output); url != "" { + return openURL(url) + } + return nil } - if url := vncURLFromOutput(output); url != "" { - return openURL(url) - } - return nil + return runSSHCommand(app, args...) } - return runSSHCommand(app, args...) + return ambiguousMutationError("create session", err) } if app.JSON { return json.NewEncoder(os.Stdout).Encode(session) } - fmt.Fprintf(os.Stdout, "session: %s\nrepo: %s\nstatus: %s\n", session.ID, session.Repo, session.Status) + fmt.Fprintf( + os.Stdout, + "session: %s\nrepo: %s\nstatus: %s\n", + fleettext.Safe(session.ID), + fleettext.Safe(session.Repo), + fleettext.Safe(session.Status), + ) if session.ParentSessionID != "" { fmt.Fprintf(os.Stdout, "parent: %s\n", fleettext.Safe(session.ParentSessionID)) } @@ -228,10 +234,10 @@ func (cmd newCmd) Run(app *cli, api *fleetapi.Client) error { fmt.Fprintf(os.Stdout, "summary: %s\n", fleettext.Safe(session.Summary)) } if session.Attachable() { - fmt.Fprintf(os.Stdout, "attach: crabfleet attach %s\n", session.ID) + fmt.Fprintf(os.Stdout, "attach: crabfleet attach %s\n", fleettext.Safe(session.ID)) } if session.VNCURL != "" { - fmt.Fprintf(os.Stdout, "vnc: %s\n", session.VNCURL) + fmt.Fprintf(os.Stdout, "vnc: %s\n", fleettext.Safe(session.VNCURL)) } if cmd.VNC && session.VNCURL != "" { return openURL(session.VNCURL) @@ -303,7 +309,7 @@ func (cmd newCmd) sshCreateArgs(req fleetapi.CreateSessionRequest) []string { args = append(args, "--vnc") } if req.Prompt != "" { - args = append(args, req.Prompt) + args = append(args, "--", req.Prompt) } return args } @@ -330,17 +336,17 @@ func (cmd statusCmd) Run(app *cli, api *fleetapi.Client) error { func (cmd deleteCmd) Run(app *cli, api *fleetapi.Client) error { session, err := api.Action(context.Background(), cmd.ID, "stop") if err != nil { - if app.NoInput || app.JSON { - return err + if canFallbackToSSH(app, err) { + return runSSH(app, "delete", cmd.ID) } - return runSSH(app, "delete", cmd.ID) + return ambiguousMutationError("delete session", err) } if app.JSON { return json.NewEncoder(os.Stdout).Encode(session) } - fmt.Fprintf(os.Stdout, "session: %s\nstatus: %s\n", session.ID, session.Status) + fmt.Fprintf(os.Stdout, "session: %s\nstatus: %s\n", fleettext.Safe(session.ID), fleettext.Safe(session.Status)) if note := session.LifecycleStopNote(); note != "" { - fmt.Fprintf(os.Stdout, "note: %s\n", note) + fmt.Fprintf(os.Stdout, "note: %s\n", fleettext.Safe(note)) } return nil } @@ -388,17 +394,17 @@ func (cmd checkpointsCmd) Run(app *cli, api *fleetapi.Client) error { return json.NewEncoder(os.Stdout).Encode(checkpoints) } if len(checkpoints.Checkpoints) == 0 { - fmt.Fprintf(os.Stdout, "session: %s\ncheckpoints: none\n", checkpoints.Session.ID) + fmt.Fprintf(os.Stdout, "session: %s\ncheckpoints: none\n", fleettext.Safe(checkpoints.Session.ID)) return nil } - fmt.Fprintf(os.Stdout, "session: %s\n", checkpoints.Session.ID) + fmt.Fprintf(os.Stdout, "session: %s\n", fleettext.Safe(checkpoints.Session.ID)) for _, checkpoint := range checkpoints.Checkpoints { fmt.Fprintf( os.Stdout, "%s %s %s\n", - checkpoint.ID, + fleettext.Safe(checkpoint.ID), time.UnixMilli(checkpoint.CreatedAt).Format(time.RFC3339), - checkpoint.Workdir, + fleettext.Safe(checkpoint.Workdir), ) } return nil @@ -407,30 +413,40 @@ func (cmd checkpointsCmd) Run(app *cli, api *fleetapi.Client) error { func (cmd checkpointCmd) Run(app *cli, api *fleetapi.Client) error { checkpoint, err := api.Checkpoint(context.Background(), cmd.ID) if err != nil { - if app.NoInput || app.JSON { - return err + if canFallbackToSSH(app, err) { + return runSSH(app, "checkpoint", cmd.ID) } - return runSSH(app, "checkpoint", cmd.ID) + return ambiguousMutationError("create checkpoint", err) } if app.JSON { return json.NewEncoder(os.Stdout).Encode(checkpoint) } - fmt.Fprintf(os.Stdout, "session: %s\ncheckpoint: %s\n", checkpoint.Session.ID, checkpoint.Checkpoint.ID) + fmt.Fprintf( + os.Stdout, + "session: %s\ncheckpoint: %s\n", + fleettext.Safe(checkpoint.Session.ID), + fleettext.Safe(checkpoint.Checkpoint.ID), + ) return nil } func (cmd restoreCmd) Run(app *cli, api *fleetapi.Client) error { checkpoint, err := api.Restore(context.Background(), cmd.ID, cmd.Checkpoint) if err != nil { - if app.NoInput || app.JSON { - return err + if canFallbackToSSH(app, err) { + return runSSH(app, "restore", cmd.ID, cmd.Checkpoint) } - return runSSH(app, "restore", cmd.ID, cmd.Checkpoint) + return ambiguousMutationError("restore checkpoint", err) } if app.JSON { return json.NewEncoder(os.Stdout).Encode(checkpoint) } - fmt.Fprintf(os.Stdout, "session: %s\nrestored: %s\n", checkpoint.Session.ID, checkpoint.Checkpoint.ID) + fmt.Fprintf( + os.Stdout, + "session: %s\nrestored: %s\n", + fleettext.Safe(checkpoint.Session.ID), + fleettext.Safe(checkpoint.Checkpoint.ID), + ) return nil } @@ -463,7 +479,7 @@ func (cmd vncCmd) Run(app *cli, api *fleetapi.Client) error { if cmd.Open { return openURL(session.VNCURL) } - fmt.Fprintln(os.Stdout, session.VNCURL) + fmt.Fprintln(os.Stdout, fleettext.Safe(session.VNCURL)) return nil } return fmt.Errorf("session %s not found", cmd.ID) @@ -518,15 +534,15 @@ func (cmd messageCmd) Run(app *cli, api *fleetapi.Client) error { return errors.New("message text is required") } if err := api.Message(context.Background(), cmd.ID, message, !cmd.NoEnter, 120, 34); err != nil { - if app.NoInput || app.JSON { - return err - } - args := []string{"message", cmd.ID} - if cmd.NoEnter { - args = append(args, "--no-enter") + if canFallbackToSSH(app, err) { + args := []string{"message", cmd.ID} + if cmd.NoEnter { + args = append(args, "--no-enter") + } + args = append(args, message) + return runSSHCommand(app, args...) } - args = append(args, message) - return runSSHCommand(app, args...) + return ambiguousMutationError("send terminal input", err) } if app.JSON { return json.NewEncoder(os.Stdout).Encode(map[string]any{ @@ -556,17 +572,17 @@ func (cmd summaryCmd) Run(app *cli, api *fleetapi.Client) error { } session, err := api.UpdateSummary(context.Background(), cmd.ID, summary, cmd.Purpose) if err != nil { - if app.NoInput || app.JSON { - return err - } - args := []string{"summary", cmd.ID} - if cmd.Purpose != "" { - args = append(args, "--purpose", cmd.Purpose) - } - if summary != "" { - args = append(args, summary) + if canFallbackToSSH(app, err) { + args := []string{"summary", cmd.ID} + if cmd.Purpose != "" { + args = append(args, "--purpose", cmd.Purpose) + } + if summary != "" { + args = append(args, summary) + } + return runSSHCommand(app, args...) } - return runSSHCommand(app, args...) + return ambiguousMutationError("update summary", err) } if app.JSON { return json.NewEncoder(os.Stdout).Encode(session) @@ -617,13 +633,28 @@ func shellQuote(value string) string { return "''" } if strings.IndexFunc(value, func(r rune) bool { - return r == ' ' || r == '\t' || r == '\n' || r == '\r' || r == '\'' || r == '"' || r == '\\' + return !isShellSafeRune(r) }) == -1 { return value } return "'" + strings.ReplaceAll(value, "'", "'\\''") + "'" } +func isShellSafeRune(r rune) bool { + return (r >= 'A' && r <= 'Z') || + (r >= 'a' && r <= 'z') || + (r >= '0' && r <= '9') || + strings.ContainsRune("_@%+=:,./-", r) +} + +func ambiguousMutationError(operation string, err error) error { + return fmt.Errorf("%s may have reached Crabfleet before confirmation; not retrying through SSH: %w", operation, err) +} + +func canFallbackToSSH(app *cli, err error) bool { + return !app.NoInput && !app.JSON && errors.Is(err, fleetapi.ErrMissingAuth) +} + func openURL(url string) error { var cmd *exec.Cmd switch runtime.GOOS { diff --git a/cmd/crabfleet/main_test.go b/cmd/crabfleet/main_test.go index f8f7bd73..328f8424 100644 --- a/cmd/crabfleet/main_test.go +++ b/cmd/crabfleet/main_test.go @@ -3,8 +3,12 @@ package main import ( "bytes" "encoding/json" + "io" "net/http" "net/http/httptest" + "os" + "path/filepath" + "strings" "testing" "github.com/alecthomas/kong" @@ -38,12 +42,129 @@ func TestShellQuoteMatchesGatewaySplitter(t *testing.T) { } } +func TestShellQuoteQuotesMetacharacters(t *testing.T) { + values := []string{ + "$(touch /tmp/pwned)", + ";id", + "&", + "`id`", + ">file", + "*", + "line\nnext", + } + for _, value := range values { + quoted := shellQuote(value) + if !strings.HasPrefix(quoted, "'") || !strings.HasSuffix(quoted, "'") { + t.Fatalf("shellQuote(%q) = %q, want single-quoted", value, quoted) + } + args, err := splitForTest("message IS-1 " + quoted) + if err != nil { + t.Fatal(err) + } + if got := args[2]; got != value { + t.Fatalf("round trip = %q, want %q", got, value) + } + } +} + +func TestMutatingAPIFailureDoesNotFallbackToSSH(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != "/api/ssh/interactive-sessions" { + t.Errorf("request = %s %s", r.Method, r.URL.Path) + w.WriteHeader(http.StatusBadRequest) + return + } + conn, _, err := w.(http.Hijacker).Hijack() + if err != nil { + t.Error(err) + return + } + _ = conn.Close() + })) + defer server.Close() + + app := &cli{API: server.URL, SSHHost: defaultSSHHost, Token: "gateway-token", Fingerprint: "SHA256:test"} + err := (newCmd{Branch: "main", Command: "codex --yolo"}).Run(app, app.apiClient()) + if err == nil { + t.Fatal("expected ambiguous mutation error") + } + if got := err.Error(); !strings.Contains(got, "not retrying through SSH") { + t.Fatalf("error = %q", got) + } +} + +func TestLocalAuthFailureStillFallsBackToSSH(t *testing.T) { + dir := t.TempDir() + argsPath := filepath.Join(dir, "ssh-args") + sshPath := filepath.Join(dir, "ssh") + if err := os.WriteFile(sshPath, []byte("#!/bin/sh\nprintf '%s\\n' \"$@\" > \"$SSH_ARGS_PATH\"\n"), 0o755); err != nil { + t.Fatal(err) + } + t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH")) + t.Setenv("SSH_ARGS_PATH", argsPath) + + app := &cli{API: defaultAPIURL, SSHHost: "crabd.test"} + if err := (newCmd{Branch: "main", Command: "codex --yolo", Repo: "openclaw/crabfleet"}).Run(app, app.apiClient()); err != nil { + t.Fatal(err) + } + data, err := os.ReadFile(argsPath) + if err != nil { + t.Fatal(err) + } + output := string(data) + if !strings.Contains(output, "crabd.test\n") || !strings.Contains(output, "new --branch main --repo openclaw/crabfleet") { + t.Fatalf("ssh args = %q", output) + } +} + +func TestNewCommandSanitizesControlPlaneOutput(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"session":{"id":"IS-1\u001b]52;c;bad\u0007","repo":"openclaw/crabfleet\u001b[31m","status":"ready","vncUrl":"https://example.test/vnc\u001b[0m","ptyAvailable":true}}`)) + })) + defer server.Close() + + app := &cli{API: server.URL, Token: "gateway-token", Fingerprint: "SHA256:test", NoInput: true} + output := captureStdout(t, func() { + if err := (newCmd{Branch: "main", Command: "codex --yolo", Detach: true}).Run(app, app.apiClient()); err != nil { + t.Fatal(err) + } + }) + if strings.ContainsAny(output, "\x1b\x07") { + t.Fatalf("output contains terminal controls: %q", output) + } + if !strings.Contains(output, "session: IS-1]52;c;bad") { + t.Fatalf("output = %q", output) + } +} + func TestFirstLineSkipsBlankLines(t *testing.T) { if got, want := firstLine("\n\n https://example.com/vnc\nignored\n"), "https://example.com/vnc"; got != want { t.Fatalf("firstLine = %q, want %q", got, want) } } +func captureStdout(t *testing.T, fn func()) string { + t.Helper() + previous := os.Stdout + reader, writer, err := os.Pipe() + if err != nil { + t.Fatal(err) + } + os.Stdout = writer + defer func() { + os.Stdout = previous + }() + fn() + _ = writer.Close() + data, err := io.ReadAll(reader) + if err != nil { + t.Fatal(err) + } + _ = reader.Close() + return string(data) +} + func TestNewRuntimeAndProfileOverridesAreOptional(t *testing.T) { t.Setenv("CRABFLEET_ROOT_SESSION_ID", "") parse := func(args ...string) cli { @@ -81,18 +202,26 @@ func TestNewRuntimeAndProfileOverridesAreOptional(t *testing.T) { if bytes.Contains(encoded, []byte(`"profile"`)) { t.Fatalf("omitted profile was serialized: %s", encoded) } - for _, arg := range cmd.sshCreateArgs(req) { + args := cmd.sshCreateArgs(req) + for _, arg := range args { if arg == "--runtime" { t.Fatal("SSH fallback forced a runtime override") } } + cmd = parse("new", "--repo", "openclaw/crabfleet", "--", "--starts-with-dash").New + req = cmd.sessionRequest(&cli{}) + args = cmd.sshCreateArgs(req) + if got, want := args[len(args)-2:], []string{"--", "--starts-with-dash"}; got[0] != want[0] || got[1] != want[1] { + t.Fatalf("prompt separator tail = %q, want %q", got, want) + } + cmd = parse("new", "--runtime", "container").New req = cmd.sessionRequest(&cli{}) if req.Runtime != "container" { t.Fatalf("runtime = %q, want explicit override", req.Runtime) } - args := cmd.sshCreateArgs(req) + args = cmd.sshCreateArgs(req) found := false for index := 0; index+1 < len(args); index++ { if args[index] == "--runtime" && args[index+1] == "container" { diff --git a/internal/fleetapi/client.go b/internal/fleetapi/client.go index 224c72ec..4dba0b8e 100644 --- a/internal/fleetapi/client.go +++ b/internal/fleetapi/client.go @@ -19,6 +19,8 @@ type TerminalSize = terminalws.Size const maxResponseBytes = 4 * 1024 * 1024 const maxErrorBytes = 512 +var ErrMissingAuth = errors.New("API mode requires SSH gateway token + fingerprint or agent token + session ID") + type authMode uint8 const ( @@ -318,7 +320,7 @@ func (a Auth) validate() error { if a.mode == authAgent && a.token != "" && a.principal != "" { return nil } - return errors.New("API mode requires SSH gateway token + fingerprint or agent token + session ID") + return ErrMissingAuth } func sessionPath(id string) string { diff --git a/internal/terminalws/client.go b/internal/terminalws/client.go index df63ffed..cb50a9ef 100644 --- a/internal/terminalws/client.go +++ b/internal/terminalws/client.go @@ -183,8 +183,19 @@ func (c *Client) Attach(ctx context.Context, terminal io.ReadWriter, resizes <-c ctx, cancel := context.WithCancel(ctx) defer cancel() + var wg sync.WaitGroup + closer, closeable := terminal.(io.Closer) + closeTerminal := func() { + cancel() + if closeable { + _ = closer.Close() + } + } + errCh := make(chan error, 3) + wg.Add(1) go func() { + defer wg.Done() buffer := make([]byte, 32*1024) for { count, err := terminal.Read(buffer) @@ -204,7 +215,9 @@ func (c *Client) Attach(ctx context.Context, terminal io.ReadWriter, resizes <-c } } }() + wg.Add(1) go func() { + defer wg.Done() for { select { case <-ctx.Done(): @@ -221,7 +234,9 @@ func (c *Client) Attach(ctx context.Context, terminal io.ReadWriter, resizes <-c } } }() + wg.Add(1) go func() { + defer wg.Done() for { current, err := c.read(ctx) if err != nil { @@ -259,7 +274,10 @@ func (c *Client) Attach(ctx context.Context, terminal io.ReadWriter, resizes <-c }() err := <-errCh - cancel() + closeTerminal() + if closeable { + wg.Wait() + } return normalizeCloseError(err) } diff --git a/internal/terminalws/client_test.go b/internal/terminalws/client_test.go index 96211558..5633865d 100644 --- a/internal/terminalws/client_test.go +++ b/internal/terminalws/client_test.go @@ -10,7 +10,9 @@ import ( "net/http" "net/http/httptest" "os" + "sync" "testing" + "time" "github.com/coder/websocket" ) @@ -21,11 +23,11 @@ type protocolFixture struct { Messages map[string]byte `json:"messages"` SubscribeFlags map[string]uint32 `json:"subscribeFlags"` Vectors struct { - OutputFrame string `json:"outputFrame"` - PingFrame string `json:"pingFrame"` - Subscribe string `json:"subscribe"` - Resize string `json:"resize"` - Ack string `json:"ack"` + OutputFrame string `json:"outputFrame"` + PingFrame string `json:"pingFrame"` + Subscribe string `json:"subscribe"` + Resize string `json:"resize"` + Ack string `json:"ack"` } `json:"vectors"` } @@ -249,7 +251,7 @@ func TestClientSubscribesSendsInputAndAcknowledgesOutput(t *testing.T) { inputReader, inputWriter := io.Pipe() defer inputReader.Close() defer inputWriter.Close() - terminal := &readWriter{reader: inputReader} + terminal := &readWriter{reader: inputReader, closer: inputReader} go func() { _, _ = inputWriter.Write([]byte("hello\n")) }() @@ -272,11 +274,95 @@ func TestClientSubscribesSendsInputAndAcknowledgesOutput(t *testing.T) { } } +func TestAttachClosesCloseableTerminalAfterRemoteClosure(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Accept(w, r, nil) + if err != nil { + t.Error(err) + return + } + defer conn.Close(websocket.StatusNormalClosure, "") + + for range 2 { + if _, _, err := conn.Read(r.Context()); err != nil { + t.Error(err) + return + } + } + subscribed, _ := json.Marshal(eventPayload{Type: "subscribed", CanInput: true}) + if err := conn.Write(r.Context(), websocket.MessageBinary, encodeFrame(frame{ + messageType: messageEvent, + sessionID: "IS-1", + payload: subscribed, + })); err != nil { + t.Error(err) + return + } + closed, _ := json.Marshal(eventPayload{Type: "closed"}) + _ = conn.Write(r.Context(), websocket.MessageBinary, encodeFrame(frame{ + messageType: messageEvent, + sessionID: "IS-1", + payload: closed, + })) + })) + defer server.Close() + + endpoint, err := Endpoint(server.URL) + if err != nil { + t.Fatal(err) + } + client, err := Dial(context.Background(), endpoint, "IS-1", Options{Cols: 120, Rows: 34}) + if err != nil { + t.Fatal(err) + } + defer client.Close() + + terminal := newBlockingTerminal() + if err := client.Attach(context.Background(), terminal, nil); err != nil { + t.Fatal(err) + } + select { + case <-terminal.closed: + case <-time.After(time.Second): + t.Fatal("terminal was not closed") + } +} + type readWriter struct { reader io.Reader + closer io.Closer bytes.Buffer } func (rw *readWriter) Read(payload []byte) (int, error) { return rw.reader.Read(payload) } + +func (rw *readWriter) Close() error { + if rw.closer == nil { + return nil + } + return rw.closer.Close() +} + +type blockingTerminal struct { + closed chan struct{} + once sync.Once + bytes.Buffer +} + +func newBlockingTerminal() *blockingTerminal { + return &blockingTerminal{closed: make(chan struct{})} +} + +func (terminal *blockingTerminal) Read(_ []byte) (int, error) { + <-terminal.closed + return 0, io.ErrClosedPipe +} + +func (terminal *blockingTerminal) Close() error { + terminal.once.Do(func() { + close(terminal.closed) + }) + return nil +} From 7dcec13ed678850f3165f6141a7f922021ff9aac Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Thu, 18 Jun 2026 17:16:51 +0800 Subject: [PATCH 02/14] chore(format): preserve oxfmt spacing --- .oxfmtrc.json | 4 ++++ 1 file changed, 4 insertions(+) create mode 100644 .oxfmtrc.json diff --git a/.oxfmtrc.json b/.oxfmtrc.json new file mode 100644 index 00000000..8973894d --- /dev/null +++ b/.oxfmtrc.json @@ -0,0 +1,4 @@ +{ + "useTabs": false, + "tabWidth": 2 +} From 5bdec8066b29b2dee1ce93058cc9a39f67af8501 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Thu, 18 Jun 2026 17:19:00 +0800 Subject: [PATCH 03/14] fix(cli): preserve auth fallback --- cmd/crabfleet/main.go | 12 ++++++++- cmd/crabfleet/main_test.go | 50 +++++++++++++++++++++++++++++++------ internal/fleetapi/client.go | 16 +++++++++++- 3 files changed, 68 insertions(+), 10 deletions(-) diff --git a/cmd/crabfleet/main.go b/cmd/crabfleet/main.go index 0b429390..4946b716 100644 --- a/cmd/crabfleet/main.go +++ b/cmd/crabfleet/main.go @@ -652,7 +652,17 @@ func ambiguousMutationError(operation string, err error) error { } func canFallbackToSSH(app *cli, err error) bool { - return !app.NoInput && !app.JSON && errors.Is(err, fleetapi.ErrMissingAuth) + if app.NoInput || app.JSON { + return false + } + if errors.Is(err, fleetapi.ErrMissingAuth) { + return true + } + var statusErr *fleetapi.StatusError + if errors.As(err, &statusErr) { + return statusErr.StatusCode == http.StatusUnauthorized || statusErr.StatusCode == http.StatusForbidden + } + return false } func openURL(url string) error { diff --git a/cmd/crabfleet/main_test.go b/cmd/crabfleet/main_test.go index 328f8424..e882bc4f 100644 --- a/cmd/crabfleet/main_test.go +++ b/cmd/crabfleet/main_test.go @@ -94,6 +94,43 @@ func TestMutatingAPIFailureDoesNotFallbackToSSH(t *testing.T) { } func TestLocalAuthFailureStillFallsBackToSSH(t *testing.T) { + argsPath := installFakeSSH(t) + + app := &cli{API: defaultAPIURL, SSHHost: "crabd.test"} + if err := (newCmd{Branch: "main", Command: "codex --yolo", Repo: "openclaw/crabfleet"}).Run(app, app.apiClient()); err != nil { + t.Fatal(err) + } + output := readFakeSSHArgs(t, argsPath) + if !strings.Contains(output, "crabd.test\n") || !strings.Contains(output, "new --branch main --repo openclaw/crabfleet") { + t.Fatalf("ssh args = %q", output) + } +} + +func TestAPIAuthRejectionStillFallsBackToSSH(t *testing.T) { + argsPath := installFakeSSH(t) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusUnauthorized) + _, _ = w.Write([]byte("bad token")) + })) + defer server.Close() + + app := &cli{ + API: server.URL, + SSHHost: "crabd.test", + Token: "stale-token", + Fingerprint: "SHA256:stale", + } + if err := (newCmd{Branch: "main", Command: "codex --yolo", Repo: "openclaw/crabfleet"}).Run(app, app.apiClient()); err != nil { + t.Fatal(err) + } + output := readFakeSSHArgs(t, argsPath) + if !strings.Contains(output, "crabd.test\n") || !strings.Contains(output, "new --branch main --repo openclaw/crabfleet") { + t.Fatalf("ssh args = %q", output) + } +} + +func installFakeSSH(t *testing.T) string { + t.Helper() dir := t.TempDir() argsPath := filepath.Join(dir, "ssh-args") sshPath := filepath.Join(dir, "ssh") @@ -102,19 +139,16 @@ func TestLocalAuthFailureStillFallsBackToSSH(t *testing.T) { } t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH")) t.Setenv("SSH_ARGS_PATH", argsPath) + return argsPath +} - app := &cli{API: defaultAPIURL, SSHHost: "crabd.test"} - if err := (newCmd{Branch: "main", Command: "codex --yolo", Repo: "openclaw/crabfleet"}).Run(app, app.apiClient()); err != nil { - t.Fatal(err) - } +func readFakeSSHArgs(t *testing.T, argsPath string) string { + t.Helper() data, err := os.ReadFile(argsPath) if err != nil { t.Fatal(err) } - output := string(data) - if !strings.Contains(output, "crabd.test\n") || !strings.Contains(output, "new --branch main --repo openclaw/crabfleet") { - t.Fatalf("ssh args = %q", output) - } + return string(data) } func TestNewCommandSanitizesControlPlaneOutput(t *testing.T) { diff --git a/internal/fleetapi/client.go b/internal/fleetapi/client.go index 4dba0b8e..9324d882 100644 --- a/internal/fleetapi/client.go +++ b/internal/fleetapi/client.go @@ -21,6 +21,16 @@ const maxErrorBytes = 512 var ErrMissingAuth = errors.New("API mode requires SSH gateway token + fingerprint or agent token + session ID") +type StatusError struct { + StatusCode int + Status string + Body string +} + +func (e *StatusError) Error() string { + return fmt.Sprintf("crabfleet API %s: %s", e.Status, e.Body) +} + type authMode uint8 const ( @@ -276,7 +286,11 @@ func (c *Client) open( func responseError(resp *http.Response) error { if resp.StatusCode < 200 || resp.StatusCode >= 300 { data, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrorBytes)) - return fmt.Errorf("crabfleet API %s: %s", resp.Status, strings.TrimSpace(string(data))) + return &StatusError{ + StatusCode: resp.StatusCode, + Status: resp.Status, + Body: strings.TrimSpace(string(data)), + } } return nil } From 4db6b658b3b70607b527d9c7bfa1568fe61cc392 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Thu, 18 Jun 2026 17:22:06 +0800 Subject: [PATCH 04/14] fix(cli): map terminal auth failures --- cmd/crabfleet/main_test.go | 23 +++++++++++++++++++++++ internal/fleetapi/client.go | 14 +++++++++++++- internal/terminalws/client.go | 26 +++++++++++++++++++++++++- 3 files changed, 61 insertions(+), 2 deletions(-) diff --git a/cmd/crabfleet/main_test.go b/cmd/crabfleet/main_test.go index e882bc4f..e74da883 100644 --- a/cmd/crabfleet/main_test.go +++ b/cmd/crabfleet/main_test.go @@ -129,6 +129,29 @@ func TestAPIAuthRejectionStillFallsBackToSSH(t *testing.T) { } } +func TestMessageWebSocketAuthRejectionStillFallsBackToSSH(t *testing.T) { + argsPath := installFakeSSH(t) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusForbidden) + _, _ = w.Write([]byte("bad websocket token")) + })) + defer server.Close() + + app := &cli{ + API: server.URL, + SSHHost: "crabd.test", + Token: "stale-token", + Fingerprint: "SHA256:stale", + } + if err := (messageCmd{ID: "IS-1", Text: []string{"hello"}}).Run(app, app.apiClient()); err != nil { + t.Fatal(err) + } + output := readFakeSSHArgs(t, argsPath) + if !strings.Contains(output, "crabd.test\n") || !strings.Contains(output, "message IS-1 hello") { + t.Fatalf("ssh args = %q", output) + } +} + func installFakeSSH(t *testing.T) string { t.Helper() dir := t.TempDir() diff --git a/internal/fleetapi/client.go b/internal/fleetapi/client.go index 9324d882..b2ec8c64 100644 --- a/internal/fleetapi/client.go +++ b/internal/fleetapi/client.go @@ -224,11 +224,23 @@ func (c *Client) terminal(ctx context.Context, id string, cols uint32, rows uint if err != nil { return nil, err } - return terminalws.Dial(ctx, endpoint, id, terminalws.Options{ + client, err := terminalws.Dial(ctx, endpoint, id, terminalws.Options{ Header: headers, Cols: cols, Rows: rows, }) + if err != nil { + var statusErr *terminalws.HandshakeStatusError + if errors.As(err, &statusErr) { + return nil, &StatusError{ + StatusCode: statusErr.StatusCode, + Status: statusErr.Status, + Body: statusErr.Body, + } + } + return nil, err + } + return client, nil } func (c *Client) doJSON(ctx context.Context, method string, path string, body any, out any) error { diff --git a/internal/terminalws/client.go b/internal/terminalws/client.go index cb50a9ef..1531bc6f 100644 --- a/internal/terminalws/client.go +++ b/internal/terminalws/client.go @@ -9,6 +9,7 @@ import ( "io" "net/http" "net/url" + "strings" "sync" "github.com/coder/websocket" @@ -18,6 +19,7 @@ const ( magic = 0x5943 version = 2 maxFrameBytes = 16 * 1024 * 1024 + maxErrorBytes = 512 messageHello = 1 messageWelcome = 2 @@ -51,6 +53,19 @@ type Options struct { Rows uint32 } +type HandshakeStatusError struct { + StatusCode int + Status string + Body string +} + +func (e *HandshakeStatusError) Error() string { + if e.Body == "" { + return fmt.Sprintf("terminal websocket %s", e.Status) + } + return fmt.Sprintf("terminal websocket %s: %s", e.Status, e.Body) +} + type Size struct { Cols uint32 Rows uint32 @@ -100,10 +115,19 @@ func Dial(ctx context.Context, endpoint string, sessionID string, options Option if sessionID == "" { return nil, errors.New("terminal session id is required") } - conn, _, err := websocket.Dial(ctx, endpoint, &websocket.DialOptions{ + conn, resp, err := websocket.Dial(ctx, endpoint, &websocket.DialOptions{ HTTPHeader: options.Header, }) if err != nil { + if resp != nil { + body := "" + if resp.Body != nil { + data, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrorBytes)) + _ = resp.Body.Close() + body = strings.TrimSpace(string(data)) + } + return nil, &HandshakeStatusError{StatusCode: resp.StatusCode, Status: resp.Status, Body: body} + } return nil, err } conn.SetReadLimit(maxFrameBytes) From c93f2a52ace95c42d427bb3571cb226a479d0366 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Thu, 18 Jun 2026 17:24:17 +0800 Subject: [PATCH 05/14] fix(gateway): preserve text flag parsing --- cmd/crabbox-ssh-gateway/main.go | 25 +++++++++++++++++++++-- cmd/crabbox-ssh-gateway/main_test.go | 30 ++++++++++++++++++++++++++++ 2 files changed, 53 insertions(+), 2 deletions(-) diff --git a/cmd/crabbox-ssh-gateway/main.go b/cmd/crabbox-ssh-gateway/main.go index 1b4f5b42..6afe2d3d 100644 --- a/cmd/crabbox-ssh-gateway/main.go +++ b/cmd/crabbox-ssh-gateway/main.go @@ -14,6 +14,7 @@ import ( "net" "net/http" "os" + "strconv" "strings" "time" @@ -636,9 +637,21 @@ type messageInput struct { func parseMessage(args []string) (messageInput, error) { var input messageInput remaining := args - if len(remaining) > 0 && remaining[0] == "--no-enter" { + if len(remaining) > 0 && (remaining[0] == "--no-enter" || remaining[0] == "-no-enter") { input.noEnter = true remaining = remaining[1:] + } else if len(remaining) > 0 { + for _, prefix := range []string{"--no-enter=", "-no-enter="} { + if value, ok := strings.CutPrefix(remaining[0], prefix); ok { + parsed, err := strconv.ParseBool(value) + if err != nil { + return messageInput{}, err + } + input.noEnter = parsed + remaining = remaining[1:] + break + } + } } if len(remaining) > 0 && remaining[0] == "--" { remaining = remaining[1:] @@ -650,12 +663,20 @@ func parseMessage(args []string) (messageInput, error) { func parseSummary(args []string) (summaryUpdate, error) { var update summaryUpdate remaining := args - if len(remaining) > 0 && remaining[0] == "--purpose" { + if len(remaining) > 0 && (remaining[0] == "--purpose" || remaining[0] == "-purpose") { if len(remaining) < 2 { return summaryUpdate{}, errors.New("flag needs an argument: -purpose") } update.purpose = remaining[1] remaining = remaining[2:] + } else if len(remaining) > 0 { + for _, prefix := range []string{"--purpose=", "-purpose="} { + if value, ok := strings.CutPrefix(remaining[0], prefix); ok { + update.purpose = value + remaining = remaining[1:] + break + } + } } if len(remaining) > 0 && remaining[0] == "--" { remaining = remaining[1:] diff --git a/cmd/crabbox-ssh-gateway/main_test.go b/cmd/crabbox-ssh-gateway/main_test.go index b24ef6de..7921ecda 100644 --- a/cmd/crabbox-ssh-gateway/main_test.go +++ b/cmd/crabbox-ssh-gateway/main_test.go @@ -116,6 +116,14 @@ func TestParseMessageKeepsNoEnterAndText(t *testing.T) { if got, want := message.text, "--help is terminal text"; got != want { t.Fatalf("text = %q, want %q", got, want) } + + message, err = parseMessage([]string{"--no-enter=true", "hello"}) + if err != nil { + t.Fatal(err) + } + if !message.noEnter || message.text != "hello" { + t.Fatalf("message = %#v", message) + } } func TestParseSummaryKeepsDashPrefixedText(t *testing.T) { @@ -137,6 +145,28 @@ func TestParseSummaryKeepsDashPrefixedText(t *testing.T) { if got, want := summary.summary, "--summary-start"; got != want { t.Fatalf("summary = %q, want %q", got, want) } + + summary, err = parseSummary([]string{"--purpose=handoff", "done"}) + if err != nil { + t.Fatal(err) + } + if got, want := summary.purpose, "handoff"; got != want { + t.Fatalf("purpose = %q, want %q", got, want) + } + if got, want := summary.summary, "done"; got != want { + t.Fatalf("summary = %q, want %q", got, want) + } + + summary, err = parseSummary([]string{"-purpose", "handoff", "done"}) + if err != nil { + t.Fatal(err) + } + if got, want := summary.purpose, "handoff"; got != want { + t.Fatalf("purpose = %q, want %q", got, want) + } + if got, want := summary.summary, "done"; got != want { + t.Fatalf("summary = %q, want %q", got, want) + } } func TestParseCreateLeavesRuntimeToDeploymentDefault(t *testing.T) { From d20f55e8d332ef0df648fab2c22328e72b22dd5e Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Thu, 18 Jun 2026 17:26:32 +0800 Subject: [PATCH 06/14] fix(terminal): avoid closing owned channels --- internal/terminalws/client.go | 16 ++++++++++------ internal/terminalws/client_test.go | 4 ++-- 2 files changed, 12 insertions(+), 8 deletions(-) diff --git a/internal/terminalws/client.go b/internal/terminalws/client.go index 1531bc6f..8c7b443a 100644 --- a/internal/terminalws/client.go +++ b/internal/terminalws/client.go @@ -90,6 +90,10 @@ type eventPayload struct { CanInput bool `json:"canInput"` } +type readCanceler interface { + CancelRead() error +} + func Endpoint(baseURL string) (string, error) { target, err := url.Parse(baseURL) if err != nil { @@ -208,11 +212,11 @@ func (c *Client) Attach(ctx context.Context, terminal io.ReadWriter, resizes <-c defer cancel() var wg sync.WaitGroup - closer, closeable := terminal.(io.Closer) - closeTerminal := func() { + canceler, cancelableRead := terminal.(readCanceler) + cancelRead := func() { cancel() - if closeable { - _ = closer.Close() + if cancelableRead { + _ = canceler.CancelRead() } } @@ -298,8 +302,8 @@ func (c *Client) Attach(ctx context.Context, terminal io.ReadWriter, resizes <-c }() err := <-errCh - closeTerminal() - if closeable { + cancelRead() + if cancelableRead { wg.Wait() } return normalizeCloseError(err) diff --git a/internal/terminalws/client_test.go b/internal/terminalws/client_test.go index 5633865d..db79db3a 100644 --- a/internal/terminalws/client_test.go +++ b/internal/terminalws/client_test.go @@ -338,7 +338,7 @@ func (rw *readWriter) Read(payload []byte) (int, error) { return rw.reader.Read(payload) } -func (rw *readWriter) Close() error { +func (rw *readWriter) CancelRead() error { if rw.closer == nil { return nil } @@ -360,7 +360,7 @@ func (terminal *blockingTerminal) Read(_ []byte) (int, error) { return 0, io.ErrClosedPipe } -func (terminal *blockingTerminal) Close() error { +func (terminal *blockingTerminal) CancelRead() error { terminal.once.Do(func() { close(terminal.closed) }) From b593e0ee047be9756ce2ad2904505f0b38f62a00 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Thu, 18 Jun 2026 17:32:32 +0800 Subject: [PATCH 07/14] fix(gateway): bound ssh control work --- cmd/crabbox-ssh-gateway/main.go | 72 ++++++++++++++++++++++- cmd/crabbox-ssh-gateway/main_test.go | 85 ++++++++++++++++++++++++++++ internal/fleetapi/client.go | 5 +- internal/fleetapi/client_test.go | 20 +++++++ 4 files changed, 178 insertions(+), 4 deletions(-) diff --git a/cmd/crabbox-ssh-gateway/main.go b/cmd/crabbox-ssh-gateway/main.go index 6afe2d3d..ee0d7df3 100644 --- a/cmd/crabbox-ssh-gateway/main.go +++ b/cmd/crabbox-ssh-gateway/main.go @@ -26,8 +26,42 @@ import ( var ( sshHandshakeTimeout = 15 * time.Second sshAuthTimeout = 15 * time.Second + sshHandshakeSlots = newConnectionLimiter(64) ) +type connectionLimiter struct { + slots chan struct{} +} + +func newConnectionLimiter(limit int) *connectionLimiter { + if limit < 1 { + limit = 1 + } + return &connectionLimiter{slots: make(chan struct{}, limit)} +} + +func (l *connectionLimiter) acquire() bool { + if l == nil { + return true + } + select { + case l.slots <- struct{}{}: + return true + default: + return false + } +} + +func (l *connectionLimiter) release() { + if l == nil { + return + } + select { + case <-l.slots: + default: + } +} + type apiClient struct { baseURL string token string @@ -133,16 +167,44 @@ func main() { log.Printf("accept: %v", err) continue } - go handleConn(conn, config, client) + acceptConn(conn, config, client) + } +} + +func acceptConn(raw net.Conn, config *ssh.ServerConfig, client *apiClient) { + if !sshHandshakeSlots.acquire() { + log.Printf("connection limit reached for %s", raw.RemoteAddr()) + raw.Close() + return } + go handleConnWithRelease(raw, config, client, sshHandshakeSlots.release) } func handleConn(raw net.Conn, config *ssh.ServerConfig, client *apiClient) { + handleConnWithRelease(raw, config, client, nil) +} + +func handleConnWithRelease( + raw net.Conn, + config *ssh.ServerConfig, + client *apiClient, + release func(), +) { + releaseHandshake := release + defer func() { + if releaseHandshake != nil { + releaseHandshake() + } + }() defer raw.Close() if err := raw.SetDeadline(time.Now().Add(sshHandshakeTimeout)); err != nil { log.Printf("handshake deadline %s: %v", raw.RemoteAddr(), err) } conn, chans, reqs, err := ssh.NewServerConn(raw, config) + if releaseHandshake != nil { + releaseHandshake() + releaseHandshake = nil + } if err != nil { log.Printf("handshake %s: %v", raw.RemoteAddr(), err) return @@ -169,6 +231,8 @@ func handleConn(raw net.Conn, config *ssh.ServerConfig, client *apiClient) { func handleSession(channel ssh.Channel, requests <-chan *ssh.Request, perms *ssh.Permissions, client *apiClient) { defer channel.Close() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() pty := sessionPTY{ cols: 120, rows: 34, @@ -180,6 +244,7 @@ func handleSession(channel ssh.Channel, requests <-chan *ssh.Request, perms *ssh select { case req, ok := <-requests: if !ok { + cancel() return } switch req.Type { @@ -212,7 +277,7 @@ func handleSession(channel ssh.Channel, requests <-chan *ssh.Request, perms *ssh commandStarted = true req.Reply(true, nil) go func(current sessionPTY) { - exitCh <- runCommand(context.Background(), channel, perms, client, "", current) + exitCh <- runCommand(ctx, channel, perms, client, "", current) }(pty) case "exec": if commandStarted { @@ -225,7 +290,7 @@ func handleSession(channel ssh.Channel, requests <-chan *ssh.Request, perms *ssh req.Reply(true, nil) go func(current sessionPTY, command string) { exitCh <- runCommand( - context.Background(), + ctx, channel, perms, client, @@ -237,6 +302,7 @@ func handleSession(channel ssh.Channel, requests <-chan *ssh.Request, perms *ssh req.Reply(false, nil) } case exit := <-exitCh: + cancel() replyExit(channel, exit) return } diff --git a/cmd/crabbox-ssh-gateway/main_test.go b/cmd/crabbox-ssh-gateway/main_test.go index 7921ecda..d903dda9 100644 --- a/cmd/crabbox-ssh-gateway/main_test.go +++ b/cmd/crabbox-ssh-gateway/main_test.go @@ -324,6 +324,91 @@ func TestHandleConnClosesStalledHandshake(t *testing.T) { } } +func TestAcceptConnRejectsWhenHandshakeSlotsFull(t *testing.T) { + previous := sshHandshakeSlots + sshHandshakeSlots = newConnectionLimiter(1) + if !sshHandshakeSlots.acquire() { + t.Fatal("failed to occupy handshake slot") + } + defer func() { + sshHandshakeSlots.release() + sshHandshakeSlots = previous + }() + + server, client := net.Pipe() + defer client.Close() + acceptConn(server, &ssh.ServerConfig{}, nil) + + done := make(chan error, 1) + go func() { + var buf [1]byte + _, err := client.Read(buf[:]) + done <- err + }() + select { + case err := <-done: + if err == nil { + t.Fatal("connection remained open after handshake slot exhaustion") + } + case <-time.After(time.Second): + t.Fatal("connection remained open after handshake slot exhaustion") + } +} + +func TestRunCommandCancelsControlPlaneRequest(t *testing.T) { + entered := make(chan struct{}) + cancelled := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/ssh/state" { + t.Errorf("path = %q", r.URL.Path) + w.WriteHeader(http.StatusNotFound) + return + } + close(entered) + select { + case <-r.Context().Done(): + close(cancelled) + case <-time.After(time.Second): + t.Error("request context was not cancelled") + } + })) + defer server.Close() + + client := &apiClient{baseURL: server.URL, token: "gateway-token", client: server.Client()} + permissions := &ssh.Permissions{Extensions: map[string]string{ + "authorized": "true", + "fingerprint": "SHA256:test", + "login": "operator", + "role": "owner", + }} + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan uint32, 1) + var output bytes.Buffer + go func() { + done <- runCommand(ctx, &output, permissions, client, "whoami", sessionPTY{}) + }() + + select { + case <-entered: + case <-time.After(time.Second): + t.Fatal("control-plane request was not started") + } + cancel() + select { + case <-cancelled: + case <-time.After(time.Second): + t.Fatal("control-plane request was not cancelled") + } + select { + case exit := <-done: + if exit != 1 { + t.Fatalf("exit = %d, want 1", exit) + } + case <-time.After(time.Second): + t.Fatalf("runCommand did not return after cancellation; output=%q", output.String()) + } +} + func TestHelpNamesDeleteAsCanonicalCommand(t *testing.T) { var output bytes.Buffer printHelp(&output, fleetapi.User{Login: "operator", Role: "owner"}) diff --git a/internal/fleetapi/client.go b/internal/fleetapi/client.go index b2ec8c64..b0a92f31 100644 --- a/internal/fleetapi/client.go +++ b/internal/fleetapi/client.go @@ -165,7 +165,10 @@ func (c *Client) Transcript(ctx context.Context, id string) (string, error) { if err := responseError(resp); err != nil { return "", err } - data, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes)) + data, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes+1)) + if len(data) > maxResponseBytes { + return "", fmt.Errorf("crabfleet API response exceeds %d bytes", maxResponseBytes) + } return string(data), err } diff --git a/internal/fleetapi/client_test.go b/internal/fleetapi/client_test.go index 71289209..714fd23a 100644 --- a/internal/fleetapi/client_test.go +++ b/internal/fleetapi/client_test.go @@ -84,3 +84,23 @@ func TestClientStreamsLargeJSONResponses(t *testing.T) { t.Fatalf("login length = %d, want %d", len(state.User.Login), len(largeLogin)) } } + +func TestClientRejectsOversizedTranscript(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/ssh/interactive-sessions/IS-7/transcript" { + t.Errorf("path = %q", r.URL.Path) + } + w.Header().Set("Content-Type", "text/markdown") + _, _ = w.Write([]byte(strings.Repeat("a", maxResponseBytes+1))) + })) + defer server.Close() + + client := NewClient(server.URL, server.Client(), SSHAuth("gateway-token", "SHA256:test")) + transcript, err := client.Transcript(context.Background(), "IS-7") + if err == nil || !strings.Contains(err.Error(), "response exceeds") { + t.Fatalf("error = %v", err) + } + if transcript != "" { + t.Fatalf("transcript length = %d, want empty", len(transcript)) + } +} From 646d5d13bd744137f0369a3577e32bd46b28da26 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Thu, 18 Jun 2026 17:34:37 +0800 Subject: [PATCH 08/14] fix(cli): preserve safe transport fallback --- cmd/crabfleet/main.go | 39 ++++++++++++++++++++++++++++++++++++++ cmd/crabfleet/main_test.go | 27 ++++++++++++++++++++++++++ 2 files changed, 66 insertions(+) diff --git a/cmd/crabfleet/main.go b/cmd/crabfleet/main.go index 4946b716..8bc3febb 100644 --- a/cmd/crabfleet/main.go +++ b/cmd/crabfleet/main.go @@ -2,11 +2,14 @@ package main import ( "context" + "crypto/x509" "encoding/json" "errors" "fmt" "io" + "net" "net/http" + "net/url" "os" "os/exec" "runtime" @@ -662,9 +665,45 @@ func canFallbackToSSH(app *cli, err error) bool { if errors.As(err, &statusErr) { return statusErr.StatusCode == http.StatusUnauthorized || statusErr.StatusCode == http.StatusForbidden } + if isPreRequestNetworkFailure(err) { + return true + } return false } +func isPreRequestNetworkFailure(err error) bool { + var urlErr *url.Error + if !errors.As(err, &urlErr) { + return false + } + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return false + } + var dnsErr *net.DNSError + if errors.As(err, &dnsErr) { + return true + } + var opErr *net.OpError + if errors.As(err, &opErr) && opErr.Op == "dial" { + return true + } + var unknownAuthority x509.UnknownAuthorityError + if errors.As(err, &unknownAuthority) { + return true + } + var hostnameErr x509.HostnameError + if errors.As(err, &hostnameErr) { + return true + } + var invalidCert x509.CertificateInvalidError + if errors.As(err, &invalidCert) { + return true + } + lower := strings.ToLower(err.Error()) + return strings.Contains(lower, "tls:") || + strings.Contains(lower, "server gave http response to https client") +} + func openURL(url string) error { var cmd *exec.Cmd switch runtime.GOOS { diff --git a/cmd/crabfleet/main_test.go b/cmd/crabfleet/main_test.go index e74da883..b4173c97 100644 --- a/cmd/crabfleet/main_test.go +++ b/cmd/crabfleet/main_test.go @@ -4,6 +4,7 @@ import ( "bytes" "encoding/json" "io" + "net" "net/http" "net/http/httptest" "os" @@ -93,6 +94,32 @@ func TestMutatingAPIFailureDoesNotFallbackToSSH(t *testing.T) { } } +func TestPreRequestAPIFailureStillFallsBackToSSH(t *testing.T) { + argsPath := installFakeSSH(t) + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + apiURL := "http://" + listener.Addr().String() + if err := listener.Close(); err != nil { + t.Fatal(err) + } + + app := &cli{ + API: apiURL, + SSHHost: "crabd.test", + Token: "gateway-token", + Fingerprint: "SHA256:test", + } + if err := (newCmd{Branch: "main", Command: "codex --yolo", Repo: "openclaw/crabfleet"}).Run(app, app.apiClient()); err != nil { + t.Fatal(err) + } + output := readFakeSSHArgs(t, argsPath) + if !strings.Contains(output, "crabd.test\n") || !strings.Contains(output, "new --branch main --repo openclaw/crabfleet") { + t.Fatalf("ssh args = %q", output) + } +} + func TestLocalAuthFailureStillFallsBackToSSH(t *testing.T) { argsPath := installFakeSSH(t) From 095af3631a00da1a8e905110381263d263224e71 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Thu, 18 Jun 2026 17:44:13 +0800 Subject: [PATCH 09/14] fix(cli): sanitize transcript output --- cmd/crabbox-ssh-gateway/main.go | 15 +++-- cmd/crabbox-ssh-gateway/main_test.go | 90 +++++++++++++++++++++++++++- cmd/crabfleet/main.go | 5 +- cmd/crabfleet/main_test.go | 25 ++++++++ internal/fleettext/text.go | 66 +++++++++++++++++++- internal/fleettext/text_test.go | 12 ++++ 6 files changed, 204 insertions(+), 9 deletions(-) diff --git a/cmd/crabbox-ssh-gateway/main.go b/cmd/crabbox-ssh-gateway/main.go index ee0d7df3..912e530f 100644 --- a/cmd/crabbox-ssh-gateway/main.go +++ b/cmd/crabbox-ssh-gateway/main.go @@ -384,7 +384,7 @@ func runCommand(ctx context.Context, out io.ReadWriter, perms *ssh.Permissions, printList(out, state) return 0 case "new": - create, err := parseCreate(args[1:], api) + create, err := parseCreate(ctx, args[1:], api) if err != nil { fmt.Fprintf(out, "usage: new [--repo owner/repo] [--branch main] [--runtime crabbox|container] [--profile name] [prompt]\nerror: %v\n", err) return 2 @@ -477,8 +477,9 @@ func runCommand(ctx context.Context, out io.ReadWriter, perms *ssh.Permissions, fmt.Fprintf(out, "error: %v\n", err) return 1 } - fmt.Fprint(out, transcript) - if !strings.HasSuffix(transcript, "\n") { + safeTranscript := fleettext.SafeMultiline(transcript) + fmt.Fprint(out, safeTranscript) + if !strings.HasSuffix(safeTranscript, "\n") { fmt.Fprintln(out) } return 0 @@ -661,7 +662,7 @@ func splitCommand(command string) ([]string, error) { return args, nil } -func parseCreate(args []string, api *fleetapi.Client) (createArgs, error) { +func parseCreate(ctx context.Context, args []string, api *fleetapi.Client) (createArgs, error) { fs := flag.NewFlagSet("new", flag.ContinueOnError) fs.SetOutput(io.Discard) var req fleetapi.CreateSessionRequest @@ -683,7 +684,11 @@ func parseCreate(args []string, api *fleetapi.Client) (createArgs, error) { } req.Prompt = strings.Join(fs.Args(), " ") if req.Repo == "" && api != nil { - if state, err := api.State(context.Background()); err == nil && len(state.Repos) > 0 { + state, err := api.State(ctx) + if err != nil && (ctx.Err() != nil || errors.Is(err, context.Canceled)) { + return createArgs{}, err + } + if err == nil && len(state.Repos) > 0 { req.Repo = state.Repos[0] } } diff --git a/cmd/crabbox-ssh-gateway/main_test.go b/cmd/crabbox-ssh-gateway/main_test.go index d903dda9..e9733283 100644 --- a/cmd/crabbox-ssh-gateway/main_test.go +++ b/cmd/crabbox-ssh-gateway/main_test.go @@ -67,6 +67,7 @@ func TestSplitCommandPreservesBackslashesInSingleQuotes(t *testing.T) { func TestParseCreateKeepsLineageAndSummaryFlags(t *testing.T) { create, err := parseCreate( + context.Background(), []string{ "--repo", "openclaw/crabfleet", "--parent", "IS-1", @@ -170,7 +171,7 @@ func TestParseSummaryKeepsDashPrefixedText(t *testing.T) { } func TestParseCreateLeavesRuntimeToDeploymentDefault(t *testing.T) { - create, err := parseCreate([]string{"--repo", "openclaw/crabfleet", "fix it"}, nil) + create, err := parseCreate(context.Background(), []string{"--repo", "openclaw/crabfleet", "fix it"}, nil) if err != nil { t.Fatal(err) } @@ -179,6 +180,7 @@ func TestParseCreateLeavesRuntimeToDeploymentDefault(t *testing.T) { } create, err = parseCreate( + context.Background(), []string{"--repo", "openclaw/crabfleet", "--runtime", "container", "fix it"}, nil, ) @@ -192,6 +194,7 @@ func TestParseCreateLeavesRuntimeToDeploymentDefault(t *testing.T) { func TestParseCreateAcceptsProfileOverride(t *testing.T) { create, err := parseCreate( + context.Background(), []string{"--repo", "openclaw/crabfleet", "--profile", "desktop-a", "fix it"}, nil, ) @@ -409,6 +412,91 @@ func TestRunCommandCancelsControlPlaneRequest(t *testing.T) { } } +func TestRunCommandCancelsDefaultRepoLookup(t *testing.T) { + entered := make(chan struct{}) + cancelled := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/ssh/state" { + t.Errorf("path = %q", r.URL.Path) + w.WriteHeader(http.StatusNotFound) + return + } + close(entered) + select { + case <-r.Context().Done(): + close(cancelled) + case <-time.After(time.Second): + t.Error("request context was not cancelled") + } + })) + defer server.Close() + + client := &apiClient{baseURL: server.URL, token: "gateway-token", client: server.Client()} + permissions := &ssh.Permissions{Extensions: map[string]string{ + "authorized": "true", + "fingerprint": "SHA256:test", + "login": "operator", + "role": "owner", + }} + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan uint32, 1) + var output bytes.Buffer + go func() { + done <- runCommand(ctx, &output, permissions, client, "new fix it", sessionPTY{}) + }() + + select { + case <-entered: + case <-time.After(time.Second): + t.Fatal("default repo lookup was not started") + } + cancel() + select { + case <-cancelled: + case <-time.After(time.Second): + t.Fatal("default repo lookup was not cancelled") + } + select { + case exit := <-done: + if exit != 2 { + t.Fatalf("exit = %d, want 2", exit) + } + case <-time.After(time.Second): + t.Fatalf("runCommand did not return after cancellation; output=%q", output.String()) + } +} + +func TestTranscriptCommandSanitizesTerminalControls(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/ssh/interactive-sessions/IS-7/transcript" { + t.Errorf("path = %q", r.URL.Path) + w.WriteHeader(http.StatusNotFound) + return + } + _, _ = w.Write([]byte("hello\n\x1b]52;c;bad\x07world\x1b[31m!\n")) + })) + defer server.Close() + + client := &apiClient{baseURL: server.URL, token: "gateway-token", client: server.Client()} + permissions := &ssh.Permissions{Extensions: map[string]string{ + "authorized": "true", + "fingerprint": "SHA256:test", + "login": "operator", + "role": "owner", + }} + var output bytes.Buffer + if exit := runCommand(context.Background(), &output, permissions, client, "transcript IS-7", sessionPTY{}); exit != 0 { + t.Fatalf("exit=%d output=%q", exit, output.String()) + } + got := output.String() + if strings.ContainsAny(got, "\x1b\x07") || strings.Contains(got, "]52") { + t.Fatalf("transcript retained terminal controls: %q", got) + } + if got != "hello\nworld!\n" { + t.Fatalf("transcript = %q", got) + } +} + func TestHelpNamesDeleteAsCanonicalCommand(t *testing.T) { var output bytes.Buffer printHelp(&output, fleetapi.User{Login: "operator", Role: "owner"}) diff --git a/cmd/crabfleet/main.go b/cmd/crabfleet/main.go index 8bc3febb..c0ab54ce 100644 --- a/cmd/crabfleet/main.go +++ b/cmd/crabfleet/main.go @@ -517,8 +517,9 @@ func (cmd transcriptCmd) Run(app *cli, api *fleetapi.Client) error { "transcript": transcript, }) } - fmt.Fprint(os.Stdout, transcript) - if !strings.HasSuffix(transcript, "\n") { + safeTranscript := fleettext.SafeMultiline(transcript) + fmt.Fprint(os.Stdout, safeTranscript) + if !strings.HasSuffix(safeTranscript, "\n") { fmt.Fprintln(os.Stdout) } return nil diff --git a/cmd/crabfleet/main_test.go b/cmd/crabfleet/main_test.go index b4173c97..6dee684d 100644 --- a/cmd/crabfleet/main_test.go +++ b/cmd/crabfleet/main_test.go @@ -222,6 +222,31 @@ func TestNewCommandSanitizesControlPlaneOutput(t *testing.T) { } } +func TestTranscriptCommandSanitizesControlPlaneOutput(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/ssh/interactive-sessions/IS-7/transcript" { + t.Errorf("path = %q", r.URL.Path) + w.WriteHeader(http.StatusNotFound) + return + } + _, _ = w.Write([]byte("hello\n\x1b]52;c;bad\x07world\x1b[31m!\n")) + })) + defer server.Close() + + app := &cli{API: server.URL, Token: "gateway-token", Fingerprint: "SHA256:test", NoInput: true} + output := captureStdout(t, func() { + if err := (transcriptCmd{ID: "IS-7"}).Run(app, app.apiClient()); err != nil { + t.Fatal(err) + } + }) + if strings.ContainsAny(output, "\x1b\x07") || strings.Contains(output, "]52") { + t.Fatalf("output contains terminal controls: %q", output) + } + if output != "hello\nworld!\n" { + t.Fatalf("output = %q", output) + } +} + func TestFirstLineSkipsBlankLines(t *testing.T) { if got, want := firstLine("\n\n https://example.com/vnc\nignored\n"), "https://example.com/vnc"; got != want { t.Fatalf("firstLine = %q, want %q", got, want) diff --git a/internal/fleettext/text.go b/internal/fleettext/text.go index 1f0eab3f..52a21324 100644 --- a/internal/fleettext/text.go +++ b/internal/fleettext/text.go @@ -15,13 +15,77 @@ func Safe(value string) string { if r == '\n' || r == '\r' || r == '\t' { return ' ' } - if r < 0x20 || r == 0x7f || (r >= 0x80 && r <= 0x9f) { + if isControl(r) { return -1 } return r }, value) } +func SafeMultiline(value string) string { + var out strings.Builder + const ( + stateText = iota + stateEscape + stateCSI + stateStringControl + stateStringControlEscape + ) + state := stateText + for _, r := range value { + switch state { + case stateText: + switch { + case r == '\x1b': + state = stateEscape + case r == '\x9b': + state = stateCSI + case r == '\x90' || r == '\x9d' || r == '\x9e' || r == '\x9f': + state = stateStringControl + case r == '\n': + out.WriteRune(r) + case r == '\t': + out.WriteRune(' ') + case isControl(r): + continue + default: + out.WriteRune(r) + } + case stateEscape: + switch r { + case '[': + state = stateCSI + case ']', 'P', '^', '_': + state = stateStringControl + default: + state = stateText + } + case stateCSI: + if r >= 0x40 && r <= 0x7e { + state = stateText + } + case stateStringControl: + switch r { + case '\x07', '\x9c': + state = stateText + case '\x1b': + state = stateStringControlEscape + } + case stateStringControlEscape: + if r == '\\' { + state = stateText + } else if r != '\x1b' { + state = stateStringControl + } + } + } + return out.String() +} + +func isControl(r rune) bool { + return r < 0x20 || r == 0x7f || (r >= 0x80 && r <= 0x9f) +} + func DisplayUser(user fleetapi.User) string { if user.Login != "" { return "@" + user.Login diff --git a/internal/fleettext/text_test.go b/internal/fleettext/text_test.go index 62b6688e..e8b8af78 100644 --- a/internal/fleettext/text_test.go +++ b/internal/fleettext/text_test.go @@ -48,3 +48,15 @@ func TestSafeRemovesTerminalControlCharacters(t *testing.T) { t.Fatalf("safe = %q, want %q", got, want) } } + +func TestSafeMultilineStripsTerminalSequences(t *testing.T) { + input := "hello\n\x1b]52;c;bad\x07world\x1b[31m!\x1b[0m\tok\rhidden" + got := SafeMultiline(input) + want := "hello\nworld! okhidden" + if got != want { + t.Fatalf("safe multiline = %q, want %q", got, want) + } + if strings.ContainsAny(got, "\x1b\x07\r") || strings.Contains(got, "]52") { + t.Fatalf("safe multiline retained terminal controls: %q", got) + } +} From b89aadce365bf6304ced2d26c51a7b12f29460b6 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Thu, 18 Jun 2026 17:48:39 +0800 Subject: [PATCH 10/14] fix(security): bound ssh sessions and quoting --- cmd/crabbox-ssh-gateway/main.go | 139 ++++++++++++++++-- cmd/crabbox-ssh-gateway/main_test.go | 202 +++++++++++++++++++++++++++ cmd/crabfleet/main.go | 36 +++-- cmd/crabfleet/main_test.go | 25 ++++ 4 files changed, 373 insertions(+), 29 deletions(-) diff --git a/cmd/crabbox-ssh-gateway/main.go b/cmd/crabbox-ssh-gateway/main.go index 912e530f..e0978cd2 100644 --- a/cmd/crabbox-ssh-gateway/main.go +++ b/cmd/crabbox-ssh-gateway/main.go @@ -26,7 +26,11 @@ import ( var ( sshHandshakeTimeout = 15 * time.Second sshAuthTimeout = 15 * time.Second + sshConnectionIdle = 30 * time.Second + sshSessionIdleTimer = 30 * time.Second sshHandshakeSlots = newConnectionLimiter(64) + sshConnectionSlots = newConnectionLimiter(256) + sshSessionChannels = 8 ) type connectionLimiter struct { @@ -172,30 +176,41 @@ func main() { } func acceptConn(raw net.Conn, config *ssh.ServerConfig, client *apiClient) { + if !sshConnectionSlots.acquire() { + log.Printf("connection limit reached for %s", raw.RemoteAddr()) + raw.Close() + return + } if !sshHandshakeSlots.acquire() { log.Printf("connection limit reached for %s", raw.RemoteAddr()) + sshConnectionSlots.release() raw.Close() return } - go handleConnWithRelease(raw, config, client, sshHandshakeSlots.release) + go handleConnWithRelease(raw, config, client, sshHandshakeSlots.release, sshConnectionSlots.release) } func handleConn(raw net.Conn, config *ssh.ServerConfig, client *apiClient) { - handleConnWithRelease(raw, config, client, nil) + handleConnWithRelease(raw, config, client, nil, nil) } func handleConnWithRelease( raw net.Conn, config *ssh.ServerConfig, client *apiClient, - release func(), + releaseHandshake func(), + releaseConnection func(), ) { - releaseHandshake := release defer func() { if releaseHandshake != nil { releaseHandshake() } }() + defer func() { + if releaseConnection != nil { + releaseConnection() + } + }() defer raw.Close() if err := raw.SetDeadline(time.Now().Add(sshHandshakeTimeout)); err != nil { log.Printf("handshake deadline %s: %v", raw.RemoteAddr(), err) @@ -215,24 +230,84 @@ func handleConnWithRelease( defer conn.Close() go ssh.DiscardRequests(reqs) - for ch := range chans { - if ch.ChannelType() != "session" { - ch.Reject(ssh.UnknownChannelType, "session channels only") - continue - } - channel, requests, err := ch.Accept() - if err != nil { - log.Printf("channel accept: %v", err) - continue + permissions := conn.Permissions + if permissions == nil { + permissions = &ssh.Permissions{Extensions: map[string]string{}} + } + sessionSlots := newConnectionLimiter(sshSessionChannels) + sessionDone := make(chan struct{}) + connectionClosed := make(chan struct{}) + defer close(connectionClosed) + activeSessions := 0 + connectionIdleTimer, connectionIdle := newConnectionIdleTimer() + defer stopTimer(connectionIdleTimer) + for { + select { + case <-connectionIdle: + return + case <-sessionDone: + if activeSessions > 0 { + activeSessions-- + } + if activeSessions == 0 { + stopTimer(connectionIdleTimer) + connectionIdleTimer, connectionIdle = newConnectionIdleTimer() + } + case ch, ok := <-chans: + if !ok { + return + } + if ch.ChannelType() != "session" { + ch.Reject(ssh.UnknownChannelType, "session channels only") + continue + } + if !sessionSlots.acquire() { + ch.Reject(ssh.ResourceShortage, "too many session channels") + continue + } + if activeSessions == 0 { + stopTimer(connectionIdleTimer) + connectionIdle = nil + } + activeSessions++ + channel, requests, err := ch.Accept() + if err != nil { + sessionSlots.release() + if activeSessions > 0 { + activeSessions-- + } + if activeSessions == 0 { + connectionIdleTimer, connectionIdle = newConnectionIdleTimer() + } + log.Printf("channel accept: %v", err) + continue + } + go handleSession(channel, requests, permissions, client, func() { + select { + case sessionDone <- struct{}{}: + case <-connectionClosed: + } + sessionSlots.release() + }) } - go handleSession(channel, requests, conn.Permissions, client) } } -func handleSession(channel ssh.Channel, requests <-chan *ssh.Request, perms *ssh.Permissions, client *apiClient) { +func handleSession( + channel ssh.Channel, + requests <-chan *ssh.Request, + perms *ssh.Permissions, + client *apiClient, + release func(), +) { + if release != nil { + defer release() + } defer channel.Close() ctx, cancel := context.WithCancel(context.Background()) defer cancel() + idleTimer, idle := newSessionIdleTimer() + defer stopTimer(idleTimer) pty := sessionPTY{ cols: 120, rows: 34, @@ -275,6 +350,8 @@ func handleSession(channel ssh.Channel, requests <-chan *ssh.Request, perms *ssh continue } commandStarted = true + stopTimer(idleTimer) + idle = nil req.Reply(true, nil) go func(current sessionPTY) { exitCh <- runCommand(ctx, channel, perms, client, "", current) @@ -287,6 +364,8 @@ func handleSession(channel ssh.Channel, requests <-chan *ssh.Request, perms *ssh var payload struct{ Command string } ssh.Unmarshal(req.Payload, &payload) commandStarted = true + stopTimer(idleTimer) + idle = nil req.Reply(true, nil) go func(current sessionPTY, command string) { exitCh <- runCommand( @@ -305,6 +384,36 @@ func handleSession(channel ssh.Channel, requests <-chan *ssh.Request, perms *ssh cancel() replyExit(channel, exit) return + case <-idle: + return + } + } +} + +func newSessionIdleTimer() (*time.Timer, <-chan time.Time) { + if sshSessionIdleTimer <= 0 { + return nil, nil + } + timer := time.NewTimer(sshSessionIdleTimer) + return timer, timer.C +} + +func newConnectionIdleTimer() (*time.Timer, <-chan time.Time) { + if sshConnectionIdle <= 0 { + return nil, nil + } + timer := time.NewTimer(sshConnectionIdle) + return timer, timer.C +} + +func stopTimer(timer *time.Timer) { + if timer == nil { + return + } + if !timer.Stop() { + select { + case <-timer.C: + default: } } } diff --git a/cmd/crabbox-ssh-gateway/main_test.go b/cmd/crabbox-ssh-gateway/main_test.go index e9733283..aedac76a 100644 --- a/cmd/crabbox-ssh-gateway/main_test.go +++ b/cmd/crabbox-ssh-gateway/main_test.go @@ -3,6 +3,8 @@ package main import ( "bytes" "context" + "crypto/ed25519" + "crypto/rand" "encoding/json" "net" "net/http" @@ -358,6 +360,160 @@ func TestAcceptConnRejectsWhenHandshakeSlotsFull(t *testing.T) { } } +func TestAcceptConnHoldsConnectionSlotUntilClose(t *testing.T) { + previousConnections := sshConnectionSlots + previousHandshakes := sshHandshakeSlots + sshConnectionSlots = newConnectionLimiter(1) + sshHandshakeSlots = newConnectionLimiter(1) + defer func() { + sshConnectionSlots = previousConnections + sshHandshakeSlots = previousHandshakes + }() + + addr, cleanup := serveTestSSHGateway(t, testSSHServerConfig(t), nil) + defer cleanup() + clientConfig := testSSHClientConfig() + first, err := ssh.Dial("tcp", addr, clientConfig) + if err != nil { + t.Fatal(err) + } + defer first.Close() + + if second, err := ssh.Dial("tcp", addr, clientConfig); err == nil { + second.Close() + t.Fatal("second connection succeeded while connection slot was occupied") + } + + first.Close() + deadline := time.Now().Add(time.Second) + for { + third, err := ssh.Dial("tcp", addr, clientConfig) + if err == nil { + third.Close() + return + } + if time.Now().After(deadline) { + t.Fatalf("connection slot was not released after close: %v", err) + } + time.Sleep(10 * time.Millisecond) + } +} + +func TestSessionChannelsAreBoundedPerConnection(t *testing.T) { + previousChannels := sshSessionChannels + previousIdle := sshSessionIdleTimer + sshSessionChannels = 1 + sshSessionIdleTimer = time.Second + defer func() { + sshSessionChannels = previousChannels + sshSessionIdleTimer = previousIdle + }() + + addr, cleanup := serveTestSSHGateway(t, testSSHServerConfig(t), nil) + defer cleanup() + client, err := ssh.Dial("tcp", addr, testSSHClientConfig()) + if err != nil { + t.Fatal(err) + } + defer client.Close() + first, err := client.NewSession() + if err != nil { + t.Fatal(err) + } + defer first.Close() + if second, err := client.NewSession(); err == nil { + second.Close() + t.Fatal("second session channel succeeded while per-connection slot was occupied") + } +} + +func TestIdleConnectionClosesWithoutSessionChannel(t *testing.T) { + previousIdle := sshConnectionIdle + previousConnections := sshConnectionSlots + sshConnectionIdle = 20 * time.Millisecond + sshConnectionSlots = newConnectionLimiter(1) + defer func() { + sshConnectionIdle = previousIdle + sshConnectionSlots = previousConnections + }() + + addr, cleanup := serveTestSSHGateway(t, testSSHServerConfig(t), nil) + defer cleanup() + clientConfig := testSSHClientConfig() + first, err := ssh.Dial("tcp", addr, clientConfig) + if err != nil { + t.Fatal(err) + } + defer first.Close() + + time.Sleep(100 * time.Millisecond) + if session, err := first.NewSession(); err == nil { + session.Close() + t.Fatal("idle connection still accepted a session channel") + } + second, err := ssh.Dial("tcp", addr, clientConfig) + if err != nil { + t.Fatalf("connection slot was not released after idle close: %v", err) + } + second.Close() +} + +func TestIdleConnectionClosesAfterLastSessionChannel(t *testing.T) { + previousIdle := sshConnectionIdle + previousSessionIdle := sshSessionIdleTimer + sshConnectionIdle = 20 * time.Millisecond + sshSessionIdleTimer = time.Second + defer func() { + sshConnectionIdle = previousIdle + sshSessionIdleTimer = previousSessionIdle + }() + + addr, cleanup := serveTestSSHGateway(t, testSSHServerConfig(t), nil) + defer cleanup() + client, err := ssh.Dial("tcp", addr, testSSHClientConfig()) + if err != nil { + t.Fatal(err) + } + defer client.Close() + session, err := client.NewSession() + if err != nil { + t.Fatal(err) + } + if err := session.Close(); err != nil && !strings.Contains(err.Error(), "EOF") { + t.Fatal(err) + } + + time.Sleep(100 * time.Millisecond) + if next, err := client.NewSession(); err == nil { + next.Close() + t.Fatal("idle connection still accepted a session after the last channel closed") + } +} + +func TestIdleSessionChannelClosesWithoutCommand(t *testing.T) { + previousIdle := sshSessionIdleTimer + sshSessionIdleTimer = 20 * time.Millisecond + defer func() { sshSessionIdleTimer = previousIdle }() + + addr, cleanup := serveTestSSHGateway(t, testSSHServerConfig(t), nil) + defer cleanup() + client, err := ssh.Dial("tcp", addr, testSSHClientConfig()) + if err != nil { + t.Fatal(err) + } + defer client.Close() + session, err := client.NewSession() + if err != nil { + t.Fatal(err) + } + defer session.Close() + + time.Sleep(100 * time.Millisecond) + if err := session.Shell(); err == nil { + t.Fatal("idle session channel still accepted a shell request") + } +} + func TestRunCommandCancelsControlPlaneRequest(t *testing.T) { entered := make(chan struct{}) cancelled := make(chan struct{}) @@ -552,3 +708,49 @@ func TestPrintListShowsOwnersAndSessionTree(t *testing.T) { } } } + +func testSSHServerConfig(t *testing.T) *ssh.ServerConfig { + t.Helper() + _, privateKey, err := ed25519.GenerateKey(rand.Reader) + if err != nil { + t.Fatal(err) + } + signer, err := ssh.NewSignerFromKey(privateKey) + if err != nil { + t.Fatal(err) + } + config := &ssh.ServerConfig{NoClientAuth: true} + config.AddHostKey(signer) + return config +} + +func testSSHClientConfig() *ssh.ClientConfig { + return &ssh.ClientConfig{ + User: "link", + HostKeyCallback: ssh.InsecureIgnoreHostKey(), + Timeout: time.Second, + } +} + +func serveTestSSHGateway(t *testing.T, config *ssh.ServerConfig, client *apiClient) (string, func()) { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + done := make(chan struct{}) + go func() { + defer close(done) + for { + conn, err := listener.Accept() + if err != nil { + return + } + acceptConn(conn, config, client) + } + }() + return listener.Addr().String(), func() { + listener.Close() + <-done + } +} diff --git a/cmd/crabfleet/main.go b/cmd/crabfleet/main.go index c0ab54ce..8f3f5c7e 100644 --- a/cmd/crabfleet/main.go +++ b/cmd/crabfleet/main.go @@ -600,7 +600,10 @@ func (openCmd) Run(app *cli, _ *fleetapi.Client) error { } func runSSH(app *cli, args ...string) error { - sshArgs := append([]string{app.SSHHost}, args...) + sshArgs := []string{app.SSHHost} + if command := sshRemoteCommand(args...); command != "" { + sshArgs = append(sshArgs, command) + } cmd := exec.Command("ssh", sshArgs...) cmd.Stdin = os.Stdin cmd.Stdout = os.Stdout @@ -609,29 +612,35 @@ func runSSH(app *cli, args ...string) error { } func runSSHCommand(app *cli, args ...string) error { - parts := make([]string, len(args)) - for i, arg := range args { - parts[i] = shellQuote(arg) - } - return runSSH(app, strings.Join(parts, " ")) + return runSSH(app, args...) } func runSSHCommandOutput(app *cli, args ...string) (string, error) { - parts := make([]string, len(args)) - for i, arg := range args { - parts[i] = shellQuote(arg) - } - return runSSHOutput(app, strings.Join(parts, " ")) + return runSSHOutput(app, args...) } func runSSHOutput(app *cli, args ...string) (string, error) { - sshArgs := append([]string{app.SSHHost}, args...) + sshArgs := []string{app.SSHHost} + if command := sshRemoteCommand(args...); command != "" { + sshArgs = append(sshArgs, command) + } cmd := exec.Command("ssh", sshArgs...) cmd.Stderr = os.Stderr output, err := cmd.Output() return string(output), err } +func sshRemoteCommand(args ...string) string { + if len(args) == 0 { + return "" + } + parts := make([]string, len(args)) + for i, arg := range args { + parts[i] = shellQuote(arg) + } + return strings.Join(parts, " ") +} + func shellQuote(value string) string { if value == "" { return "''" @@ -701,8 +710,7 @@ func isPreRequestNetworkFailure(err error) bool { return true } lower := strings.ToLower(err.Error()) - return strings.Contains(lower, "tls:") || - strings.Contains(lower, "server gave http response to https client") + return strings.Contains(lower, "server gave http response to https client") } func openURL(url string) error { diff --git a/cmd/crabfleet/main_test.go b/cmd/crabfleet/main_test.go index 6dee684d..be709c49 100644 --- a/cmd/crabfleet/main_test.go +++ b/cmd/crabfleet/main_test.go @@ -3,10 +3,12 @@ package main import ( "bytes" "encoding/json" + "errors" "io" "net" "net/http" "net/http/httptest" + "net/url" "os" "path/filepath" "strings" @@ -68,6 +70,18 @@ func TestShellQuoteQuotesMetacharacters(t *testing.T) { } } +func TestRunSSHQuotesRemoteCommandArguments(t *testing.T) { + argsPath := installFakeSSH(t) + app := &cli{SSHHost: "crabd.test"} + if err := runSSH(app, "attach", "IS-1; touch /tmp/pwned"); err != nil { + t.Fatal(err) + } + output := readFakeSSHArgs(t, argsPath) + if got, want := output, "crabd.test\nattach 'IS-1; touch /tmp/pwned'\n"; got != want { + t.Fatalf("ssh args = %q, want %q", got, want) + } +} + func TestMutatingAPIFailureDoesNotFallbackToSSH(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost || r.URL.Path != "/api/ssh/interactive-sessions" { @@ -120,6 +134,17 @@ func TestPreRequestAPIFailureStillFallsBackToSSH(t *testing.T) { } } +func TestAmbiguousTLSMutationFailureDoesNotFallbackToSSH(t *testing.T) { + err := &url.Error{ + Op: "Post", + URL: "https://crabfleet.test/api/ssh/interactive-sessions", + Err: errors.New("tls: bad record MAC"), + } + if canFallbackToSSH(&cli{SSHHost: "crabd.test"}, err) { + t.Fatal("generic TLS failure was treated as safe to retry") + } +} + func TestLocalAuthFailureStillFallsBackToSSH(t *testing.T) { argsPath := installFakeSSH(t) From 5d389eaa86b49a167abd056be43e564ff3923635 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Thu, 18 Jun 2026 18:01:03 +0800 Subject: [PATCH 11/14] fix(cli): validate terminal output inputs --- cmd/crabbox-ssh-gateway/main.go | 10 ++- cmd/crabbox-ssh-gateway/main_test.go | 39 ++++++++ cmd/crabfleet/main.go | 75 +++++++++++++--- cmd/crabfleet/main_test.go | 128 ++++++++++++++++++++++++++- 4 files changed, 237 insertions(+), 15 deletions(-) diff --git a/cmd/crabbox-ssh-gateway/main.go b/cmd/crabbox-ssh-gateway/main.go index e0978cd2..dc13acdc 100644 --- a/cmd/crabbox-ssh-gateway/main.go +++ b/cmd/crabbox-ssh-gateway/main.go @@ -503,9 +503,15 @@ func runCommand(ctx context.Context, out io.ReadWriter, perms *ssh.Permissions, fmt.Fprintf(out, "error: %v\n", err) return 1 } - fmt.Fprintf(out, "session: %s\nrepo: %s\nstatus: %s\n", session.ID, session.Repo, session.Status) + fmt.Fprintf( + out, + "session: %s\nrepo: %s\nstatus: %s\n", + fleettext.Safe(session.ID), + fleettext.Safe(session.Repo), + fleettext.Safe(session.Status), + ) if session.Attachable() { - fmt.Fprintf(out, "attach: ssh crabfleet attach %s\n", session.ID) + fmt.Fprintf(out, "attach: ssh crabfleet attach %s\n", fleettext.Safe(session.ID)) } if session.VNCURL != "" { fmt.Fprintf(out, "vnc: %s\n", fleettext.Safe(session.VNCURL)) diff --git a/cmd/crabbox-ssh-gateway/main_test.go b/cmd/crabbox-ssh-gateway/main_test.go index aedac76a..1f8c63d9 100644 --- a/cmd/crabbox-ssh-gateway/main_test.go +++ b/cmd/crabbox-ssh-gateway/main_test.go @@ -653,6 +653,45 @@ func TestTranscriptCommandSanitizesTerminalControls(t *testing.T) { } } +func TestNewCommandSanitizesTerminalControls(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != "/api/ssh/interactive-sessions" { + t.Errorf("request = %s %s", r.Method, r.URL.Path) + w.WriteHeader(http.StatusNotFound) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"session":{"id":"IS-7\u001b]52;c;bad\u0007","repo":"openclaw/crabfleet\u001b[31m","status":"ready","ptyAvailable":true}}`)) + })) + defer server.Close() + + client := &apiClient{baseURL: server.URL, token: "gateway-token", client: server.Client()} + permissions := &ssh.Permissions{Extensions: map[string]string{ + "authorized": "true", + "fingerprint": "SHA256:test", + "login": "operator", + "role": "owner", + }} + var output bytes.Buffer + if exit := runCommand(context.Background(), &output, permissions, client, "new --detach --repo openclaw/crabfleet fix", sessionPTY{}); exit != 0 { + t.Fatalf("exit=%d output=%q", exit, output.String()) + } + got := output.String() + if strings.ContainsAny(got, "\x1b\x07\r") { + t.Fatalf("new output retained terminal controls: %q", got) + } + for _, want := range []string{ + "session: IS-7]52;c;bad\n", + "repo: openclaw/crabfleet[31m\n", + "status: ready\n", + "attach: ssh crabfleet attach IS-7]52;c;bad\n", + } { + if !strings.Contains(got, want) { + t.Fatalf("new output missing %q:\n%s", want, got) + } + } +} + func TestHelpNamesDeleteAsCanonicalCommand(t *testing.T) { var output bytes.Buffer printHelp(&output, fleetapi.User{Login: "operator", Role: "owner"}) diff --git a/cmd/crabfleet/main.go b/cmd/crabfleet/main.go index 8f3f5c7e..1d7d46f2 100644 --- a/cmd/crabfleet/main.go +++ b/cmd/crabfleet/main.go @@ -23,6 +23,7 @@ import ( const defaultAPIURL = "https://crabfleet.openclaw.ai" const defaultSSHHost = "crabd.sh" +const maxMessageBytes = 64 * 1024 var version = "dev" @@ -209,7 +210,7 @@ func (cmd newCmd) Run(app *cli, api *fleetapi.Client) error { return captureErr } if url := vncURLFromOutput(output); url != "" { - return openURL(url) + return openWebVNCURL(url) } return nil } @@ -243,7 +244,7 @@ func (cmd newCmd) Run(app *cli, api *fleetapi.Client) error { fmt.Fprintf(os.Stdout, "vnc: %s\n", fleettext.Safe(session.VNCURL)) } if cmd.VNC && session.VNCURL != "" { - return openURL(session.VNCURL) + return openWebVNCURL(session.VNCURL) } if !cmd.Detach && !app.NoInput && isTerminal(os.Stdin) && isTerminal(os.Stdout) && session.Attachable() { return runSSH(app, "attach", session.ID) @@ -379,7 +380,7 @@ func (doctorCmd) Run(app *cli, api *fleetapi.Client) error { keys := []string{"api", "auth", "user", "role", "sessions"} for _, key := range keys { if value := result[key]; value != "" { - fmt.Fprintf(os.Stdout, "%s: %s\n", key, value) + fmt.Fprintf(os.Stdout, "%s: %s\n", key, fleettext.Safe(value)) } } return nil @@ -468,7 +469,7 @@ func (cmd vncCmd) Run(app *cli, api *fleetapi.Client) error { if url == "" { return errors.New("ssh gateway did not return a WebVNC URL") } - return openURL(url) + return openWebVNCURL(url) } return runSSH(app, "vnc", cmd.ID) } @@ -480,7 +481,7 @@ func (cmd vncCmd) Run(app *cli, api *fleetapi.Client) error { return fmt.Errorf("session %s has no WebVNC URL yet", cmd.ID) } if cmd.Open { - return openURL(session.VNCURL) + return openWebVNCURL(session.VNCURL) } fmt.Fprintln(os.Stdout, fleettext.Safe(session.VNCURL)) return nil @@ -528,10 +529,13 @@ func (cmd transcriptCmd) Run(app *cli, api *fleetapi.Client) error { func (cmd messageCmd) Run(app *cli, api *fleetapi.Client) error { message := strings.Join(cmd.Text, " ") if message == "" && !isTerminal(os.Stdin) { - data, err := io.ReadAll(io.LimitReader(os.Stdin, 64*1024)) + data, err := io.ReadAll(io.LimitReader(os.Stdin, maxMessageBytes+1)) if err != nil { return err } + if len(data) > maxMessageBytes { + return fmt.Errorf("message exceeds %d bytes", maxMessageBytes) + } message = strings.TrimRight(string(data), "\r\n") } if message == "" { @@ -600,9 +604,9 @@ func (openCmd) Run(app *cli, _ *fleetapi.Client) error { } func runSSH(app *cli, args ...string) error { - sshArgs := []string{app.SSHHost} - if command := sshRemoteCommand(args...); command != "" { - sshArgs = append(sshArgs, command) + sshArgs, err := sshInvocationArgs(app, args...) + if err != nil { + return err } cmd := exec.Command("ssh", sshArgs...) cmd.Stdin = os.Stdin @@ -620,9 +624,9 @@ func runSSHCommandOutput(app *cli, args ...string) (string, error) { } func runSSHOutput(app *cli, args ...string) (string, error) { - sshArgs := []string{app.SSHHost} - if command := sshRemoteCommand(args...); command != "" { - sshArgs = append(sshArgs, command) + sshArgs, err := sshInvocationArgs(app, args...) + if err != nil { + return "", err } cmd := exec.Command("ssh", sshArgs...) cmd.Stderr = os.Stderr @@ -630,6 +634,18 @@ func runSSHOutput(app *cli, args ...string) (string, error) { return string(output), err } +func sshInvocationArgs(app *cli, args ...string) ([]string, error) { + host := strings.TrimSpace(app.SSHHost) + if host == "" || strings.HasPrefix(host, "-") { + return nil, fmt.Errorf("invalid SSH host: %q", app.SSHHost) + } + sshArgs := []string{"--", host} + if command := sshRemoteCommand(args...); command != "" { + sshArgs = append(sshArgs, command) + } + return sshArgs, nil +} + func sshRemoteCommand(args ...string) string { if len(args) == 0 { return "" @@ -726,6 +742,41 @@ func openURL(url string) error { return cmd.Run() } +func openWebVNCURL(raw string) error { + safeURL, err := validateWebVNCURL(raw) + if err != nil { + return err + } + return openURL(safeURL) +} + +func validateWebVNCURL(raw string) (string, error) { + parsed, err := url.Parse(raw) + if err != nil || parsed.Scheme == "" || parsed.Host == "" { + return "", fmt.Errorf("invalid WebVNC URL: %q", raw) + } + if parsed.User != nil { + return "", errors.New("invalid WebVNC URL: credentials are not allowed") + } + switch parsed.Scheme { + case "https": + return parsed.String(), nil + case "http": + if isLoopbackHost(parsed.Hostname()) { + return parsed.String(), nil + } + } + return "", fmt.Errorf("invalid WebVNC URL scheme: %s", parsed.Scheme) +} + +func isLoopbackHost(host string) bool { + if strings.EqualFold(host, "localhost") { + return true + } + ip := net.ParseIP(host) + return ip != nil && ip.IsLoopback() +} + func firstLine(value string) string { for _, line := range strings.Split(value, "\n") { if trimmed := strings.TrimSpace(line); trimmed != "" { diff --git a/cmd/crabfleet/main_test.go b/cmd/crabfleet/main_test.go index be709c49..e451120a 100644 --- a/cmd/crabfleet/main_test.go +++ b/cmd/crabfleet/main_test.go @@ -77,11 +77,20 @@ func TestRunSSHQuotesRemoteCommandArguments(t *testing.T) { t.Fatal(err) } output := readFakeSSHArgs(t, argsPath) - if got, want := output, "crabd.test\nattach 'IS-1; touch /tmp/pwned'\n"; got != want { + if got, want := output, "--\ncrabd.test\nattach 'IS-1; touch /tmp/pwned'\n"; got != want { t.Fatalf("ssh args = %q, want %q", got, want) } } +func TestRunSSHRejectsOptionLikeHost(t *testing.T) { + installFakeSSH(t) + app := &cli{SSHHost: "-oProxyCommand=bad"} + err := runSSH(app, "whoami") + if err == nil || !strings.Contains(err.Error(), "invalid SSH host") { + t.Fatalf("error = %v", err) + } +} + func TestMutatingAPIFailureDoesNotFallbackToSSH(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost || r.URL.Path != "/api/ssh/interactive-sessions" { @@ -217,6 +226,17 @@ func installFakeSSH(t *testing.T) string { return argsPath } +func installOutputSSH(t *testing.T, output string) { + t.Helper() + dir := t.TempDir() + sshPath := filepath.Join(dir, "ssh") + if err := os.WriteFile(sshPath, []byte("#!/bin/sh\nprintf '%s' \"$SSH_OUTPUT\"\n"), 0o755); err != nil { + t.Fatal(err) + } + t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH")) + t.Setenv("SSH_OUTPUT", output) +} + func readFakeSSHArgs(t *testing.T, argsPath string) string { t.Helper() data, err := os.ReadFile(argsPath) @@ -247,6 +267,31 @@ func TestNewCommandSanitizesControlPlaneOutput(t *testing.T) { } } +func TestDoctorSanitizesControlPlaneErrors(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/healthz" { + _, _ = w.Write([]byte("ok")) + return + } + w.WriteHeader(http.StatusInternalServerError) + _, _ = w.Write([]byte("bad\x1b]52;c;secret\x07state")) + })) + defer server.Close() + + app := &cli{API: server.URL, Token: "gateway-token", Fingerprint: "SHA256:test"} + output := captureStdout(t, func() { + if err := (doctorCmd{}).Run(app, app.apiClient()); err != nil { + t.Fatal(err) + } + }) + if strings.ContainsAny(output, "\x1b\x07") { + t.Fatalf("doctor output contains terminal controls: %q", output) + } + if !strings.Contains(output, "auth: failed: crabfleet API 500 Internal Server Error: bad]52;c;secretstate") { + t.Fatalf("doctor output = %q", output) + } +} + func TestTranscriptCommandSanitizesControlPlaneOutput(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path != "/api/ssh/interactive-sessions/IS-7/transcript" { @@ -272,6 +317,87 @@ func TestTranscriptCommandSanitizesControlPlaneOutput(t *testing.T) { } } +func TestMessageRejectsOversizedPipedInput(t *testing.T) { + calls := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls++ + w.WriteHeader(http.StatusNoContent) + })) + defer server.Close() + + previousStdin := os.Stdin + reader, writer, err := os.Pipe() + if err != nil { + t.Fatal(err) + } + os.Stdin = reader + defer func() { + os.Stdin = previousStdin + reader.Close() + }() + go func() { + _, _ = writer.Write([]byte(strings.Repeat("a", maxMessageBytes+1))) + _ = writer.Close() + }() + + app := &cli{API: server.URL, Token: "gateway-token", Fingerprint: "SHA256:test", NoInput: true} + err = (messageCmd{ID: "IS-7"}).Run(app, app.apiClient()) + if err == nil || !strings.Contains(err.Error(), "message exceeds") { + t.Fatalf("error = %v", err) + } + if calls != 0 { + t.Fatalf("message request calls = %d, want 0", calls) + } +} + +func TestValidateWebVNCURL(t *testing.T) { + tests := []struct { + raw string + allowed bool + }{ + {raw: "https://example.test/vnc", allowed: true}, + {raw: "http://localhost:6080/vnc", allowed: true}, + {raw: "http://127.0.0.1:6080/vnc", allowed: true}, + {raw: "http://example.test/vnc", allowed: false}, + {raw: "file:///tmp/vnc", allowed: false}, + {raw: "custom:vnc", allowed: false}, + {raw: "/relative/vnc", allowed: false}, + {raw: "https://user@example.test/vnc", allowed: false}, + } + for _, tt := range tests { + _, err := validateWebVNCURL(tt.raw) + if tt.allowed && err != nil { + t.Fatalf("validateWebVNCURL(%q) = %v", tt.raw, err) + } + if !tt.allowed && err == nil { + t.Fatalf("validateWebVNCURL(%q) unexpectedly succeeded", tt.raw) + } + } +} + +func TestNewVNCFallbackValidatesCapturedURL(t *testing.T) { + installOutputSSH(t, "vnc: http://example.test/not-webvnc\n") + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + apiURL := "http://" + listener.Addr().String() + if err := listener.Close(); err != nil { + t.Fatal(err) + } + + app := &cli{ + API: apiURL, + SSHHost: "crabd.test", + Token: "gateway-token", + Fingerprint: "SHA256:test", + } + err = (newCmd{Branch: "main", Command: "codex --yolo", Repo: "openclaw/crabfleet", VNC: true}).Run(app, app.apiClient()) + if err == nil || !strings.Contains(err.Error(), "invalid WebVNC URL scheme") { + t.Fatalf("error = %v", err) + } +} + func TestFirstLineSkipsBlankLines(t *testing.T) { if got, want := firstLine("\n\n https://example.com/vnc\nignored\n"), "https://example.com/vnc"; got != want { t.Fatalf("firstLine = %q, want %q", got, want) From 27c2810cd67d8fb46e8ab2d4b37df170b739cce5 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Thu, 18 Jun 2026 18:12:47 +0800 Subject: [PATCH 12/14] fix(api): bound json and patch summaries --- cmd/crabbox-ssh-gateway/main.go | 11 +++++- cmd/crabbox-ssh-gateway/main_test.go | 36 ++++++++++++++++++ cmd/crabfleet/main.go | 9 ++++- cmd/crabfleet/main_test.go | 56 ++++++++++++++++++++++++++++ internal/fleetapi/client.go | 26 ++++++++----- internal/fleetapi/client_test.go | 9 ++--- 6 files changed, 128 insertions(+), 19 deletions(-) diff --git a/cmd/crabbox-ssh-gateway/main.go b/cmd/crabbox-ssh-gateway/main.go index dc13acdc..b8eeb012 100644 --- a/cmd/crabbox-ssh-gateway/main.go +++ b/cmd/crabbox-ssh-gateway/main.go @@ -643,7 +643,14 @@ func runCommand(ctx context.Context, out io.ReadWriter, perms *ssh.Permissions, fmt.Fprintf(out, "session %s not found\n", fleettext.Safe(args[1])) return 1 } - session, err := api.UpdateSummary(ctx, args[1], update.summary, update.purpose) + fields := map[string]string{} + if update.summary != "" { + fields["summary"] = update.summary + } + if update.purpose != "" { + fields["purpose"] = update.purpose + } + session, err := api.UpdateSummary(ctx, args[1], fields) if err != nil { fmt.Fprintf(out, "error: %v\n", err) return 1 @@ -800,7 +807,7 @@ func parseCreate(ctx context.Context, args []string, api *fleetapi.Client) (crea req.Prompt = strings.Join(fs.Args(), " ") if req.Repo == "" && api != nil { state, err := api.State(ctx) - if err != nil && (ctx.Err() != nil || errors.Is(err, context.Canceled)) { + if err != nil { return createArgs{}, err } if err == nil && len(state.Repos) > 0 { diff --git a/cmd/crabbox-ssh-gateway/main_test.go b/cmd/crabbox-ssh-gateway/main_test.go index 1f8c63d9..815278ba 100644 --- a/cmd/crabbox-ssh-gateway/main_test.go +++ b/cmd/crabbox-ssh-gateway/main_test.go @@ -622,6 +622,42 @@ func TestRunCommandCancelsDefaultRepoLookup(t *testing.T) { } } +func TestRunCommandStopsAfterDefaultRepoLookupFailure(t *testing.T) { + createCalled := false + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/api/ssh/state": + w.WriteHeader(http.StatusInternalServerError) + _, _ = w.Write([]byte("state unavailable")) + case "/api/ssh/interactive-sessions": + createCalled = true + w.WriteHeader(http.StatusCreated) + default: + t.Errorf("path = %q", r.URL.Path) + w.WriteHeader(http.StatusNotFound) + } + })) + defer server.Close() + + client := &apiClient{baseURL: server.URL, token: "gateway-token", client: server.Client()} + permissions := &ssh.Permissions{Extensions: map[string]string{ + "authorized": "true", + "fingerprint": "SHA256:test", + "login": "operator", + "role": "owner", + }} + var output bytes.Buffer + if exit := runCommand(context.Background(), &output, permissions, client, "new fix it", sessionPTY{}); exit != 2 { + t.Fatalf("exit=%d output=%q", exit, output.String()) + } + if createCalled { + t.Fatal("create session was called after default repo lookup failed") + } + if !strings.Contains(output.String(), "state unavailable") { + t.Fatalf("output = %q", output.String()) + } +} + func TestTranscriptCommandSanitizesTerminalControls(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path != "/api/ssh/interactive-sessions/IS-7/transcript" { diff --git a/cmd/crabfleet/main.go b/cmd/crabfleet/main.go index 1d7d46f2..1310f466 100644 --- a/cmd/crabfleet/main.go +++ b/cmd/crabfleet/main.go @@ -578,7 +578,14 @@ func (cmd summaryCmd) Run(app *cli, api *fleetapi.Client) error { fleettext.WriteSessionSummary(os.Stdout, session) return nil } - session, err := api.UpdateSummary(context.Background(), cmd.ID, summary, cmd.Purpose) + fields := map[string]string{} + if summary != "" { + fields["summary"] = summary + } + if cmd.Purpose != "" { + fields["purpose"] = cmd.Purpose + } + session, err := api.UpdateSummary(context.Background(), cmd.ID, fields) if err != nil { if canFallbackToSSH(app, err) { args := []string{"summary", cmd.ID} diff --git a/cmd/crabfleet/main_test.go b/cmd/crabfleet/main_test.go index e451120a..f5ca8a58 100644 --- a/cmd/crabfleet/main_test.go +++ b/cmd/crabfleet/main_test.go @@ -350,6 +350,62 @@ func TestMessageRejectsOversizedPipedInput(t *testing.T) { } } +func TestSummaryUpdateOmitsUnchangedFields(t *testing.T) { + tests := []struct { + name string + command summaryCmd + wantField string + wantValue string + omitField string + }{ + { + name: "purpose only", + command: summaryCmd{ID: "IS-7", Purpose: "new purpose"}, + wantField: "purpose", + wantValue: "new purpose", + omitField: "summary", + }, + { + name: "summary only", + command: summaryCmd{ID: "IS-7", Text: []string{"new", "summary"}}, + wantField: "summary", + wantValue: "new summary", + omitField: "purpose", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var body map[string]string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/ssh/interactive-sessions/IS-7/summary" { + t.Errorf("path = %q", r.URL.Path) + w.WriteHeader(http.StatusNotFound) + return + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Error(err) + w.WriteHeader(http.StatusBadRequest) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"session":{"id":"IS-7","status":"ready"}}`)) + })) + defer server.Close() + + app := &cli{API: server.URL, Token: "gateway-token", Fingerprint: "SHA256:test", NoInput: true} + if err := tt.command.Run(app, app.apiClient()); err != nil { + t.Fatal(err) + } + if got := body[tt.wantField]; got != tt.wantValue { + t.Fatalf("%s = %q, want %q in body %#v", tt.wantField, got, tt.wantValue, body) + } + if _, ok := body[tt.omitField]; ok { + t.Fatalf("%s was present in body %#v", tt.omitField, body) + } + }) + } +} + func TestValidateWebVNCURL(t *testing.T) { tests := []struct { raw string diff --git a/internal/fleetapi/client.go b/internal/fleetapi/client.go index b0a92f31..baa1f4a0 100644 --- a/internal/fleetapi/client.go +++ b/internal/fleetapi/client.go @@ -165,21 +165,15 @@ func (c *Client) Transcript(ctx context.Context, id string) (string, error) { if err := responseError(resp); err != nil { return "", err } - data, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes+1)) - if len(data) > maxResponseBytes { - return "", fmt.Errorf("crabfleet API response exceeds %d bytes", maxResponseBytes) - } + data, err := readBoundedResponse(resp.Body) return string(data), err } -func (c *Client) UpdateSummary(ctx context.Context, id string, summary string, purpose string) (Session, error) { +func (c *Client) UpdateSummary(ctx context.Context, id string, fields map[string]string) (Session, error) { var out struct { Session Session `json:"session"` } - err := c.doJSON(ctx, http.MethodPost, sessionPath(id)+"/summary", map[string]string{ - "summary": summary, - "purpose": purpose, - }, &out) + err := c.doJSON(ctx, http.MethodPost, sessionPath(id)+"/summary", fields, &out) return out.Session, err } @@ -258,7 +252,11 @@ func (c *Client) doJSON(ctx context.Context, method string, path string, body an if out == nil { return nil } - return json.NewDecoder(resp.Body).Decode(out) + data, err := readBoundedResponse(resp.Body) + if err != nil { + return err + } + return json.Unmarshal(data, out) } func (c *Client) open( @@ -310,6 +308,14 @@ func responseError(resp *http.Response) error { return nil } +func readBoundedResponse(body io.Reader) ([]byte, error) { + data, err := io.ReadAll(io.LimitReader(body, maxResponseBytes+1)) + if len(data) > maxResponseBytes { + return nil, fmt.Errorf("crabfleet API response exceeds %d bytes", maxResponseBytes) + } + return data, err +} + func (a Auth) path(path string) (string, error) { if err := a.validate(); err != nil { return "", err diff --git a/internal/fleetapi/client_test.go b/internal/fleetapi/client_test.go index 714fd23a..1610a0da 100644 --- a/internal/fleetapi/client_test.go +++ b/internal/fleetapi/client_test.go @@ -66,7 +66,7 @@ func TestClientRejectsIncompleteAuthentication(t *testing.T) { } } -func TestClientStreamsLargeJSONResponses(t *testing.T) { +func TestClientRejectsOversizedJSONResponses(t *testing.T) { largeLogin := strings.Repeat("a", maxResponseBytes+1) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write([]byte(`{"user":{"login":"`)) @@ -77,11 +77,8 @@ func TestClientStreamsLargeJSONResponses(t *testing.T) { client := NewClient(server.URL, server.Client(), SSHAuth("gateway-token", "SHA256:test")) state, err := client.State(context.Background()) - if err != nil { - t.Fatal(err) - } - if state.User.Login != largeLogin { - t.Fatalf("login length = %d, want %d", len(state.User.Login), len(largeLogin)) + if err == nil || !strings.Contains(err.Error(), "response exceeds") { + t.Fatalf("state=%#v error=%v", state, err) } } From 087c5015af9f7338bc4aa287429a30f3830c41b7 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Thu, 18 Jun 2026 18:16:22 +0800 Subject: [PATCH 13/14] fix(gateway): sanitize delete notes --- cmd/crabbox-ssh-gateway/main.go | 43 +++++++++++++++--------- cmd/crabbox-ssh-gateway/main_test.go | 50 ++++++++++++++++++++++++++++ 2 files changed, 77 insertions(+), 16 deletions(-) diff --git a/cmd/crabbox-ssh-gateway/main.go b/cmd/crabbox-ssh-gateway/main.go index b8eeb012..dae62dea 100644 --- a/cmd/crabbox-ssh-gateway/main.go +++ b/cmd/crabbox-ssh-gateway/main.go @@ -452,14 +452,14 @@ func runCommand(ctx context.Context, out io.ReadWriter, perms *ssh.Permissions, }, } if !auth.authorized { - fmt.Fprintf(out, "Crabfleet SSH key not linked.\n\nOpen this URL to connect it:\n%s\n\nThen run ssh again.\n", auth.linkURL) + fmt.Fprintf(out, "Crabfleet SSH key not linked.\n\nOpen this URL to connect it:\n%s\n\nThen run ssh again.\n", fleettext.Safe(auth.linkURL)) return 1 } api := client.controlPlane(auth.fingerprint) args, err := splitCommand(command) if err != nil { - fmt.Fprintf(out, "error: %v\n", err) + writeError(out, err) return 2 } if len(args) == 0 { @@ -473,7 +473,7 @@ func runCommand(ctx context.Context, out io.ReadWriter, perms *ssh.Permissions, case "whoami": state, err := api.State(ctx) if err != nil { - fmt.Fprintf(out, "error: %v\n", err) + writeError(out, err) return 1 } fmt.Fprintf( @@ -487,7 +487,7 @@ func runCommand(ctx context.Context, out io.ReadWriter, perms *ssh.Permissions, case "list", "ls": state, err := api.State(ctx) if err != nil { - fmt.Fprintf(out, "error: %v\n", err) + writeError(out, err) return 1 } printList(out, state) @@ -495,12 +495,12 @@ func runCommand(ctx context.Context, out io.ReadWriter, perms *ssh.Permissions, case "new": create, err := parseCreate(ctx, args[1:], api) if err != nil { - fmt.Fprintf(out, "usage: new [--repo owner/repo] [--branch main] [--runtime crabbox|container] [--profile name] [prompt]\nerror: %v\n", err) + fmt.Fprintf(out, "usage: new [--repo owner/repo] [--branch main] [--runtime crabbox|container] [--profile name] [prompt]\nerror: %s\n", safeError(err)) return 2 } session, err := api.CreateSession(ctx, create.request) if err != nil { - fmt.Fprintf(out, "error: %v\n", err) + writeError(out, err) return 1 } fmt.Fprintf( @@ -539,7 +539,7 @@ func runCommand(ctx context.Context, out io.ReadWriter, perms *ssh.Permissions, } state, err := api.State(ctx) if err != nil { - fmt.Fprintf(out, "error: %v\n", err) + writeError(out, err) return 1 } for _, session := range state.InteractiveSessions { @@ -562,12 +562,12 @@ func runCommand(ctx context.Context, out io.ReadWriter, perms *ssh.Permissions, } session, err := api.Action(ctx, args[1], "stop") if err != nil { - fmt.Fprintf(out, "error: %v\n", err) + writeError(out, err) return 1 } fmt.Fprintf(out, "session: %s\nstatus: %s\n", fleettext.Safe(session.ID), fleettext.Safe(session.Status)) if note := session.LifecycleStopNote(); note != "" { - fmt.Fprintf(out, "note: %s\n", note) + fmt.Fprintf(out, "note: %s\n", fleettext.Safe(note)) } return 0 case "logs": @@ -577,7 +577,7 @@ func runCommand(ctx context.Context, out io.ReadWriter, perms *ssh.Permissions, } logs, err := api.Logs(ctx, args[1]) if err != nil { - fmt.Fprintf(out, "error: %v\n", err) + writeError(out, err) return 1 } fleettext.WriteSessionLogs(out, logs) @@ -589,7 +589,7 @@ func runCommand(ctx context.Context, out io.ReadWriter, perms *ssh.Permissions, } transcript, err := api.Transcript(ctx, args[1]) if err != nil { - fmt.Fprintf(out, "error: %v\n", err) + writeError(out, err) return 1 } safeTranscript := fleettext.SafeMultiline(transcript) @@ -605,7 +605,7 @@ func runCommand(ctx context.Context, out io.ReadWriter, perms *ssh.Permissions, } message, err := parseMessage(args[2:]) if err != nil { - fmt.Fprintf(out, "usage: message SESSION_ID [--no-enter] TEXT\nerror: %v\n", err) + fmt.Fprintf(out, "usage: message SESSION_ID [--no-enter] TEXT\nerror: %s\n", safeError(err)) return 2 } if message.text == "" { @@ -613,7 +613,7 @@ func runCommand(ctx context.Context, out io.ReadWriter, perms *ssh.Permissions, return 2 } if err := api.Message(ctx, args[1], message.text, !message.noEnter, pty.cols, pty.rows); err != nil { - fmt.Fprintf(out, "error: %v\n", err) + writeError(out, err) return 1 } fmt.Fprintf(out, "sent: %s\n", fleettext.Safe(args[1])) @@ -625,13 +625,13 @@ func runCommand(ctx context.Context, out io.ReadWriter, perms *ssh.Permissions, } update, err := parseSummary(args[2:]) if err != nil { - fmt.Fprintf(out, "usage: summary SESSION_ID [--purpose text] [summary text]\nerror: %v\n", err) + fmt.Fprintf(out, "usage: summary SESSION_ID [--purpose text] [summary text]\nerror: %s\n", safeError(err)) return 2 } if update.summary == "" && update.purpose == "" { state, err := api.State(ctx) if err != nil { - fmt.Fprintf(out, "error: %v\n", err) + writeError(out, err) return 1 } for _, session := range state.InteractiveSessions { @@ -652,7 +652,7 @@ func runCommand(ctx context.Context, out io.ReadWriter, perms *ssh.Permissions, } session, err := api.UpdateSummary(ctx, args[1], fields) if err != nil { - fmt.Fprintf(out, "error: %v\n", err) + writeError(out, err) return 1 } fleettext.WriteSessionSummary(out, session) @@ -690,6 +690,17 @@ func printHelp(out io.Writer, user fleetapi.User) { fmt.Fprintln(out, " open") } +func writeError(out io.Writer, err error) { + fmt.Fprintf(out, "error: %s\n", safeError(err)) +} + +func safeError(err error) string { + if err == nil { + return "" + } + return fleettext.Safe(err.Error()) +} + func printList(out io.Writer, state fleetapi.State) { fmt.Fprintf( out, diff --git a/cmd/crabbox-ssh-gateway/main_test.go b/cmd/crabbox-ssh-gateway/main_test.go index 815278ba..48b5543e 100644 --- a/cmd/crabbox-ssh-gateway/main_test.go +++ b/cmd/crabbox-ssh-gateway/main_test.go @@ -658,6 +658,56 @@ func TestRunCommandStopsAfterDefaultRepoLookupFailure(t *testing.T) { } } +func TestRunCommandSanitizesControlPlaneErrors(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/ssh/state" { + t.Errorf("path = %q", r.URL.Path) + w.WriteHeader(http.StatusNotFound) + return + } + w.WriteHeader(http.StatusInternalServerError) + _, _ = w.Write([]byte("bad\x1b]52;c;secret\x07state")) + })) + defer server.Close() + + client := &apiClient{baseURL: server.URL, token: "gateway-token", client: server.Client()} + permissions := &ssh.Permissions{Extensions: map[string]string{ + "authorized": "true", + "fingerprint": "SHA256:test", + "login": "operator", + "role": "owner", + }} + var output bytes.Buffer + if exit := runCommand(context.Background(), &output, permissions, client, "whoami", sessionPTY{}); exit != 1 { + t.Fatalf("exit=%d output=%q", exit, output.String()) + } + got := output.String() + if strings.ContainsAny(got, "\x1b\x07") { + t.Fatalf("error output retained terminal controls: %q", got) + } + if !strings.Contains(got, "bad]52;c;secretstate") { + t.Fatalf("error output = %q", got) + } +} + +func TestRunCommandSanitizesLinkURL(t *testing.T) { + permissions := &ssh.Permissions{Extensions: map[string]string{ + "authorized": "false", + "link_url": "https://example.test/link\x1b]52;c;secret\x07", + }} + var output bytes.Buffer + if exit := runCommand(context.Background(), &output, permissions, nil, "whoami", sessionPTY{}); exit != 1 { + t.Fatalf("exit=%d output=%q", exit, output.String()) + } + got := output.String() + if strings.ContainsAny(got, "\x1b\x07") { + t.Fatalf("link output retained terminal controls: %q", got) + } + if !strings.Contains(got, "https://example.test/link]52;c;secret") { + t.Fatalf("link output = %q", got) + } +} + func TestTranscriptCommandSanitizesTerminalControls(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path != "/api/ssh/interactive-sessions/IS-7/transcript" { From 755710a9bdff38153b0a6d065cda85c90d3819dc Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Thu, 18 Jun 2026 18:25:40 +0800 Subject: [PATCH 14/14] fix(cli): sanitize ssh fallback output --- cmd/crabbox-ssh-gateway/main.go | 14 +++++++++++--- cmd/crabbox-ssh-gateway/main_test.go | 23 +++++++++++++++++++++++ cmd/crabfleet/main.go | 2 +- cmd/crabfleet/main_test.go | 14 ++++++++++++-- 4 files changed, 47 insertions(+), 6 deletions(-) diff --git a/cmd/crabbox-ssh-gateway/main.go b/cmd/crabbox-ssh-gateway/main.go index dae62dea..faf37be9 100644 --- a/cmd/crabbox-ssh-gateway/main.go +++ b/cmd/crabbox-ssh-gateway/main.go @@ -332,7 +332,10 @@ func handleSession( Height uint32 Modes string } - ssh.Unmarshal(req.Payload, &payload) + if err := ssh.Unmarshal(req.Payload, &payload); err != nil { + req.Reply(false, nil) + continue + } pty.resize(payload.Cols, payload.Rows, commandStarted) req.Reply(true, nil) case "window-change": @@ -342,7 +345,9 @@ func handleSession( Width uint32 Height uint32 } - ssh.Unmarshal(req.Payload, &payload) + if err := ssh.Unmarshal(req.Payload, &payload); err != nil { + continue + } pty.resize(payload.Cols, payload.Rows, commandStarted) case "shell": if commandStarted { @@ -362,7 +367,10 @@ func handleSession( continue } var payload struct{ Command string } - ssh.Unmarshal(req.Payload, &payload) + if err := ssh.Unmarshal(req.Payload, &payload); err != nil { + req.Reply(false, nil) + continue + } commandStarted = true stopTimer(idleTimer) idle = nil diff --git a/cmd/crabbox-ssh-gateway/main_test.go b/cmd/crabbox-ssh-gateway/main_test.go index 48b5543e..e561806c 100644 --- a/cmd/crabbox-ssh-gateway/main_test.go +++ b/cmd/crabbox-ssh-gateway/main_test.go @@ -514,6 +514,29 @@ func TestIdleSessionChannelClosesWithoutCommand(t *testing.T) { } } +func TestMalformedExecRequestIsRejected(t *testing.T) { + addr, cleanup := serveTestSSHGateway(t, testSSHServerConfig(t), nil) + defer cleanup() + client, err := ssh.Dial("tcp", addr, testSSHClientConfig()) + if err != nil { + t.Fatal(err) + } + defer client.Close() + channel, _, err := client.OpenChannel("session", nil) + if err != nil { + t.Fatal(err) + } + defer channel.Close() + + ok, err := channel.SendRequest("exec", true, []byte{0, 0, 0, 8, 'h', 'e', 'l', 'p'}) + if err != nil { + t.Fatal(err) + } + if ok { + t.Fatal("malformed exec request was accepted") + } +} + func TestRunCommandCancelsControlPlaneRequest(t *testing.T) { entered := make(chan struct{}) cancelled := make(chan struct{}) diff --git a/cmd/crabfleet/main.go b/cmd/crabfleet/main.go index 1310f466..b52cbe2a 100644 --- a/cmd/crabfleet/main.go +++ b/cmd/crabfleet/main.go @@ -204,7 +204,7 @@ func (cmd newCmd) Run(app *cli, api *fleetapi.Client) error { if cmd.VNC { output, captureErr := runSSHCommandOutput(app, args...) if output != "" { - fmt.Fprint(os.Stdout, output) + fmt.Fprint(os.Stdout, fleettext.SafeMultiline(output)) } if captureErr != nil { return captureErr diff --git a/cmd/crabfleet/main_test.go b/cmd/crabfleet/main_test.go index f5ca8a58..06692f57 100644 --- a/cmd/crabfleet/main_test.go +++ b/cmd/crabfleet/main_test.go @@ -432,7 +432,7 @@ func TestValidateWebVNCURL(t *testing.T) { } func TestNewVNCFallbackValidatesCapturedURL(t *testing.T) { - installOutputSSH(t, "vnc: http://example.test/not-webvnc\n") + installOutputSSH(t, "\x1b]52;c;secret\x07\nvnc: http://example.test/not-webvnc\n") listener, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { t.Fatal(err) @@ -448,10 +448,20 @@ func TestNewVNCFallbackValidatesCapturedURL(t *testing.T) { Token: "gateway-token", Fingerprint: "SHA256:test", } - err = (newCmd{Branch: "main", Command: "codex --yolo", Repo: "openclaw/crabfleet", VNC: true}).Run(app, app.apiClient()) + var runErr error + output := captureStdout(t, func() { + runErr = (newCmd{Branch: "main", Command: "codex --yolo", Repo: "openclaw/crabfleet", VNC: true}).Run(app, app.apiClient()) + }) + err = runErr if err == nil || !strings.Contains(err.Error(), "invalid WebVNC URL scheme") { t.Fatalf("error = %v", err) } + if strings.ContainsAny(output, "\x1b\x07") { + t.Fatalf("fallback output retained terminal controls: %q", output) + } + if !strings.Contains(output, "vnc: http://example.test/not-webvnc") { + t.Fatalf("fallback output = %q", output) + } } func TestFirstLineSkipsBlankLines(t *testing.T) {