From 05a6e8cc2d9d419815f49c543ab0825998874f3a Mon Sep 17 00:00:00 2001 From: mptyl Date: Tue, 11 Aug 2026 00:58:47 +0200 Subject: [PATCH] fix: harden workspace preprocessing contract --- docs/contracts/workspace-preprocessing-cli.md | 14 +- tools/thothctl/cmd/thothctl/main.go | 13 +- tools/thothctl/cmd/thothctl/main_test.go | 25 +++ .../internal/compose/process_windows.go | 68 +++++++- .../internal/compose/process_windows_test.go | 23 +++ tools/thothctl/internal/safeio/files_unix.go | 8 +- .../internal/safeio/files_unix_test.go | 48 ++++++ .../thothctl/internal/safeio/files_windows.go | 39 ++++- .../internal/safeio/files_windows_test.go | 6 + .../internal/workspaceops/operations.go | 119 ++++++++++++- .../internal/workspaceops/operations_test.go | 163 ++++++++++++++++++ 11 files changed, 494 insertions(+), 32 deletions(-) create mode 100644 tools/thothctl/internal/compose/process_windows_test.go diff --git a/docs/contracts/workspace-preprocessing-cli.md b/docs/contracts/workspace-preprocessing-cli.md index 0d9ff37a..a21bc44c 100644 --- a/docs/contracts/workspace-preprocessing-cli.md +++ b/docs/contracts/workspace-preprocessing-cli.md @@ -14,8 +14,16 @@ workspace preprocess run --workspace ID [--resume RUN] [--json] IDs are lowercase `[a-z][a-z0-9-]{2,62}` and outer run IDs are exactly 32 lowercase hexadecimal characters. `--from-sql` accepts at most 32 canonical regular non-symlink files, each at most 1 MiB and 16 MiB total. `--assume` accepts at most 256 values of 256 bytes, each matching `column=table`. Annotations are one canonical UTF-8 file at most 16 MiB. Existing output, links, hardlinks, directories, replacement races, and reparse points are refused with a generic unsafe-file error. -The bounded schema-v1 stdin request contains only operation-appropriate fields. SQL and annotations are represented by a logical basename, base64 bytes, and a declared `sha256:` digest; no host path or raw content enters the request. The complete request is at most 1 MiB, has exactly one JSON value, and cannot replace command-derived fields. Candidate responses may carry the internal `hostExport` (`mediaType`, `sha256`, `contentBase64`) only for `suggest-fks`; it is verified (UTF-8 YAML, digest, and 700 KiB maximum), removed from the public result, and written exclusively only after result/run/identity validation. +The bounded schema-v1 stdin envelope is exact: omitted fields are not equivalent to explicit zero, empty, or `null` fields. Its fields are, per operation (in addition to the always-required `schemaVersion:1`, `operation`, and `workspaceId`): -The public result has schema version 1 and only these fields: `status`, `code`, workspace/revision/descriptor/operation identities, optional run and child run IDs, completed stages, counts, artifact identities, and warnings. Revisions/descriptors are 40-hex; run IDs are 32-hex; artifact digests are `sha256:`. A nonzero child exit is accepted only for the matching blocked (3) or failed (1) result. Stdout is capped at 1 MiB and stderr at 64 KiB. Compose is invoked only as `compose run --rm --no-deps --no-TTY workspace-maintenance ...`; output never includes child stderr or secrets. +* `inspect`: no additional fields. +* `dwh`: `resume` only when `--resume RUN` was supplied. +* `run`: `resume` only when `--resume RUN` was supplied. +* `evidence`: `resume` only when supplied, and `dryRun` only when `--dry-run` was supplied. +* `suggest-fks`: `sql` only when one or more `--from-sql` files were supplied, and `assume` only when one or more `--assume` values were supplied. Each `sql` item is exactly `{basename, contentBase64, sha256}`. +* `check`: `resume` is required; `annotations` and `reviewedCandidates` are either both present or both absent. `annotations` is exactly `{basename, contentBase64, sha256}`. +* `index-schema`: no additional fields. -Exit 0 means succeeded, unchanged, or dry-run; exit 3 means an expected operator checkpoint/block; exit 2 means grammar or unsafe local-file failure; exit 1 means operational failure. Human mode prints only allowlisted identity/status fields. For `registry_bootstrap_recovery_conflict` it prints exactly: `Bootstrap recovery is ambiguous or corrupt; inspect the installation registry jobs.` +No other fields, duplicate JSON value, host path, raw SQL, or raw annotation content are accepted. SQL and annotations use a logical basename, base64 bytes, and a declared `sha256:` digest. The complete request is at most 1 MiB and every supplied field/value must exactly match the command-derived envelope. Candidate responses may carry the internal `hostExport` (`mediaType`, `sha256`, `contentBase64`) only for `suggest-fks`; its object is strict (unknown fields rejected) and is always verified, even without `--output`: YAML media type (`application/yaml` or `text/yaml`), UTF-8, digest, and decoded size at most 700 KiB. It is removed from the public result and written exclusively only after result/run/identity validation. Output publication uses restrictive mode `0600` and refuses existing leaves, links, hardlinks, directories, replacement races, and reparse points. + +The public result has schema version 1 and only these fields: `status`, `code`, workspace/revision/descriptor/operation identities, optional run and child run IDs, completed stages, counts, artifact identities, and warnings. Revisions/descriptors are 40-hex; run IDs are 32-hex; artifact digests are `sha256:`. Allowed statuses are `succeeded`, `unchanged`, `dry_run`, `blocked`, and `failed`. Allowed codes are `ok`, `workspace_not_found`, `workspace_not_activatable`, `binding_missing`, `preprocessing_conflict`, `preprocessing_resume_mismatch`, `manual_review_required`, `evidence_materialization_required`, `effective_config_mismatch`, `semantic_index_incompatible`, `annotation_invalid`, `egress_policy_refused`, and `registry_bootstrap_recovery_conflict`. `succeeded`, `unchanged`, and `dry_run` require `ok` and child exit 0. `blocked` requires one of `manual_review_required`, `evidence_materialization_required`, `preprocessing_conflict`, `preprocessing_resume_mismatch`, or `registry_bootstrap_recovery_conflict`, and child exit 3. `failed` requires a non-`ok` operational code other than those blocked-only codes, and child exit 1. A nonzero child exit is never accepted for another status/code combination. Stdout is capped at 1 MiB and stderr at 64 KiB. Compose is invoked only as `compose run --rm --no-deps --no-TTY workspace-maintenance ...`; output never includes child stderr or secrets. diff --git a/tools/thothctl/cmd/thothctl/main.go b/tools/thothctl/cmd/thothctl/main.go index afc55f0c..8c614244 100644 --- a/tools/thothctl/cmd/thothctl/main.go +++ b/tools/thothctl/cmd/thothctl/main.go @@ -95,7 +95,7 @@ func run(ctx context.Context, args []string, stdout, stderr io.Writer) int { } result, operationErr := workspaceops.Run(ctx, installation, runner, workspaceCommand, nil) if operationErr != nil { - if errors.Is(operationErr, compose.ErrOutputLimit) || strings.Contains(operationErr.Error(), "unsafe") || strings.Contains(operationErr.Error(), "exceeds") || strings.Contains(operationErr.Error(), "invalid workspace") { + if workspaceUsageError(operationErr) { return commandUsageError(stderr, operationErr.Error()) } fmt.Fprintf(stderr, "thothctl: %s\n", output.Sanitize(operationErr.Error(), secretValues)) @@ -210,6 +210,17 @@ func run(ctx context.Context, args []string, stdout, stderr io.Writer) int { return writeResult(result, err, secretValues, stdout, stderr) } +func workspaceUsageError(err error) bool { + if err == nil || errors.Is(err, compose.ErrOutputLimit) { + return false + } + message := err.Error() + // These are host-side grammar/local-file failures. Child envelope/result and + // bounded-stream failures are operational and deliberately remain exit 1. + return strings.Contains(message, "unsafe") || strings.Contains(message, "request exceeds") || + strings.Contains(message, "SQL input exceeds") || strings.Contains(message, "annotation input exceeds") +} + func renderWorkspaceHuman(w io.Writer, result workspaceops.Result) { if result.Code == workspaceops.CodeRegistryBootstrapRecoveryConflict { fmt.Fprintln(w, "Bootstrap recovery is ambiguous or corrupt; inspect the installation registry jobs.") diff --git a/tools/thothctl/cmd/thothctl/main_test.go b/tools/thothctl/cmd/thothctl/main_test.go index 432bde9b..fa3b7b83 100644 --- a/tools/thothctl/cmd/thothctl/main_test.go +++ b/tools/thothctl/cmd/thothctl/main_test.go @@ -3,6 +3,7 @@ package main import ( "bytes" "context" + "errors" "fmt" "io" "os" @@ -104,6 +105,17 @@ func TestUsageDocumentsClosedConfigureUpdateSourcesAndMaintenanceRecovery(t *tes } } +func TestRunWorkspaceRejectsInvalidCommandBeforeDocker(t *testing.T) { + fixture := newCLIFixture(t, "") + fixture.setEnvironment(t) + var stdout, stderr bytes.Buffer + code := run(context.Background(), []string{"--installation", fixture.installationPath, "workspace", "inspect", "--workspace", "PSd"}, &stdout, &stderr) + if code != 2 || !strings.Contains(stderr.String(), "workspace") { + t.Fatalf("exit=%d stderr=%q", code, stderr.String()) + } + assertDockerNotInvoked(t, fixture) +} + func TestRunSessionsMigrateRequiresExplicitConfirmationBeforeDocker(t *testing.T) { fixture := newCLIFixture(t, "") fixture.setProfile(t, "server") @@ -855,3 +867,16 @@ func TestRenderWorkspaceHumanHidesRecoveryIdentity(t *testing.T) { t.Fatalf("human output = %q", out.String()) } } + +func TestWorkspaceOperationalEnvelopeFailuresAreExitOne(t *testing.T) { + for _, err := range []error{compose.ErrOutputLimit, errors.New("invalid workspace result"), errors.New("invalid workspace request"), errors.New("invalid host export")} { + if workspaceUsageError(err) { + t.Errorf("classified operational error %q as usage", err) + } + } + for _, err := range []error{errors.New("unsafe output file"), errors.New("request exceeds limit"), errors.New("unsafe SQL input")} { + if !workspaceUsageError(err) { + t.Errorf("classified host error %q as operational", err) + } + } +} diff --git a/tools/thothctl/internal/compose/process_windows.go b/tools/thothctl/internal/compose/process_windows.go index 6a34a721..66e208df 100644 --- a/tools/thothctl/internal/compose/process_windows.go +++ b/tools/thothctl/internal/compose/process_windows.go @@ -3,9 +3,12 @@ package compose import ( - "golang.org/x/sys/windows" + "errors" "os/exec" "sync" + "unsafe" + + "golang.org/x/sys/windows" ) var ownedJobs = struct { @@ -13,23 +16,76 @@ var ownedJobs = struct { m map[*exec.Cmd]windows.Handle }{m: make(map[*exec.Cmd]windows.Handle)} +// CREATE_SUSPENDED closes the registration race: no child code can create a +// descendant until the real process HANDLE has been assigned to the job. func configureOwnedProcess(c *exec.Cmd) { - c.SysProcAttr = &windows.SysProcAttr{CreationFlags: windows.CREATE_NEW_PROCESS_GROUP} + c.SysProcAttr = &windows.SysProcAttr{CreationFlags: windows.CREATE_NEW_PROCESS_GROUP | windows.CREATE_SUSPENDED} } + func registerOwnedProcess(c *exec.Cmd) error { - h, err := windows.CreateJobObject(nil, nil) + if c.Process == nil { + return errors.New("owned process has no process handle") + } + job, err := windows.CreateJobObject(nil, nil) if err != nil { return err } - if err = windows.AssignProcessToJobObject(h, windows.Handle(c.Process.Pid)); err != nil { - windows.CloseHandle(h) + closeJob := true + defer func() { + if closeJob { + windows.CloseHandle(job) + } + }() + limits := windows.JOBOBJECT_EXTENDED_LIMIT_INFORMATION{} + limits.BasicLimitInformation.LimitFlags = windows.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE + if _, err = windows.SetInformationJobObject(job, windows.JobObjectExtendedLimitInformation, uintptr(unsafe.Pointer(&limits)), uint32(unsafe.Sizeof(limits))); err != nil { return err } + // c.Process.Pid is used only to obtain a genuine process HANDLE; a PID is + // never passed to AssignProcessToJobObject. + process, err := windows.OpenProcess(windows.PROCESS_SET_QUOTA|windows.PROCESS_TERMINATE|windows.PROCESS_QUERY_LIMITED_INFORMATION, false, uint32(c.Process.Pid)) + if err != nil { + return err + } + defer windows.CloseHandle(process) + if err = windows.AssignProcessToJobObject(job, process); err != nil { + return err + } + thread, err := suspendedPrimaryThread(uint32(c.Process.Pid)) + if err != nil { + _ = windows.TerminateJobObject(job, 1) + return err + } + _, resumeErr := windows.ResumeThread(thread) + windows.CloseHandle(thread) + if resumeErr != nil { + _ = windows.TerminateJobObject(job, 1) + return resumeErr + } ownedJobs.Lock() - ownedJobs.m[c] = h + ownedJobs.m[c] = job ownedJobs.Unlock() + closeJob = false return nil } + +func suspendedPrimaryThread(pid uint32) (windows.Handle, error) { + snapshot, err := windows.CreateToolhelp32Snapshot(windows.TH32CS_SNAPTHREAD, 0) + if err != nil { + return 0, err + } + defer windows.CloseHandle(snapshot) + entry := windows.ThreadEntry32{Size: uint32(unsafe.Sizeof(windows.ThreadEntry32{}))} + err = windows.Thread32First(snapshot, &entry) + for err == nil { + if entry.OwnerProcessID == pid { + return windows.OpenThread(windows.THREAD_SUSPEND_RESUME, false, entry.ThreadID) + } + err = windows.Thread32Next(snapshot, &entry) + } + return 0, err +} + func releaseOwnedProcess(c *exec.Cmd) { ownedJobs.Lock() h := ownedJobs.m[c] diff --git a/tools/thothctl/internal/compose/process_windows_test.go b/tools/thothctl/internal/compose/process_windows_test.go new file mode 100644 index 00000000..bdf050cd --- /dev/null +++ b/tools/thothctl/internal/compose/process_windows_test.go @@ -0,0 +1,23 @@ +//go:build windows + +package compose + +import ( + "golang.org/x/sys/windows" + "os/exec" + "testing" +) + +func TestOwnedProcessConfigurationStartsSuspended(t *testing.T) { + cmd := exec.Command("cmd.exe") + configureOwnedProcess(cmd) + if cmd.SysProcAttr == nil || cmd.SysProcAttr.CreationFlags&windows.CREATE_SUSPENDED == 0 { + t.Fatal("owned Windows process must start suspended until job assignment") + } +} + +func TestOwnedJobPolicyKillsDescendantsOnClose(t *testing.T) { + if windows.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE == 0 { + t.Fatal("owned job must use kill-on-close policy") + } +} diff --git a/tools/thothctl/internal/safeio/files_unix.go b/tools/thothctl/internal/safeio/files_unix.go index 7662aaf7..cb2d5470 100644 --- a/tools/thothctl/internal/safeio/files_unix.go +++ b/tools/thothctl/internal/safeio/files_unix.go @@ -90,7 +90,7 @@ func writeCanonicalExclusive(path string, contents []byte, mode fs.FileMode) err if err != nil { return ErrUnsafeFile } - defer unix.Close(dir) + defer func() { unix.Close(dir) }() for _, component := range components[:len(components)-1] { next, err := unix.Openat(dir, component, unix.O_RDONLY|unix.O_DIRECTORY|unix.O_CLOEXEC|unix.O_NOFOLLOW, 0) if err != nil { @@ -147,7 +147,7 @@ func validateCanonicalOutputPath(path string) error { if err != nil { return ErrUnsafeFile } - defer unix.Close(dir) + defer func() { unix.Close(dir) }() for _, component := range components[:len(components)-1] { next, err := unix.Openat(dir, component, unix.O_RDONLY|unix.O_DIRECTORY|unix.O_CLOEXEC|unix.O_NOFOLLOW, 0) if err != nil { @@ -173,7 +173,7 @@ func recheckUnixParents(components []string, retained []int) bool { if err != nil { return false } - defer unix.Close(dir) + defer func() { unix.Close(dir) }() for i, component := range components[:len(components)-1] { next, e := unix.Openat(dir, component, unix.O_RDONLY|unix.O_DIRECTORY|unix.O_CLOEXEC|unix.O_NOFOLLOW, 0) if e != nil { @@ -195,7 +195,7 @@ func recheckUnixParentPath(components []string, retained int) bool { if err != nil { return false } - defer unix.Close(dir) + defer func() { unix.Close(dir) }() for _, component := range components { next, e := unix.Openat(dir, component, unix.O_RDONLY|unix.O_DIRECTORY|unix.O_CLOEXEC|unix.O_NOFOLLOW, 0) if e != nil { diff --git a/tools/thothctl/internal/safeio/files_unix_test.go b/tools/thothctl/internal/safeio/files_unix_test.go index 30aebf53..1f2f147b 100644 --- a/tools/thothctl/internal/safeio/files_unix_test.go +++ b/tools/thothctl/internal/safeio/files_unix_test.go @@ -30,3 +30,51 @@ func TestReadCanonicalRegularRejectsNamedPipeWithoutBlocking(t *testing.T) { t.Fatalf("named pipe error = %v, want ErrUnsafeFile", err) } } + +func TestCanonicalDescriptorOwnershipDoesNotLeakAcrossNestedOperations(t *testing.T) { + root, err := filepath.EvalSymlinks(t.TempDir()) + if err != nil { + t.Fatal(err) + } + nested := filepath.Join(root, "one", "two") + if err := os.MkdirAll(nested, 0o700); err != nil { + t.Fatal(err) + } + input := filepath.Join(nested, "input.sql") + if err := os.WriteFile(input, []byte("select 1"), 0o600); err != nil { + t.Fatal(err) + } + fdCount := func() int { + f, err := os.Open("/dev/fd") + if err != nil { + t.Fatal(err) + } + defer f.Close() + names, err := f.Readdirnames(-1) + if err != nil { + t.Fatal(err) + } + return len(names) + } + baseline := fdCount() + for i := 0; i < 20; i++ { + if _, err := ReadCanonicalRegular(input, 1024); err != nil { + t.Fatal(err) + } + if err := validateCanonicalOutputPath(filepath.Join(nested, "out-"+string(rune('a'+i))+".yaml")); err != nil { + t.Fatal(err) + } + } + if got := fdCount(); got > baseline+2 { + t.Fatalf("descriptor leak after read/validate: baseline=%d got=%d", baseline, got) + } + for i := 0; i < 20; i++ { + path := filepath.Join(nested, "write-"+string(rune('a'+i))+".yaml") + if err := writeCanonicalExclusive(path, []byte("ok"), 0o600); err != nil { + t.Fatal(err) + } + } + if got := fdCount(); got > baseline+2 { + t.Fatalf("descriptor leak after writes: baseline=%d got=%d", baseline, got) + } +} diff --git a/tools/thothctl/internal/safeio/files_windows.go b/tools/thothctl/internal/safeio/files_windows.go index 115e0ed0..d6322902 100644 --- a/tools/thothctl/internal/safeio/files_windows.go +++ b/tools/thothctl/internal/safeio/files_windows.go @@ -122,7 +122,14 @@ func writeCanonicalExclusive(path string, contents []byte, mode fs.FileMode) err if err := ValidateCanonicalPath(path); err != nil || len(contents) > 16<<20 || mode.Perm() == 0 || mode.Perm()&0o077 != 0 { return ErrUnsafeFile } - h, err := windows.CreateFile(windows.StringToUTF16Ptr(path), windows.GENERIC_WRITE, 0, nil, windows.CREATE_NEW, windows.FILE_ATTRIBUTE_NORMAL|windows.FILE_FLAG_OPEN_REPARSE_POINT, 0) + parent, retainedParents, err := openWindowsParents(path) + if err != nil { + return ErrUnsafeFile + } + defer closeWindowsHandles(retainedParents) + // Parent handles stay open with delete sharing denied until publication and + // identity recheck complete; this is the Windows equivalent of retained dirfds. + h, err := windows.CreateFile(windows.StringToUTF16Ptr(filepath.Join(parent, filepath.Base(path))), windows.GENERIC_WRITE, 0, nil, windows.CREATE_NEW, windows.FILE_ATTRIBUTE_NORMAL|windows.FILE_FLAG_OPEN_REPARSE_POINT, 0) if err != nil { return ErrUnsafeFile } @@ -132,6 +139,9 @@ func writeCanonicalExclusive(path string, contents []byte, mode fs.FileMode) err return ErrUnsafeFile } defer f.Close() + if err := f.Chmod(mode); err != nil { + return ErrUnsafeFile + } if _, err := f.Write(contents); err != nil { return ErrUnsafeFile } @@ -153,26 +163,37 @@ func writeCanonicalExclusive(path string, contents []byte, mode fs.FileMode) err return nil } -func validateCanonicalOutputPath(path string) error { - if err := ValidateCanonicalPath(path); err != nil { - return err - } +func openWindowsParents(path string) (string, []windows.Handle, error) { volume := filepath.VolumeName(path) root := volume + string(filepath.Separator) components := strings.Split(strings.TrimPrefix(path, root), string(filepath.Separator)) if volume == "" || len(components) < 2 || components[0] == "" { - return ErrUnsafeFile + return "", nil, ErrUnsafeFile } current := root + parents := make([]windows.Handle, 0, len(components)-1) for _, component := range components[:len(components)-1] { current = filepath.Join(current, component) h, err := openWindowsComponent(current, true) if err != nil { - return ErrUnsafeFile + closeWindowsHandles(parents) + return "", nil, ErrUnsafeFile } - windows.CloseHandle(h) + parents = append(parents, h) } - if _, err := os.Lstat(filepath.Join(current, components[len(components)-1])); err == nil || !os.IsNotExist(err) { + return current, parents, nil +} + +func validateCanonicalOutputPath(path string) error { + if err := ValidateCanonicalPath(path); err != nil { + return err + } + current, parents, err := openWindowsParents(path) + if err != nil { + return ErrUnsafeFile + } + defer closeWindowsHandles(parents) + if _, err := os.Lstat(filepath.Join(current, filepath.Base(path))); err == nil || !os.IsNotExist(err) { return ErrUnsafeFile } return nil diff --git a/tools/thothctl/internal/safeio/files_windows_test.go b/tools/thothctl/internal/safeio/files_windows_test.go index 745ea672..5a1da6f4 100644 --- a/tools/thothctl/internal/safeio/files_windows_test.go +++ b/tools/thothctl/internal/safeio/files_windows_test.go @@ -66,3 +66,9 @@ func TestOpenWindowsComponentBlocksMutationWhileHandleIsRetained(t *testing.T) { } }) } + +func TestWriteCanonicalExclusiveRequiresRestrictiveMode(t *testing.T) { + if err := writeCanonicalExclusive(`C:\\tmp\\thothctl-output.yaml`, []byte("x"), 0o640); err == nil { + t.Fatal("accepted non-restrictive output mode") + } +} diff --git a/tools/thothctl/internal/workspaceops/operations.go b/tools/thothctl/internal/workspaceops/operations.go index 459e9cd4..572c19a7 100644 --- a/tools/thothctl/internal/workspaceops/operations.go +++ b/tools/thothctl/internal/workspaceops/operations.go @@ -404,7 +404,7 @@ func validateIngress(payload []byte, expected inputEnvelope) error { var raw map[string]json.RawMessage d := json.NewDecoder(bytes.NewReader(payload)) d.UseNumber() - if d.Decode(&raw) != nil { + if d.Decode(&raw) != nil || hasDuplicateJSONFields(payload) { return errors.New("invalid workspace request") } var extra any @@ -417,6 +417,12 @@ func validateIngress(payload []byte, expected inputEnvelope) error { if dec.Decode(&supplied) != nil || supplied.SchemaVersion != 1 || supplied.Operation != expected.Operation || supplied.WorkspaceID != expected.WorkspaceID { return errors.New("invalid workspace request") } + // Presence is part of the command-derived envelope contract. Decoding into Go values + // alone would make an omitted field indistinguishable from an explicit zero/null value. + var generatedRaw map[string]json.RawMessage + if json.Unmarshal(mustJSON(expected), &generatedRaw) != nil || !sameJSONFieldSet(raw, generatedRaw) { + return errors.New("workspace request does not match command") + } if !reflect.DeepEqual(supplied, expected) { return errors.New("workspace request does not match command") } @@ -476,7 +482,7 @@ func Run(ctx context.Context, installation config.Installation, runner compose.R } var raw map[string]json.RawMessage d := json.NewDecoder(strings.NewReader(cr.Stdout)) - if d.Decode(&raw) != nil { + if d.Decode(&raw) != nil || hasDuplicateJSONFields([]byte(cr.Stdout)) { return Result{}, errors.New("invalid workspace result") } var extra any @@ -490,9 +496,20 @@ func Run(ctx context.Context, installation config.Installation, runner compose.R if operationName(command) != "suggest-fks" { return Result{}, errors.New("invalid workspace result") } - if json.Unmarshal(h, &export) != nil { + var candidate hostExport + if hasDuplicateJSONFields(h) { return Result{}, errors.New("invalid host export") } + exportDecoder := json.NewDecoder(bytes.NewReader(h)) + exportDecoder.DisallowUnknownFields() + if exportDecoder.Decode(&candidate) != nil { + return Result{}, errors.New("invalid host export") + } + var trailing any + if exportDecoder.Decode(&trailing) != io.EOF { + return Result{}, errors.New("invalid host export") + } + export = &candidate delete(raw, "hostExport") } b, _ := json.Marshal(raw) @@ -505,10 +522,13 @@ func Run(ctx context.Context, installation config.Installation, runner compose.R if e = validateResult(result, env.WorkspaceID, operationName(command)); e != nil { return Result{}, e } + if !resultExitMatches(result.Status, cr.ExitCode) { + return Result{}, runErrOr(runErr, "workspace result status does not match child exit") + } if runErr != nil { - if result.Status == "blocked" && cr.ExitCode == 3 { - } else if result.Status == "failed" && cr.ExitCode == 1 { - } else { + // Exit 1/3 are represented by the validated public result; callers still + // render it and select the corresponding process exit code. + if result.Status != "blocked" && result.Status != "failed" { return Result{}, runErr } } @@ -552,9 +572,6 @@ func candidateBoundToResult(x hostExport, result Result) bool { } func publishCandidate(x *hostExport, result Result, path string) error { - if path == "" { - return nil - } if x.MediaType != "application/yaml" && x.MediaType != "text/yaml" { return errors.New("invalid candidate export") } @@ -575,6 +592,9 @@ func publishCandidate(x *hostExport, result Result, path string) error { if result.RunID == "" || !runIDPattern.MatchString(result.RunID) { return errors.New("invalid candidate identity") } + if path == "" { + return nil + } if e = safeio.WriteCanonicalExclusive(path, b, 0o600); e != nil { return errors.New("unsafe output file") } @@ -606,8 +626,89 @@ func validateResult(r Result, workspace, operation string) error { if r.Status != "blocked" && r.Code == CodeRegistryBootstrapRecoveryConflict { return errors.New("invalid workspace result") } + if (r.Status == "succeeded" || r.Status == "unchanged" || r.Status == "dry_run") && r.Code != "ok" { + return errors.New("invalid workspace result") + } + if r.Status == "failed" && (r.Code == "ok" || r.Code == "manual_review_required" || r.Code == "evidence_materialization_required" || r.Code == "preprocessing_conflict" || r.Code == "preprocessing_resume_mismatch") { + return errors.New("invalid workspace result") + } return nil } + +func resultExitMatches(status string, exitCode int) bool { + switch status { + case "succeeded", "unchanged", "dry_run": + return exitCode == 0 + case "blocked": + return exitCode == 3 + case "failed": + return exitCode == 1 + default: + return false + } +} + +func hasDuplicateJSONFields(payload []byte) bool { + decoder := json.NewDecoder(bytes.NewReader(payload)) + decoder.UseNumber() + if !scanJSONValue(decoder) { + return true + } + _, err := decoder.Token() + return err != io.EOF +} + +func scanJSONValue(decoder *json.Decoder) bool { + token, err := decoder.Token() + if err != nil { + return false + } + delimiter, isDelimiter := token.(json.Delim) + if !isDelimiter { + return true + } + switch delimiter { + case '{': + seen := map[string]bool{} + for decoder.More() { + keyToken, keyErr := decoder.Token() + key, ok := keyToken.(string) + if keyErr != nil || !ok || seen[key] || !scanJSONValue(decoder) { + return false + } + seen[key] = true + } + _, err = decoder.Token() + return err == nil + case '[': + for decoder.More() { + if !scanJSONValue(decoder) { + return false + } + } + _, err = decoder.Token() + return err == nil + default: + return false + } +} + +func sameJSONFieldSet(a, b map[string]json.RawMessage) bool { + if len(a) != len(b) { + return false + } + for key := range a { + if _, ok := b[key]; !ok { + return false + } + } + return true +} + +func mustJSON(v any) []byte { + b, _ := json.Marshal(v) + return b +} func DigestBytes(b []byte) string { s := sha256.Sum256(b); return "sha256:" + hex.EncodeToString(s[:]) } func validStatus(v string) bool { switch v { diff --git a/tools/thothctl/internal/workspaceops/operations_test.go b/tools/thothctl/internal/workspaceops/operations_test.go index 3a11b977..2f8149f0 100644 --- a/tools/thothctl/internal/workspaceops/operations_test.go +++ b/tools/thothctl/internal/workspaceops/operations_test.go @@ -1,7 +1,9 @@ package workspaceops import ( + "bytes" "encoding/base64" + "fmt" "os" "path/filepath" "reflect" @@ -132,3 +134,164 @@ func TestRunPublishesOnlyVerifiedCandidateExport(t *testing.T) { t.Fatalf("candidate = %q, %v", b, err) } } + +func validWorkspaceResult(status, code string) Result { + return Result{SchemaVersion: 1, Status: status, Code: code, WorkspaceID: "psd", WorkspaceRevision: strings.Repeat("0", 40), DescriptorBlob: strings.Repeat("a", 40), Operation: "evidence", CompletedStages: []string{}} +} + +func TestValidateIngressRejectsExplicitOmittedEvidenceZeroFields(t *testing.T) { + command := EvidenceRequest{WorkspaceID: "psd"} + expected, _, err := makeInput(command) + if err != nil { + t.Fatal(err) + } + for _, payload := range []string{ + `{"schemaVersion":1,"operation":"evidence","workspaceId":"psd","resume":""}`, + `{"schemaVersion":1,"operation":"evidence","workspaceId":"psd","workspaceId":"psd"}`, + `{"schemaVersion":1,"operation":"evidence","workspaceId":"psd","dryRun":false}`, + `{"schemaVersion":1,"operation":"evidence","workspaceId":"psd","resume":null}`, + } { + if err := validateIngress([]byte(payload), expected); err == nil { + t.Errorf("accepted non-derived envelope %s", payload) + } + } +} + +func TestValidateIngressRequiresExactDerivedFieldSetForOptionalInputs(t *testing.T) { + command := SuggestFksRequest{WorkspaceID: "psd"} + expected, _, err := makeInput(command) + if err != nil { + t.Fatal(err) + } + payload := `{"schemaVersion":1,"operation":"suggest-fks","workspaceId":"psd","sql":null,"assume":[]}` + if err := validateIngress([]byte(payload), expected); err == nil { + t.Fatal("accepted explicit null/empty optional fields omitted by command") + } +} + +func TestPublishCandidateValidatesInternalExportWithoutOutput(t *testing.T) { + candidate := []byte("candidates: []\n") + digest := DigestBytes(candidate) + result := validWorkspaceResult("succeeded", "ok") + result.Operation = "suggest-fks" + result.RunID = strings.Repeat("b", 32) + result.ArtifactIdentities = []ArtifactIdentity{{Kind: "fk-candidates", Digest: digest}} + for _, export := range []hostExport{ + {MediaType: "application/json", SHA256: digest, ContentBase64: base64.StdEncoding.EncodeToString(candidate)}, + {MediaType: "application/yaml", SHA256: digest, ContentBase64: base64.StdEncoding.EncodeToString(candidate)}, + } { + if export.MediaType == "application/json" { + if err := publishCandidate(&export, result, ""); err == nil { + t.Fatal("accepted invalid media type without output") + } + } else if err := publishCandidate(&export, result, ""); err != nil { + t.Fatal(err) + } + } +} + +func TestValidateResultBindsStatusCodeAndExit(t *testing.T) { + cases := []struct { + status, code string + exit int + valid bool + }{ + {"succeeded", "ok", 0, true}, {"unchanged", "ok", 0, true}, {"dry_run", "ok", 0, true}, + {"blocked", "manual_review_required", 3, true}, {"failed", "workspace_not_found", 1, true}, + {"blocked", "manual_review_required", 0, false}, {"failed", "workspace_not_found", 0, false}, + {"succeeded", "workspace_not_found", 0, false}, {"failed", "ok", 1, false}, + } + for _, tc := range cases { + r := validWorkspaceResult(tc.status, tc.code) + err := validateResult(r, "psd", "evidence") + got := err == nil && resultExitMatches(tc.status, tc.exit) + if got != tc.valid { + t.Errorf("%s/%s exit %d valid=%v err=%v", tc.status, tc.code, tc.exit, tc.valid, err) + } + } +} + +func TestRunRejectsUnknownHostExportFieldWithoutOutput(t *testing.T) { + d := t.TempDir() + candidate := []byte("candidates: []\n") + digest := DigestBytes(candidate) + response := `{"schemaVersion":1,"status":"succeeded","code":"ok","workspaceId":"psd","workspaceRevision":"` + strings.Repeat("0", 40) + `","descriptorBlob":"` + strings.Repeat("a", 40) + `","operation":"suggest-fks","runId":"` + strings.Repeat("b", 32) + `","completedStages":[],"artifactIdentities":[{"kind":"fk-candidates","digest":"` + digest + `"}],"hostExport":{"mediaType":"application/yaml","sha256":"` + digest + `","contentBase64":"` + base64.StdEncoding.EncodeToString(candidate) + `","extra":1}}` + fake := filepath.Join(d, "docker") + if err := os.WriteFile(fake, []byte("#!/bin/sh\ncat >/dev/null\nprintf '%s' '"+response+"'\n"), 0o700); err != nil { + t.Fatal(err) + } + cmd := SuggestFksRequest{WorkspaceID: "psd"} + if _, err := Run(context.Background(), config.Installation{ProjectDirectory: d}, compose.NewRunner(fake), cmd, nil); err == nil { + t.Fatal("accepted unknown hostExport field") + } +} + +func TestPublishCandidateRejectsOversizedExportWithoutOutput(t *testing.T) { + candidate := bytes.Repeat([]byte("y"), maxCandidate+1) + digest := DigestBytes(candidate) + x := hostExport{MediaType: "application/yaml", SHA256: digest, ContentBase64: base64.StdEncoding.EncodeToString(candidate)} + r := validWorkspaceResult("succeeded", "ok") + r.Operation, r.RunID = "suggest-fks", strings.Repeat("b", 32) + if err := publishCandidate(&x, r, ""); err == nil { + t.Fatal("accepted oversized export without output") + } +} + +func TestRunAcceptsOnlyMatchingChildExitForStatus(t *testing.T) { + for _, tc := range []struct { + name, status, code string + exit int + wantErr bool + }{ + {"blocked-exit3", "blocked", "manual_review_required", 3, false}, + {"blocked-exit0", "blocked", "manual_review_required", 0, true}, + {"failed-exit1", "failed", "workspace_not_found", 1, false}, + {"failed-exit0", "failed", "workspace_not_found", 0, true}, + } { + t.Run(tc.name, func(t *testing.T) { + d := t.TempDir() + response := fmt.Sprintf(`{"schemaVersion":1,"status":"%s","code":"%s","workspaceId":"psd","workspaceRevision":"%s","descriptorBlob":"%s","operation":"evidence","completedStages":[]}`, + tc.status, tc.code, strings.Repeat("0", 40), strings.Repeat("a", 40)) + fake := filepath.Join(d, "docker") + if err := os.WriteFile(fake, []byte("#!/bin/sh\ncat >/dev/null\nprintf '%s' '"+response+"'\nexit "+fmt.Sprint(tc.exit)+"\n"), 0o700); err != nil { + t.Fatal(err) + } + _, err := Run(context.Background(), config.Installation{ProjectDirectory: d}, compose.NewRunner(fake), EvidenceRequest{WorkspaceID: "psd"}, nil) + if (err != nil) != tc.wantErr { + t.Fatalf("error=%v wantErr=%v", err, tc.wantErr) + } + }) + } +} + +func TestParseWorkspaceRejectsDuplicateMissingAndUnknownOptions(t *testing.T) { + validResume := strings.Repeat("0", 32) + for _, argv := range [][]string{ + {"workspace", "inspect", "--json", "--json", "--workspace", "psd"}, + {"workspace", "inspect", "--workspace", "psd", "--workspace", "psd"}, + {"workspace", "preprocess", "dwh", "--workspace", "psd", "--resume", validResume, "--resume", validResume}, + {"workspace", "schema", "suggest-fks", "--workspace", "psd", "--output", "a", "--output", "b"}, + {"workspace", "inspect", "--json", "--unknown", "x", "--workspace", "psd"}, + {"workspace", "inspect", "--json"}, + } { + if _, err := ParseWorkspaceCommand(argv); err == nil { + t.Errorf("accepted invalid argv %v", argv) + } + } +} + +func TestParseWorkspaceAcceptsAssumptionAndEnforcesLimits(t *testing.T) { + if _, err := ParseWorkspaceCommand([]string{"workspace", "schema", "suggest-fks", "--workspace", "psd", "--assume", "orders.id=customers.id"}); err != nil { + t.Fatal(err) + } + if _, err := ParseWorkspaceCommand([]string{"workspace", "schema", "suggest-fks", "--workspace", "psd", "--assume", strings.Repeat("a", maxAssumptionBytes+1)}); err == nil { + t.Fatal("accepted oversized assumption") + } + args := []string{"workspace", "schema", "suggest-fks", "--workspace", "psd"} + for i := 0; i < 33; i++ { + args = append(args, "--from-sql", fmt.Sprintf("schema-%d.sql", i)) + } + if _, err := ParseWorkspaceCommand(args); err == nil { + t.Fatal("accepted 33 SQL files") + } +}