You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by li...@apache.org on 2026/10/10 06:38:37 UTC
(rocketmq-dashboard) branch rocketmq-studio updated: fix(rmqctl): three CLI defects in nested cells, confirm-token replay and stdio error frames (#5643)
This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new b205d5f4f fix(rmqctl): three CLI defects in nested cells, confirm-token replay and stdio error frames (#5643)
b205d5f4f is described below
commit b205d5f4fa9b78cf912fc389074129c3d7d394b5
Author: Apulupie <16...@users.noreply.github.com>
AuthorDate: Sat Oct 10 14:38:31 2026 +0800
fix(rmqctl): three CLI defects in nested cells, confirm-token replay and stdio error frames (#5643)
Three rmqctl defects.
- Nested objects and arrays in table cells were rendered with Go's `map[k:v]` syntax instead of the JSON
the server sent; `stringify` is the single funnel, so one change covers every column. (#5643)
- A confirm token could be replayed with an auto-filled timestamp: the signing side does not include
`timestamp` in `CONTROL_FIELDS` while the filling side refreshes it per call, so the guard now
rejects that combination. (#5654)
- The stdio proxy's fallback error frame dropped the underlying cause; it now reuses the existing
`normalizeCLIError` / `safeStdioDiagnostic` path. (#5656)
`go test ./...` green across all seven rmqctl packages, `make fmt` clean. Rebased onto today's
#6057 (`fix(rmqctl): restore numeric types before YAML encoding`), which the yaml golden test needs.
Folded in #5654 and #5656 (same author).
---
rmqctl/cmd/catalog.go | 33 ++++++-
rmqctl/cmd/catalog_instance_test.go | 68 ++++++++++++++
rmqctl/cmd/mcp_stdio.go | 23 +++++
rmqctl/cmd/mcp_stdio_error_test.go | 166 ++++++++++++++++++++++++++++++++++
rmqctl/internal/output/output.go | 9 ++
rmqctl/internal/output/output_test.go | 31 +++++++
6 files changed, 327 insertions(+), 3 deletions(-)
diff --git a/rmqctl/cmd/catalog.go b/rmqctl/cmd/catalog.go
index 0f230cb51..7f1dddb11 100644
--- a/rmqctl/cmd/catalog.go
+++ b/rmqctl/cmd/catalog.go
@@ -192,7 +192,7 @@ func runTool(
// Fill client-side defaults (x-client-default: NOW) before required
// validation and argument assembly so a schema-required field with a
// client default is self-consistent. Explicit flags always win.
- applyClientDefaults(tool, arguments)
+ filled := applyClientDefaults(tool, arguments)
// Pass the explicit --instance-id value through to tools whose schema
// declares the instance identifier (decision 7: pass-through of the
// caller's explicit value, not a default injection). Platform-level
@@ -203,6 +203,9 @@ func runTool(
if err := validateSchemaArguments(tool, tool.InputSchema, arguments); err != nil {
return err
}
+ if err := rejectReplayWithAutoFilledDefaults(arguments, filled); err != nil {
+ return err
+ }
if err := confirmRisk(cmd, runtime, tool, arguments); err != nil {
return err
@@ -264,8 +267,10 @@ var nowMillis = func() int64 { return time.Now().UnixMilli() }
// Filling happens before required validation so required fields with a client
// default (e.g. group reset-offset --timestamp) validate cleanly. The catalog
// generator rejects x-client-default on nested properties, so only top-level
-// fields need to be considered.
-func applyClientDefaults(tool toolcatalog.Tool, arguments map[string]any) {
+// fields need to be considered. The filled fields are returned so a caller can
+// tell an auto-filled value from an explicitly supplied one.
+func applyClientDefaults(tool toolcatalog.Tool, arguments map[string]any) []toolcatalog.Field {
+ var filled []toolcatalog.Field
for _, field := range tool.InputSchema.Fields {
if field.ClientDefault != toolcatalog.ClientDefaultNow {
continue
@@ -274,7 +279,29 @@ func applyClientDefaults(tool toolcatalog.Tool, arguments map[string]any) {
continue
}
arguments[field.Name] = nowMillis()
+ filled = append(filled, field)
+ }
+ return filled
+}
+
+// rejectReplayWithAutoFilledDefaults refuses a --confirm-token replay whose
+// previewed input cannot be reproduced. The server signs the previewed
+// business input into the token (the timestamp of a group reset-offset
+// preview included), while an x-client-default: NOW field is filled on every
+// invocation — so a replay that omits the flag carries a fresh value, never
+// matches the preview, and is answered with a mismatch error whose hint
+// ("without changing the operation input") the caller did obey. Naming the
+// flag to pin turns that dead end into an instruction.
+func rejectReplayWithAutoFilledDefaults(arguments map[string]any, filled []toolcatalog.Field) error {
+ if len(filled) == 0 || !argumentPresent(arguments, "confirm_token") {
+ return nil
}
+ field := filled[0]
+ return types.NewCLIError(types.CodeInvalidArgument,
+ fmt.Sprintf("--confirm-token cannot replay an auto-filled --%s", field.Flag),
+ fmt.Sprintf("The preview signed its own --%s into the token, and this call filled a new value, so the server "+
+ "would reject it as a preview mismatch. Pin the previewed value with --%s <value from the preview plan>, or "+
+ "drop --confirm-token to let --yes run the preview and the call together.", field.Flag, field.Flag))
}
func schemaFlagUsage(field toolcatalog.Field) string {
diff --git a/rmqctl/cmd/catalog_instance_test.go b/rmqctl/cmd/catalog_instance_test.go
index 8fbc3a6ca..e8fdbb59a 100644
--- a/rmqctl/cmd/catalog_instance_test.go
+++ b/rmqctl/cmd/catalog_instance_test.go
@@ -19,8 +19,10 @@ package cmd
import (
"bytes"
"encoding/json"
+ "errors"
"net/http"
"net/http/httptest"
+ "strings"
"testing"
"time"
@@ -209,3 +211,69 @@ func TestClientDefaultNowSatisfiesRequiredValidation(t *testing.T) {
t.Fatalf("arguments = %#v, want the filled timestamp", request.Arguments)
}
}
+
+func syntheticConfirmedResetTool() toolcatalog.Tool {
+ return toolcatalog.Tool{
+ Name: "rmq.synthetic.confirmed-reset",
+ CLI: toolcatalog.CLI{Resource: "synthetic-confirmed-reset", Verb: "run"},
+ Description: "Synthetic L2 tool combining an x-client-default NOW field with the two-phase handshake.",
+ RiskLevel: "L2",
+ Permission: "synthetic:write",
+ RequiredCapabilities: []string{},
+ InputSchema: toolcatalog.InputSchema{
+ Fields: []toolcatalog.Field{
+ {Name: "instanceId", Flag: "instance-id", Kind: toolcatalog.StringField, Required: true, MinLength: 1},
+ {Name: "groupName", Flag: "group-name", Kind: toolcatalog.StringField, Required: true, MinLength: 1},
+ {Name: "timestamp", Flag: "timestamp", Kind: toolcatalog.IntegerField, Required: true, ClientDefault: toolcatalog.ClientDefaultNow},
+ {Name: "dry_run", Flag: "dry-run", Kind: toolcatalog.BooleanField},
+ {Name: "confirm_token", Flag: "confirm-token", Kind: toolcatalog.StringField, MinLength: 1},
+ },
+ },
+ ViewHint: "object",
+ }
+}
+
+// TestConfirmTokenReplayWithAutoFilledTimestampIsRefused verifies the
+// two-phase handshake against the server's signing rules: the token signs the
+// previewed business input, including the timestamp of a group reset-offset
+// preview, while an x-client-default: NOW field is filled on every invocation.
+// A replay that carries a --confirm-token but omits the flag therefore cannot
+// ever match its preview, so the CLI refuses it up front and names the flag to
+// pin instead of sending a call the server is guaranteed to answer with a
+// preview-mismatch error.
+func TestConfirmTokenReplayWithAutoFilledTimestampIsRefused(t *testing.T) {
+ const pinnedNow = int64(1757600000000)
+ original := nowMillis
+ nowMillis = func() int64 { return pinnedNow }
+ defer func() { nowMillis = original }()
+
+ request, stderr, err := executeSyntheticTool(t, syntheticConfirmedResetTool(),
+ "--group-name", "gid-orders", "--confirm-token", "server-issued", "--yes")
+ if err == nil {
+ t.Fatalf("a confirm-token replay with an auto-filled timestamp must be refused, request=%#v", request)
+ }
+ var cliError *types.CLIError
+ if !errors.As(err, &cliError) || cliError.Code != types.CodeInvalidArgument {
+ t.Fatalf("err = %v, want INVALID_ARGUMENT", err)
+ }
+ if !strings.Contains(cliError.Hint, "--timestamp") {
+ t.Fatalf("hint = %q, want it to name the flag to pin", cliError.Hint)
+ }
+ if request.Name != "" {
+ t.Fatalf("refused replay must not reach the server, observed %q, stderr=%s", request.Name, stderr)
+ }
+}
+
+// TestExplicitTimestampStillReplaysConfirmToken verifies the refusal is
+// specific to the auto-filled value: pinning the previewed timestamp keeps the
+// token usable, which is the path the hint points at.
+func TestExplicitTimestampStillReplaysConfirmToken(t *testing.T) {
+ request, stderr, err := executeSyntheticTool(t, syntheticConfirmedResetTool(),
+ "--group-name", "gid-orders", "--timestamp", "1791394886405", "--confirm-token", "server-issued", "--yes")
+ if err != nil {
+ t.Fatalf("an explicit timestamp must keep the token usable: %v, stderr=%s", err, stderr)
+ }
+ if request.Arguments["confirm_token"] != "server-issued" {
+ t.Fatalf("arguments = %#v, want the confirm token to be sent", request.Arguments)
+ }
+}
diff --git a/rmqctl/cmd/mcp_stdio.go b/rmqctl/cmd/mcp_stdio.go
index f5f96d5b4..087197e36 100644
--- a/rmqctl/cmd/mcp_stdio.go
+++ b/rmqctl/cmd/mcp_stdio.go
@@ -306,6 +306,29 @@ func writeStdioError(out io.Writer, errOut io.Writer, requestLine string, callEr
Hint: hint,
},
}
+ } else {
+ // Connection refusals, TLS failures, timeouts and protocol-decode errors carry no HTTP
+ // status, but the caller still needs the cause: an MCP client that only receives
+ // "stdio proxy call failed" cannot tell the user (or the model) anything but "go look at
+ // the logs". normalizeCLIError classifies them the same way the CLI's own error path
+ // does, and the sanitized text travels in both message and data so the frame is
+ // actionable on its own.
+ cliError := normalizeCLIError(callErr)
+ code := safeStdioDiagnostic(cliError.Code)
+ message := safeStdioDiagnostic(cliError.Message)
+ hint := safeStdioDiagnostic(cliError.Hint)
+ if message == "" {
+ message = "stdio proxy call failed"
+ }
+ rpcError = jsonRPCError{
+ Code: jsonrpcServerErrorCode,
+ Message: message,
+ Data: &jsonRPCErrorData{
+ Code: code,
+ Message: message,
+ Hint: hint,
+ },
+ }
}
errorPayload := jsonRPCErrorPayload{
JSONRPC: "2.0",
diff --git a/rmqctl/cmd/mcp_stdio_error_test.go b/rmqctl/cmd/mcp_stdio_error_test.go
new file mode 100644
index 000000000..c8ac48bef
--- /dev/null
+++ b/rmqctl/cmd/mcp_stdio_error_test.go
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package cmd
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "net"
+ "strings"
+ "testing"
+
+ "github.com/apache/rocketmq-dashboard/rmqctl/internal/studio"
+)
+
+func decodeStdioErrorFrame(t *testing.T, out *bytes.Buffer) jsonRPCErrorPayload {
+ t.Helper()
+ frameLine := strings.TrimSpace(out.String())
+ if frameLine == "" {
+ t.Fatal("no error frame was written to stdout")
+ }
+ if strings.Contains(frameLine, "\n") {
+ t.Fatalf("stdout carries more than one line: %q", frameLine)
+ }
+ var payload jsonRPCErrorPayload
+ if err := json.Unmarshal([]byte(frameLine), &payload); err != nil {
+ t.Fatalf("stdout frame is not JSON: %v (%q)", err, frameLine)
+ }
+ return payload
+}
+
+// TestStdioErrorFrameKeepsTheFailureCause verifies that a call failure without
+// an HTTP status still tells the MCP client what went wrong: the classified
+// code, the sanitized cause and a hint travel in the frame's data, instead of
+// collapsing into a bare "stdio proxy call failed".
+func TestStdioErrorFrameKeepsTheFailureCause(t *testing.T) {
+ connectionRefused := &net.OpError{Op: "dial", Net: "tcp", Err: errors.New("connect: connection refused")}
+ cases := []struct {
+ name string
+ callErr error
+ wantCode string
+ wantMessage string
+ }{
+ {
+ name: "transport failure",
+ callErr: fmt.Errorf("failed to send request: %w", connectionRefused),
+ wantCode: "UNAVAILABLE",
+ wantMessage: "connection refused",
+ },
+ {
+ name: "deadline exceeded",
+ callErr: fmt.Errorf("mcp call: %w", context.DeadlineExceeded),
+ wantCode: "TIMEOUT",
+ wantMessage: "deadline exceeded",
+ },
+ {
+ name: "protocol decode failure",
+ callErr: errors.New("decode MCP JSON-RPC response: unexpected end of JSON input"),
+ wantCode: "COMMAND_FAILED",
+ wantMessage: "unexpected end of JSON input",
+ },
+ }
+ for _, testCase := range cases {
+ t.Run(testCase.name, func(t *testing.T) {
+ out := &bytes.Buffer{}
+ errOut := &bytes.Buffer{}
+ writeStdioError(out, errOut, `{"jsonrpc":"2.0","id":7,"method":"tools/call"}`, testCase.callErr)
+
+ frame := decodeStdioErrorFrame(t, out)
+ if frame.Error.Code != jsonrpcServerErrorCode {
+ t.Fatalf("frame code = %d, want %d", frame.Error.Code, jsonrpcServerErrorCode)
+ }
+ if frame.Error.Data == nil {
+ t.Fatal("error frame carries no data payload")
+ }
+ if frame.Error.Data.Code != testCase.wantCode {
+ t.Fatalf("data.code = %q, want %q", frame.Error.Data.Code, testCase.wantCode)
+ }
+ if !strings.Contains(frame.Error.Message, testCase.wantMessage) {
+ t.Fatalf("message = %q, want it to contain %q", frame.Error.Message, testCase.wantMessage)
+ }
+ if !strings.Contains(frame.Error.Data.Message, testCase.wantMessage) {
+ t.Fatalf("data.message = %q, want it to contain %q", frame.Error.Data.Message, testCase.wantMessage)
+ }
+ if frame.Error.Data.Hint == "" {
+ t.Fatal("data.hint is empty, so the caller has nothing to act on")
+ }
+ if !strings.Contains(errOut.String(), testCase.wantMessage) {
+ t.Fatalf("stderr = %q, want the cause to stay there too", errOut.String())
+ }
+ })
+ }
+}
+
+// TestStdioErrorFrameKeepsTheHTTPStatusShape pins the branch that already
+// worked: a structured MCP HTTP error keeps its own code, message and hint.
+func TestStdioErrorFrameKeepsTheHTTPStatusShape(t *testing.T) {
+ out := &bytes.Buffer{}
+ errOut := &bytes.Buffer{}
+ writeStdioError(out, errOut, `{"jsonrpc":"2.0","id":"srv-1","method":"tools/call"}`,
+ &studio.MCPHTTPStatusError{StatusCode: 403, Code: "FORBIDDEN", Message: "instance mismatch", Hint: "check the instance id"})
+
+ frame := decodeStdioErrorFrame(t, out)
+ if frame.Error.Code != jsonrpcServerErrorCode {
+ t.Fatalf("frame code = %d, want %d", frame.Error.Code, jsonrpcServerErrorCode)
+ }
+ if frame.Error.Data == nil || frame.Error.Data.Code != "FORBIDDEN" {
+ t.Fatalf("data = %#v, want the server's FORBIDDEN code", frame.Error.Data)
+ }
+ if frame.Error.Message != "instance mismatch" || frame.Error.Data.Hint != "check the instance id" {
+ t.Fatalf("frame = %#v, want the server's message and hint", frame.Error)
+ }
+}
+
+// TestStdioErrorFrameSanitizesAndCapsTheCause verifies the frame stays a
+// single line and within the diagnostic cap even for a multi-line, oversized
+// failure text.
+func TestStdioErrorFrameSanitizesAndCapsTheCause(t *testing.T) {
+ out := &bytes.Buffer{}
+ errOut := &bytes.Buffer{}
+ writeStdioError(out, errOut, `{"jsonrpc":"2.0","id":1,"method":"tools/call"}`,
+ fmt.Errorf("first line\r\nsecond\tline %s", strings.Repeat("x", 4*stdioDiagnosticLimit)))
+
+ frame := decodeStdioErrorFrame(t, out)
+ if len(frame.Error.Message) > stdioDiagnosticLimit {
+ t.Fatalf("message length = %d, want at most %d", len(frame.Error.Message), stdioDiagnosticLimit)
+ }
+ if strings.ContainsAny(frame.Error.Message, "\r\n\t") {
+ t.Fatalf("message = %q, want control characters flattened", frame.Error.Message)
+ }
+ if !strings.Contains(frame.Error.Message, "first line") || !strings.Contains(frame.Error.Message, "second line") {
+ t.Fatalf("message = %q, want the sanitized cause", frame.Error.Message)
+ }
+}
+
+// TestStdioErrorFrameSkipsUnparsableRequests pins the deliberate part of the
+// current behaviour: without a request id there is nothing a JSON-RPC error
+// frame could address, so only stderr carries the diagnostic.
+func TestStdioErrorFrameSkipsUnparsableRequests(t *testing.T) {
+ out := &bytes.Buffer{}
+ errOut := &bytes.Buffer{}
+ writeStdioError(out, errOut, "not json", errors.New("connect: connection refused"))
+
+ if out.Len() != 0 {
+ t.Fatalf("stdout = %q, want no frame for an unparsable request", out.String())
+ }
+ if !strings.Contains(errOut.String(), "connection refused") {
+ t.Fatalf("stderr = %q, want the diagnostic", errOut.String())
+ }
+}
diff --git a/rmqctl/internal/output/output.go b/rmqctl/internal/output/output.go
index 90ec62bb6..4860c933c 100644
--- a/rmqctl/internal/output/output.go
+++ b/rmqctl/internal/output/output.go
@@ -213,5 +213,14 @@ func stringify(value any) string {
if f, ok := value.(float64); ok {
return strconv.FormatFloat(f, 'f', -1, 64)
}
+ // Nested objects and arrays decoded from the tool output are JSON to begin with;
+ // render them as JSON instead of Go's default map/slice syntax ("map[k:v]"),
+ // which is noisy and not what the server sent.
+ switch value.(type) {
+ case map[string]any, []any:
+ if encoded, err := json.Marshal(value); err == nil {
+ return string(encoded)
+ }
+ }
return fmt.Sprint(value)
}
diff --git a/rmqctl/internal/output/output_test.go b/rmqctl/internal/output/output_test.go
index e6f9b5e13..28e517647 100644
--- a/rmqctl/internal/output/output_test.go
+++ b/rmqctl/internal/output/output_test.go
@@ -97,3 +97,34 @@ func TestRowsEscapesControlCharactersInCellValues(t *testing.T) {
t.Fatalf("data line does not contain escaped value: %q", data)
}
}
+
+func TestRowsRendersNestedValuesAsJSON(t *testing.T) {
+ buf := &bytes.Buffer{}
+ rows := []map[string]any{
+ {
+ "name": "group-a",
+ "quota": map[string]any{"max": float64(10), "used": float64(3)},
+ "tags": []any{"a", "b"},
+ },
+ }
+ columns := []Column{
+ {Header: "NAME", Key: "name"},
+ {Header: "QUOTA", Key: "quota"},
+ {Header: "TAGS", Key: "tags"},
+ }
+ if err := Rows(buf, rows, columns); err != nil {
+ t.Fatal(err)
+ }
+ data := strings.Split(strings.TrimSuffix(buf.String(), "\n"), "\n")[1]
+ // Go's default map/slice formatting is noise ("map[max:10 used:3]"), and nested
+ // values are JSON to begin with, so the cell should carry the JSON form instead.
+ if strings.Contains(data, "map[") || strings.Contains(data, "[a b]") {
+ t.Fatalf("data line renders Go syntax for nested values: %q", data)
+ }
+ if !strings.Contains(data, `"max":10`) {
+ t.Fatalf("data line does not render the nested object as JSON: %q", data)
+ }
+ if !strings.Contains(data, `["a","b"]`) {
+ t.Fatalf("data line does not render the nested array as JSON: %q", data)
+ }
+}