diff --git a/internal/mcp/tools/spark_submit.go b/internal/mcp/tools/spark_submit.go index e34c013..a88ac3c 100644 --- a/internal/mcp/tools/spark_submit.go +++ b/internal/mcp/tools/spark_submit.go @@ -3,6 +3,9 @@ package tools import ( "context" "fmt" + "log/slog" + "path/filepath" + "strings" "github.com/mark3labs/mcp-go/mcp" @@ -15,7 +18,7 @@ const SparkSubmitName = "spark_submit" // NewSparkSubmitTool returns the schema for the spark_submit MCP Tool. func NewSparkSubmitTool() mcp.Tool { return mcp.NewTool(SparkSubmitName, - mcp.WithDescription("Submit a Spark application to a configured cluster using the cluster's local spark-submit binary. The `args` array is appended to the cluster's default_submit_args; pass one argv element per CLI flag. If the script to run was just uploaded via upload_file, put the `path` field from that response as one of the args — do NOT construct your own path. Cluster-local paths (e.g. /opt/spark/examples/pi.py) are also accepted. Returns {app_id, exit_code, stdout_tail, stderr_tail, duration_ms}. Binary is invoked as a child process — no shell, no command injection."), + mcp.WithDescription("Submit a Spark application to a configured cluster using the cluster's local spark-submit binary. The `args` array is appended to the cluster's default_submit_args; pass one argv element per CLI flag. If the script to run was just uploaded via upload_file, put the `path` field from that response as one of the args — do NOT construct your own path. Cluster-local paths (e.g. /opt/spark/examples/pi.py) are accepted but will be logged as a warning. To avoid the warning, always use the `path` field returned by upload_file. Returns {app_id, exit_code, stdout_tail, stderr_tail, duration_ms}. Binary is invoked as a child process — no shell, no command injection."), mcp.WithString("cluster_id", mcp.Required(), mcp.Description("ID of the configured cluster (from list_clusters)"), @@ -64,7 +67,7 @@ func (d *Deps) SparkSubmitHandler(ctx context.Context, req mcp.CallToolRequest) } fullArgs := append([]string{}, cl.DefaultSubmitArgs...) - fullArgs = append(fullArgs, translateArgs(userArgs, d.UploadStore)...) + fullArgs = append(fullArgs, translateArgs(userArgs, d.UploadStore, d.Logger)...) result, err := executor.Run(ctx, executor.SparkSubmitOpts{ Binary: cl.SparkSubmitExecuteBin, @@ -103,19 +106,49 @@ func (d *Deps) SparkSubmitHandler(ctx context.Context, req mcp.CallToolRequest) // translateArgs canonicalizes server-minted upload paths and leaves // cluster-local paths untouched. It is a code-level guardrail, not a // validator: non-minted absolute paths are intentionally accepted. -func translateArgs(userArgs []string, store *uploads.Store) []string { +func translateArgs(userArgs []string, store *uploads.Store, logger *slog.Logger) []string { if store == nil { return userArgs } out := make([]string, len(userArgs)) for i, arg := range userArgs { - if fileID, err := store.Validate(arg); err == nil { - if canonical, err := store.AbsPath(fileID); err == nil { - out[i] = canonical - continue + pathPart, isPath := pathLikeArg(arg) + if isPath { + if fileID, err := store.Validate(pathPart); err == nil { + if canonical, err := store.AbsPath(fileID); err == nil { + out[i] = canonical + continue + } + } else if logger != nil && filepath.IsAbs(pathPart) { + // Absolute path that this server did not mint. Still allowed, + // but warn so operators can spot LLM-invented remote paths. + logger.Warn("spark_submit.unminted_path", + "path", pathPart, + "note", "not produced by upload_file on this server; cluster-local paths are valid, but if this came from a remote agent the path is unreachable") } } out[i] = arg } return out } + +// pathLikeArg extracts an absolute path from a spark-submit argv element. +// It handles plain absolute paths ("/opt/spark/pi.py") and flag-style values +// ("--conf=/x/y"). Non-path args like "--class" or "Main" return ok=false. +func pathLikeArg(arg string) (string, bool) { + if arg == "" { + return "", false + } + if strings.HasPrefix(arg, "/") { + return arg, true + } + if strings.HasPrefix(arg, "-") { + if eq := strings.Index(arg, "="); eq > 0 && eq < len(arg)-1 { + val := arg[eq+1:] + if strings.HasPrefix(val, "/") { + return val, true + } + } + } + return "", false +} diff --git a/internal/mcp/tools/spark_submit_path_test.go b/internal/mcp/tools/spark_submit_path_test.go index dac36b3..1eb26fa 100644 --- a/internal/mcp/tools/spark_submit_path_test.go +++ b/internal/mcp/tools/spark_submit_path_test.go @@ -1,7 +1,10 @@ package tools import ( + "bytes" "context" + "encoding/json" + "log/slog" "os" "path/filepath" "strings" @@ -70,6 +73,97 @@ func TestSparkSubmit_PathPassthrough(t *testing.T) { } } +func TestTranslateArgs_Logging(t *testing.T) { + deps, _ := testDepsWithDataDir(t) + + t.Run("minted_path_no_warn", func(t *testing.T) { + store := deps.UploadStore + fileID, _, _, _, absPath, err := store.Save([]byte("print('hi')\n"), "hello.py") + if err != nil { + t.Fatalf("save upload: %v", err) + } + + logger, buf := capturingLogger(t) + out := translateArgs([]string{absPath, "--class", "Main"}, deps.UploadStore, logger) + if out[0] != absPath { + t.Errorf("out[0]=%q, want %q", out[0], absPath) + } + + for _, rec := range parseLogRecords(t, buf) { + if rec["msg"] == "spark_submit.unminted_path" { + t.Errorf("unexpected warn for minted path: %v", rec) + } + } + + if !strings.HasSuffix(absPath, "/"+fileID) { + t.Errorf("absPath=%q does not end with fileID %q", absPath, fileID) + } + }) + + t.Run("cluster_local_path_warns", func(t *testing.T) { + logger, buf := capturingLogger(t) + out := translateArgs([]string{"/opt/spark/examples/pi.py"}, deps.UploadStore, logger) + if len(out) != 1 || out[0] != "/opt/spark/examples/pi.py" { + t.Errorf("out=%v, want [/opt/spark/examples/pi.py]", out) + } + + var warns int + for _, rec := range parseLogRecords(t, buf) { + if rec["msg"] == "spark_submit.unminted_path" { + warns++ + if rec["path"] != "/opt/spark/examples/pi.py" { + t.Errorf("warn path=%v, want /opt/spark/examples/pi.py", rec["path"]) + } + } + } + if warns != 1 { + t.Errorf("warns=%d, want 1; log=%s", warns, buf.String()) + } + }) + + t.Run("non_path_args_never_warn", func(t *testing.T) { + logger, buf := capturingLogger(t) + out := translateArgs([]string{"--class", "Main", "--master", "yarn"}, deps.UploadStore, logger) + want := []string{"--class", "Main", "--master", "yarn"} + if len(out) != len(want) { + t.Fatalf("out=%v, want %v", out, want) + } + for i := range out { + if out[i] != want[i] { + t.Errorf("out[%d]=%q, want %q", i, out[i], want[i]) + } + } + + for _, rec := range parseLogRecords(t, buf) { + if rec["msg"] == "spark_submit.unminted_path" { + t.Errorf("unexpected warn for non-path args: %v", rec) + } + } + }) +} + +func capturingLogger(t *testing.T) (*slog.Logger, *bytes.Buffer) { + t.Helper() + buf := &bytes.Buffer{} + return slog.New(slog.NewJSONHandler(buf, &slog.HandlerOptions{Level: slog.LevelDebug})), buf +} + +func parseLogRecords(t *testing.T, buf *bytes.Buffer) []map[string]any { + t.Helper() + var recs []map[string]any + for _, line := range strings.Split(strings.TrimSpace(buf.String()), "\n") { + if line == "" { + continue + } + var rec map[string]any + if err := json.Unmarshal([]byte(line), &rec); err != nil { + t.Fatalf("invalid log JSON: %q (err=%v)", line, err) + } + recs = append(recs, rec) + } + return recs +} + func TestSparkSubmit_ClusterLocalPathPassthrough(t *testing.T) { deps, repo := testDepsWithDataDir(t) deps.SparkSubmitTimeout = 5 * time.Second