From c9e54a3f278e9be567047afbfc812151198b3d41 Mon Sep 17 00:00:00 2001 From: "tao.chen" <93983997+taochen-ct@users.noreply.github.com> Date: Mon, 13 Jul 2026 19:49:26 +0800 Subject: [PATCH] =?UTF-8?q?tools:=20=E5=BC=BA=E5=88=B6=20spark=5Fsubmit=20?= =?UTF-8?q?=E7=9A=84=20script=5Fpath=20=E5=BF=85=E9=A1=BB=E6=9D=A5?= =?UTF-8?q?=E8=87=AA=20upload=5Ffile?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 在 spark_submit handler 中增加 server-minted 校验：script_path 必须通过 本服务器 upload_file 的 UploadStore.Validate，否则立即返回清晰错误，避免 远程 agent 把自身本地路径转发给服务器侧 spark-submit 时出现难以定位的 FileNotFoundException。 校验对 nil UploadStore 是安全的：当 d.UploadStore 为 nil 时直接跳过，保留 不强制依赖 store 的测试或降级场景。当前 testDepsWithDataDir 始终会注入 UploadStore，所以测试走的是真实校验路径。 同步更新了 script_path 的工具描述，明确声明非 upload_file 铸造的路径会被 拒绝。这是对 7920dc9 中“path 永远在最后”规则的收紧——现在不仅位置固定， 而且必须是本机 uploads 目录下的有效上传文件。 测试调整： - 新增 TestSparkSubmit_RejectsNonMintedPath：使用 t.TempDir() 下未通过 upload_file 写入的文件作为 script_path，断言返回错误并包含 "not from upload_file"。 - TestSparkSubmit_MissingRequiredField / TestSparkSubmit_BadSparkConfValue 原来使用 "/tmp/script.py" 作为占位路径，现在会先通过 UploadStore.Save 生成一个铸造路径，确保它们继续分别验证 master 缺失和 spark_conf 值类型 错误，而不是被新的路径守卫拦截。 Co-Authored-By: Claude --- internal/mcp/tools/spark_submit.go | 12 ++++- internal/mcp/tools/spark_submit_path_test.go | 48 +++++++++++++++++++- 2 files changed, 57 insertions(+), 3 deletions(-) diff --git a/internal/mcp/tools/spark_submit.go b/internal/mcp/tools/spark_submit.go index 3b3883d..3c79078 100644 --- a/internal/mcp/tools/spark_submit.go +++ b/internal/mcp/tools/spark_submit.go @@ -32,7 +32,7 @@ func NewSparkSubmitTool() mcp.Tool { ), mcp.WithString("script_path", mcp.Required(), - mcp.Description("Absolute path to the script. Pass the path field returned by upload_file — do not construct your own path. The script is always the last argv element."), + mcp.Description("Absolute path to the script. Pass the `path` field returned by upload_file — do not construct your own path. The script is always the last argv element. Paths not minted by upload_file on this server are rejected."), ), mcp.WithString("queue", mcp.Required(), @@ -79,6 +79,16 @@ func (d *Deps) SparkSubmitHandler(ctx context.Context, req mcp.CallToolRequest) if err != nil { return errResult("spark_submit: " + err.Error()), nil } + + // Reject paths that didn't come through upload_file on this server. + // Cluster-local paths and arbitrary host paths are unreachable from the MCP + // server's spark-submit, so we fail fast with a clear message rather than + // letting spark-submit produce a confusing FileNotFoundException later. + if d.UploadStore != nil { + if _, err := d.UploadStore.Validate(scriptPath); err != nil { + return errResult(fmt.Errorf("spark_submit: script_path %q is not from upload_file on this server; upload the file first via upload_file and pass back the path field: %w", scriptPath, err).Error()), nil + } + } queue, err := req.RequireString("queue") if err != nil { return errResult("spark_submit: " + err.Error()), nil diff --git a/internal/mcp/tools/spark_submit_path_test.go b/internal/mcp/tools/spark_submit_path_test.go index e8cc638..5cdfff9 100644 --- a/internal/mcp/tools/spark_submit_path_test.go +++ b/internal/mcp/tools/spark_submit_path_test.go @@ -97,11 +97,16 @@ func TestSparkSubmit_StructuredCommand(t *testing.T) { func TestSparkSubmit_MissingRequiredField(t *testing.T) { deps, _ := testDepsWithDataDir(t) + store := deps.UploadStore + _, _, _, _, mintedPath, err := store.Save([]byte("# dummy\n"), "dummy.py") + if err != nil { + t.Fatalf("save upload: %v", err) + } req := newToolRequest(SparkSubmitName, map[string]any{ "cluster_id": "cluster-echo", "deploy_mode": "cluster", - "script_path": "/tmp/script.py", + "script_path": mintedPath, "queue": "default", "executor_memory": "2G", "executor_cores": 1, @@ -125,12 +130,17 @@ func TestSparkSubmit_MissingRequiredField(t *testing.T) { func TestSparkSubmit_BadSparkConfValue(t *testing.T) { deps, _ := testDepsWithDataDir(t) + store := deps.UploadStore + _, _, _, _, mintedPath, err := store.Save([]byte("# dummy\n"), "dummy.py") + if err != nil { + t.Fatalf("save upload: %v", err) + } req := newToolRequest(SparkSubmitName, map[string]any{ "cluster_id": "cluster-echo", "master": "yarn", "deploy_mode": "cluster", - "script_path": "/tmp/script.py", + "script_path": mintedPath, "queue": "default", "executor_memory": "2G", "executor_cores": 1, @@ -152,3 +162,37 @@ func TestSparkSubmit_BadSparkConfValue(t *testing.T) { t.Errorf("error text=%q, want mention of spark_conf", text.Text) } } + +func TestSparkSubmit_RejectsNonMintedPath(t *testing.T) { + deps, _ := testDepsWithDataDir(t) + + unminted := filepath.Join(t.TempDir(), "unminted.py") + if err := os.WriteFile(unminted, []byte("print('not from upload_file')\n"), 0o644); err != nil { + t.Fatalf("write unminted file: %v", err) + } + + req := newToolRequest(SparkSubmitName, map[string]any{ + "cluster_id": "cluster-echo", + "master": "yarn", + "deploy_mode": "cluster", + "script_path": unminted, + "queue": "default", + "executor_memory": "2G", + "executor_cores": 1, + "num_executors": 4, + }) + res, err := deps.SparkSubmitHandler(context.Background(), req) + if err != nil { + t.Fatalf("handler error: %v", err) + } + if !res.IsError { + t.Fatalf("expected error result, got: %v", res.Content) + } + text, ok := mcp.AsTextContent(res.Content[0]) + if !ok { + t.Fatalf("content is not text: %T", res.Content[0]) + } + if !strings.Contains(text.Text, "not from upload_file") { + t.Errorf("error text=%q, want mention of not from upload_file", text.Text) + } +}