diff --git a/docs/contracts/workspace-preprocessing-cli.md b/docs/contracts/workspace-preprocessing-cli.md index 1c3e1e04..0d9ff37a 100644 --- a/docs/contracts/workspace-preprocessing-cli.md +++ b/docs/contracts/workspace-preprocessing-cli.md @@ -1,19 +1,21 @@ # Workspace preprocessing CLI contract -`thothctl workspace` is a closed native host interface. It accepts only the commands and options listed below; unknown options, passthrough separators, shell fragments, and bootstrap run selectors are rejected before Compose is invoked. +`thothctl workspace` is a closed native host interface. Only the seven paths below are representable; there is no passthrough, shell, arbitrary child argv, or host-selected bootstrap run ID. ```text -thothctl --installation /absolute/thothii-installation.yaml workspace inspect --workspace ID [--json] -thothctl ... workspace preprocess dwh --workspace ID [--resume RUN] [--json] -thothctl ... workspace schema suggest-fks --workspace ID [--from-sql FILE]... [--assume COLUMN=TABLE]... [--output FILE] [--json] -thothctl ... workspace schema check --workspace ID --resume RUN [--annotations FILE --reviewed-candidates sha256:HEX] [--json] -thothctl ... workspace index-schema --workspace ID [--json] -thothctl ... workspace preprocess evidence --workspace ID [--dry-run] [--resume RUN] [--json] -thothctl ... workspace preprocess run --workspace ID [--resume RUN] [--json] +workspace inspect --workspace ID [--json] +workspace preprocess dwh --workspace ID [--resume RUN] [--json] +workspace schema suggest-fks --workspace ID [--from-sql FILE]... [--assume COLUMN=TABLE]... [--output FILE] [--json] +workspace schema check --workspace ID --resume RUN [--annotations FILE --reviewed-candidates sha256:HEX] [--json] +workspace index-schema --workspace ID [--json] +workspace preprocess evidence --workspace ID [--dry-run] [--resume RUN] [--json] +workspace preprocess run --workspace ID [--resume RUN] [--json] ``` -Workspace IDs are lowercase `[a-z][a-z0-9-]{2,62}` and outer run IDs are exactly 32 lowercase hexadecimal characters. SQL ingress is limited to 32 regular, canonical, non-symlink files of at most 1 MiB each and 16 MiB total. Annotation ingress is UTF-8 and at most 16 MiB. Inputs are read without following links and are never mounted as host directories. Suggestion candidate output is bounded to 700 KiB. +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. -`workspace inspect` has no resume or bootstrap option. Bootstrap recovery is automatic and installation-scoped; the host never chooses a run ID. `schema check` requires an explicit outer run ID and requires `--annotations` and `--reviewed-candidates` together. +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 one-shot operation emits one schema-v1 JSON envelope. Stdout is capped at 1 MiB and sanitized stderr at 64 KiB. Exit status is 0 for success/unchanged/dry-run, 3 for an expected blocked/manual checkpoint, 2 for command or unsafe-file errors, and 1 for operational failures. +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. + +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.` diff --git a/tools/thothctl/cmd/thothctl/main.go b/tools/thothctl/cmd/thothctl/main.go index 74d41af6..afc55f0c 100644 --- a/tools/thothctl/cmd/thothctl/main.go +++ b/tools/thothctl/cmd/thothctl/main.go @@ -90,11 +90,48 @@ func run(ctx context.Context, args []string, stdout, stderr io.Writer) int { runner := compose.NewRunner("") if command == "workspace" { workspaceCommand, parseErr := workspaceops.ParseWorkspaceCommand(append([]string{"workspace"}, commandArgs...)) - if parseErr != nil { return commandUsageError(stderr, parseErr.Error()) } + if parseErr != nil { + return commandUsageError(stderr, parseErr.Error()) + } 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") { return commandUsageError(stderr, operationErr.Error()) }; fmt.Fprintf(stderr, "thothctl: %s\n", output.Sanitize(operationErr.Error(), secretValues)); return 1 } - if err := json.NewEncoder(stdout).Encode(result); err != nil { return 1 } - if result.Status == "blocked" { return 3 }; if result.Status == "failed" { return 1 }; return 0 + 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") { + return commandUsageError(stderr, operationErr.Error()) + } + fmt.Fprintf(stderr, "thothctl: %s\n", output.Sanitize(operationErr.Error(), secretValues)) + return 1 + } + jsonMode := true + switch c := workspaceCommand.(type) { + case workspaceops.InspectCommand: + jsonMode = c.JSON + case workspaceops.DwhRequest: + jsonMode = c.JSON + case workspaceops.SuggestFksRequest: + jsonMode = c.JSON + case workspaceops.CheckSchemaRequest: + jsonMode = c.JSON + case workspaceops.IndexSchemaRequest: + jsonMode = c.JSON + case workspaceops.EvidenceRequest: + jsonMode = c.JSON + case workspaceops.RunRequest: + jsonMode = c.JSON + } + if jsonMode { + if err := json.NewEncoder(stdout).Encode(result); err != nil { + return 1 + } + } else { + renderWorkspaceHuman(stdout, result) + } + if result.Status == "blocked" { + return 3 + } + if result.Status == "failed" { + return 1 + } + return 0 } var result compose.Result switch command { @@ -173,6 +210,20 @@ func run(ctx context.Context, args []string, stdout, stderr io.Writer) int { return writeResult(result, err, secretValues, stdout, stderr) } +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.") + return + } + fmt.Fprintf(w, "Workspace %s: %s (%s)\n", result.WorkspaceID, result.Status, result.Code) + if result.RunID != "" { + fmt.Fprintf(w, "Run: %s\n", result.RunID) + } + if len(result.CompletedStages) > 0 { + fmt.Fprintf(w, "Completed stages: %s\n", strings.Join(result.CompletedStages, ", ")) + } +} + func writeRemovalTargets(outputWriter io.Writer, project string, targets []serverops.Container) { fmt.Fprintf(outputWriter, "Removal targets for installation project %s:\n", project) if len(targets) == 0 { diff --git a/tools/thothctl/cmd/thothctl/main_test.go b/tools/thothctl/cmd/thothctl/main_test.go index 630abc1d..432bde9b 100644 --- a/tools/thothctl/cmd/thothctl/main_test.go +++ b/tools/thothctl/cmd/thothctl/main_test.go @@ -15,6 +15,7 @@ import ( "github.com/aritmolab/thothii/tools/thothctl/internal/config" "github.com/aritmolab/thothii/tools/thothctl/internal/pi" "github.com/aritmolab/thothii/tools/thothctl/internal/testsupport" + "github.com/aritmolab/thothii/tools/thothctl/internal/workspaceops" ) func TestInstallationRunnerMapsProfileToSessionInventoryScope(t *testing.T) { @@ -846,3 +847,11 @@ func assertInvocationContains(t *testing.T, invocations [][]string, want ...stri } t.Fatalf("invocations = %#v, want %#v", invocations, want) } + +func TestRenderWorkspaceHumanHidesRecoveryIdentity(t *testing.T) { + var out bytes.Buffer + renderWorkspaceHuman(&out, workspaceops.Result{WorkspaceID: "psd", Status: "blocked", Code: workspaceops.CodeRegistryBootstrapRecoveryConflict, RunID: "0123456789abcdef0123456789abcdef"}) + if out.String() != "Bootstrap recovery is ambiguous or corrupt; inspect the installation registry jobs.\n" { + t.Fatalf("human output = %q", out.String()) + } +} diff --git a/tools/thothctl/internal/compose/process_unix.go b/tools/thothctl/internal/compose/process_unix.go index 872d6c42..f69852d9 100644 --- a/tools/thothctl/internal/compose/process_unix.go +++ b/tools/thothctl/internal/compose/process_unix.go @@ -1,5 +1,18 @@ //go:build !windows + package compose -import ("os/exec"; "syscall") -func configureOwnedProcess(c *exec.Cmd) { c.SysProcAttr=&syscall.SysProcAttr{Setpgid:true} } -func terminateOwnedProcess(c *exec.Cmd) { if c.Process!=nil { _=syscall.Kill(-c.Process.Pid, syscall.SIGKILL); _=c.Process.Kill() } } + +import ( + "os/exec" + "syscall" +) + +func configureOwnedProcess(c *exec.Cmd) { c.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} } +func registerOwnedProcess(c *exec.Cmd) error { return nil } +func releaseOwnedProcess(c *exec.Cmd) {} +func terminateOwnedProcess(c *exec.Cmd) { + if c.Process != nil { + _ = syscall.Kill(-c.Process.Pid, syscall.SIGKILL) + _ = c.Process.Kill() + } +} diff --git a/tools/thothctl/internal/compose/process_windows.go b/tools/thothctl/internal/compose/process_windows.go index 3a8fc1b6..6a34a721 100644 --- a/tools/thothctl/internal/compose/process_windows.go +++ b/tools/thothctl/internal/compose/process_windows.go @@ -1,5 +1,51 @@ //go:build windows + package compose -import ("os/exec"; "syscall") -func configureOwnedProcess(c *exec.Cmd) { c.SysProcAttr=&syscall.SysProcAttr{} } -func terminateOwnedProcess(c *exec.Cmd) { if c.Process!=nil { _=c.Process.Kill() } } + +import ( + "golang.org/x/sys/windows" + "os/exec" + "sync" +) + +var ownedJobs = struct { + sync.Mutex + m map[*exec.Cmd]windows.Handle +}{m: make(map[*exec.Cmd]windows.Handle)} + +func configureOwnedProcess(c *exec.Cmd) { + c.SysProcAttr = &windows.SysProcAttr{CreationFlags: windows.CREATE_NEW_PROCESS_GROUP} +} +func registerOwnedProcess(c *exec.Cmd) error { + h, 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) + return err + } + ownedJobs.Lock() + ownedJobs.m[c] = h + ownedJobs.Unlock() + return nil +} +func releaseOwnedProcess(c *exec.Cmd) { + ownedJobs.Lock() + h := ownedJobs.m[c] + delete(ownedJobs.m, c) + ownedJobs.Unlock() + if h != 0 { + windows.CloseHandle(h) + } +} +func terminateOwnedProcess(c *exec.Cmd) { + ownedJobs.Lock() + h := ownedJobs.m[c] + ownedJobs.Unlock() + if h != 0 { + _ = windows.TerminateJobObject(h, 1) + } else if c.Process != nil { + _ = c.Process.Kill() + } +} diff --git a/tools/thothctl/internal/compose/runner.go b/tools/thothctl/internal/compose/runner.go index 8e8148fe..f915a4a1 100644 --- a/tools/thothctl/internal/compose/runner.go +++ b/tools/thothctl/internal/compose/runner.go @@ -7,111 +7,154 @@ import ( "errors" "fmt" "io" - "sync" "os" "os/exec" + "sync" ) -// Result is the captured output and process exit code for one Docker invocation. var ErrOutputLimit = errors.New("compose output limit exceeded") -// CaptureLimits bounds each stream independently. Overflow is never retained. -type CaptureLimits struct { StdoutBytes int64; StderrBytes int64 } - +type CaptureLimits struct { + StdoutBytes int64 + StderrBytes int64 +} type Result struct { Stdout string Stderr string ExitCode int } +type Runner struct{ binary string } -// Runner executes the Docker CLI. It never invokes a shell. -type Runner struct { - binary string -} - -// NewRunner returns a runner for binary. An empty binary selects docker from PATH. func NewRunner(binary string) Runner { if binary == "" { binary = "docker" } return Runner{binary: binary} } - -// Run invokes Docker with the supplied argument array and optional standard input. func (r Runner) Run(ctx context.Context, args []string, stdin io.Reader) (Result, error) { - command := exec.CommandContext(ctx, r.binary, args...) - command.Stdin = stdin - var stdout, stderr bytes.Buffer - command.Stdout = &stdout - command.Stderr = &stderr - err := command.Run() - result := Result{Stdout: stdout.String(), Stderr: stderr.String()} + c := exec.CommandContext(ctx, r.binary, args...) + c.Stdin = stdin + var out, er bytes.Buffer + c.Stdout = &out + c.Stderr = &er + err := c.Run() + res := Result{Stdout: out.String(), Stderr: er.String()} if err == nil { - return result, nil + return res, nil } - var exitError *exec.ExitError - if errors.As(err, &exitError) { - result.ExitCode = exitError.ExitCode() - return result, err + var x *exec.ExitError + if errors.As(err, &x) { + res.ExitCode = x.ExitCode() + return res, err } if errors.Is(err, exec.ErrNotFound) || errors.Is(err, os.ErrNotExist) { - result.ExitCode = 127 - return result, fmt.Errorf("%w: %w", exec.ErrNotFound, err) + res.ExitCode = 127 + return res, fmt.Errorf("%w: %w", exec.ErrNotFound, err) } - return result, err + return res, err } - -// RunBounded executes one owned process while retaining at most the configured bytes per stream. -// The complete process group is terminated as soon as either stream exceeds its limit. +// RunBounded streams each pipe into a preallocated fixed-capacity collector. It owns the +// process group and tears it down on the first overflow or cancellation. func (r Runner) RunBounded(ctx context.Context, args []string, stdin io.Reader, limits CaptureLimits) (Result, error) { - if limits.StdoutBytes < 0 || limits.StderrBytes < 0 { return Result{}, ErrOutputLimit } - command := exec.Command(r.binary, args...) - command.Stdin = stdin - stdoutPipe, err := command.StdoutPipe(); if err != nil { return Result{}, err } - stderrPipe, err := command.StderrPipe(); if err != nil { return Result{}, err } - configureOwnedProcess(command) - if err := command.Start(); err != nil { return Result{ExitCode: 127}, err } - type stream struct { b []byte; overflow bool } - var out, er stream + if limits.StdoutBytes < 0 || limits.StderrBytes < 0 { + return Result{}, ErrOutputLimit + } + c := exec.Command(r.binary, args...) + c.Stdin = stdin + op, err := c.StdoutPipe() + if err != nil { + return Result{}, err + } + ep, err := c.StderrPipe() + if err != nil { + return Result{}, err + } + configureOwnedProcess(c) + if err = c.Start(); err != nil { + return Result{ExitCode: 127}, err + } + if err = registerOwnedProcess(c); err != nil { + terminateOwnedProcess(c) + _ = c.Wait() + return Result{}, err + } + defer releaseOwnedProcess(c) + type collector struct { + buf []byte + n int64 + overflow bool + } + out := collector{buf: make([]byte, limits.StdoutBytes)} + er := collector{buf: make([]byte, limits.StderrBytes)} overflow := make(chan struct{}, 1) - var wg sync.WaitGroup; wg.Add(2) - read := func(rd io.Reader, max int64, target *stream) { + var wg sync.WaitGroup + wg.Add(2) + read := func(rd io.Reader, max int64, col *collector) { defer wg.Done() - buf := make([]byte, 32*1024) + chunk := make([]byte, 32*1024) for { - n, readErr := rd.Read(buf) + n, e := rd.Read(chunk) if n > 0 { - remain := max - int64(len(target.b)) - if remain > 0 { take := int64(n); if take > remain { take = remain }; target.b = append(target.b, buf[:take]...) } + remain := max - col.n + if remain > 0 { + take := int64(n) + if take > remain { + take = remain + } + copy(col.buf[col.n:col.n+take], chunk[:take]) + col.n += take + } if int64(n) > remain { - target.overflow = true - select { case overflow <- struct{}{}: default: } + col.overflow = true + select { + case overflow <- struct{}{}: + default: + } return } } - if readErr != nil { return } + if e != nil { + return + } } } - go read(stdoutPipe, limits.StdoutBytes, &out); go read(stderrPipe, limits.StderrBytes, &er) - finished := make(chan error, 1); go func() { finished <- command.Wait() }() + go read(op, limits.StdoutBytes, &out) + go read(ep, limits.StderrBytes, &er) + waited := make(chan error, 1) + go func() { waited <- c.Wait() }() var waitErr error + var cancelled error select { - case waitErr = <-finished: - // The parent may exit while a descendant inherited stdout/stderr; tear down the owned group. - terminateOwnedProcess(command) + case waitErr = <-waited: + terminateOwnedProcess(c) case <-overflow: - terminateOwnedProcess(command); waitErr = <-finished + terminateOwnedProcess(c) + waitErr = <-waited case <-ctx.Done(): - terminateOwnedProcess(command); waitErr = <-finished + cancelled = ctx.Err() + terminateOwnedProcess(c) + waitErr = <-waited } wg.Wait() - result := Result{Stdout: string(out.b), Stderr: string(er.b)} - if out.overflow || er.overflow { return result, ErrOutputLimit } - if waitErr == nil { return result, nil } - var exitErr *exec.ExitError - if errors.As(waitErr, &exitErr) { result.ExitCode = exitErr.ExitCode(); return result, waitErr } - if errors.Is(waitErr, exec.ErrNotFound) || errors.Is(waitErr, os.ErrNotExist) { result.ExitCode = 127; return result, fmt.Errorf("%w: %w", exec.ErrNotFound, waitErr) } - if ctx.Err() != nil { return result, ctx.Err() } - return result, waitErr + res := Result{Stdout: string(out.buf[:out.n]), Stderr: string(er.buf[:er.n])} + if out.overflow || er.overflow { + return res, ErrOutputLimit + } + if cancelled != nil { + return res, cancelled + } + if waitErr == nil { + return res, nil + } + var ex *exec.ExitError + if errors.As(waitErr, &ex) { + res.ExitCode = ex.ExitCode() + return res, waitErr + } + if errors.Is(waitErr, exec.ErrNotFound) || errors.Is(waitErr, os.ErrNotExist) { + res.ExitCode = 127 + return res, fmt.Errorf("%w: %w", exec.ErrNotFound, waitErr) + } + return res, waitErr } diff --git a/tools/thothctl/internal/compose/runner_test.go b/tools/thothctl/internal/compose/runner_test.go index 72b4c325..fd630064 100644 --- a/tools/thothctl/internal/compose/runner_test.go +++ b/tools/thothctl/internal/compose/runner_test.go @@ -6,8 +6,10 @@ import ( "os" "os/exec" "path/filepath" + "runtime" "strings" "testing" + "time" ) func TestRunnerPassesEachArgumentWithoutShellSplitting(t *testing.T) { @@ -65,18 +67,78 @@ func writeExecutable(t *testing.T, contents string) string { return path } - func TestRunnerBoundedCapturesExactlyAtLimit(t *testing.T) { runner := NewRunner(writeExecutable(t, "#!/bin/sh\nprintf 1234567890\nprintf error12345 >&2\n")) result, err := runner.RunBounded(context.Background(), nil, nil, CaptureLimits{StdoutBytes: 10, StderrBytes: 10}) - if err != nil { t.Fatal(err) } - if result.Stdout != "1234567890" || result.Stderr != "error12345" { t.Fatalf("bounded result = %#v", result) } + if err != nil { + t.Fatal(err) + } + if result.Stdout != "1234567890" || result.Stderr != "error12345" { + t.Fatalf("bounded result = %#v", result) + } } func TestRunnerBoundedRejectsOneByteOverEachStream(t *testing.T) { for _, script := range []string{"#!/bin/sh\nprintf 12345678901\n", "#!/bin/sh\nprintf 12345678901 >&2\n"} { - runner := NewRunner(writeExecutable(t, script)); result, err := runner.RunBounded(context.Background(), nil, nil, CaptureLimits{StdoutBytes: 10, StderrBytes: 10}) - if !errors.Is(err, ErrOutputLimit) { t.Fatalf("err = %v result=%#v", err, result) } - if len(result.Stdout) > 10 || len(result.Stderr) > 10 { t.Fatalf("retained overflow: %#v", result) } + runner := NewRunner(writeExecutable(t, script)) + result, err := runner.RunBounded(context.Background(), nil, nil, CaptureLimits{StdoutBytes: 10, StderrBytes: 10}) + if !errors.Is(err, ErrOutputLimit) { + t.Fatalf("err = %v result=%#v", err, result) + } + if len(result.Stdout) > 10 || len(result.Stderr) > 10 { + t.Fatalf("retained overflow: %#v", result) + } + } +} + +func TestRunnerBoundedTerminatesInfiniteStreams(t *testing.T) { + for _, script := range []string{"#!/bin/sh\nwhile :; do printf x; done", "#!/bin/sh\nwhile :; do printf x >&2; done"} { + runner := NewRunner(writeExecutable(t, script)) + result, err := runner.RunBounded(context.Background(), nil, nil, CaptureLimits{StdoutBytes: 1024, StderrBytes: 1024}) + if !errors.Is(err, ErrOutputLimit) { + t.Fatalf("err = %v result = %#v", err, result) + } + if len(result.Stdout) > 1024 || len(result.Stderr) > 1024 { + t.Fatalf("retained overflow: %#v", result) + } + } +} + +func TestRunnerBoundedCancellationWinsOverChildKill(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + runner := NewRunner(writeExecutable(t, "#!/bin/sh\nsleep 30\n")) + done := make(chan error, 1) + go func() { + _, err := runner.RunBounded(ctx, nil, nil, CaptureLimits{StdoutBytes: 32, StderrBytes: 32}) + done <- err + }() + cancel() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("err = %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("cancellation did not terminate process") + } +} + +func TestRunnerBoundedTerminatesGrandchildHoldingPipe(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("shell process-group fixture is Unix-specific") + } + runner := NewRunner(writeExecutable(t, "#!/bin/sh\n(sleep 30; printf survivor) &\nwhile :; do printf x; done\n")) + done := make(chan error, 1) + go func() { + _, err := runner.RunBounded(context.Background(), nil, nil, CaptureLimits{StdoutBytes: 64, StderrBytes: 64}) + done <- err + }() + select { + case err := <-done: + if !errors.Is(err, ErrOutputLimit) { + t.Fatalf("err = %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("grandchild kept owned pipe alive") } } diff --git a/tools/thothctl/internal/config/installation_test.go b/tools/thothctl/internal/config/installation_test.go index d12e918d..2e630a74 100644 --- a/tools/thothctl/internal/config/installation_test.go +++ b/tools/thothctl/internal/config/installation_test.go @@ -178,7 +178,11 @@ func TestLoadRejectsRelativeInstallationPaths(t *testing.T) { func writeInstallation(t *testing.T, profile string) (string, string, string, string) { t.Helper() - root := filepath.Join(t.TempDir(), "installation folder with spaces") + rootBase, err := filepath.EvalSymlinks(t.TempDir()) + if err != nil { + t.Fatal(err) + } + root := filepath.Join(rootBase, "installation folder with spaces") projectDirectory := filepath.Join(root, "project directory with spaces") if err := os.MkdirAll(filepath.Join(projectDirectory, "deploy"), 0o755); err != nil { t.Fatal(err) diff --git a/tools/thothctl/internal/safeio/files_unix.go b/tools/thothctl/internal/safeio/files_unix.go index 2a493b30..7662aaf7 100644 --- a/tools/thothctl/internal/safeio/files_unix.go +++ b/tools/thothctl/internal/safeio/files_unix.go @@ -56,9 +56,19 @@ func ReadCanonicalRegular(path string, maximum int64) ([]byte, error) { if err != nil { return nil, ErrUnsafeFile } + // Checking the retained descriptor alone misses a pathname replacement while the + // read is in progress. The parent descriptor and final name must still resolve to + // the exact open file after reading. if err := unix.Fstat(int(file.Fd()), &after); err != nil || after.Nlink > 1 || after.Mode != before.Mode || after.Ino != before.Ino || after.Dev != before.Dev || after.Size != before.Size { return nil, ErrUnsafeFile } + var named unix.Stat_t + if err := unix.Fstatat(directory, components[len(components)-1], &named, unix.AT_SYMLINK_NOFOLLOW); err != nil || named.Nlink > 1 || named.Mode != before.Mode || named.Ino != before.Ino || named.Dev != before.Dev { + return nil, ErrUnsafeFile + } + if !recheckUnixParents(components, directories) { + return nil, ErrUnsafeFile + } return contents, nil } @@ -89,6 +99,10 @@ func writeCanonicalExclusive(path string, contents []byte, mode fs.FileMode) err unix.Close(dir) dir = next } + var parentBefore unix.Stat_t + if err := unix.Fstat(dir, &parentBefore); err != nil { + return ErrUnsafeFile + } fd, err := unix.Openat(dir, components[len(components)-1], unix.O_WRONLY|unix.O_CREAT|unix.O_EXCL|unix.O_CLOEXEC|unix.O_NOFOLLOW, uint32(mode.Perm())) if err != nil { return ErrUnsafeFile @@ -108,6 +122,16 @@ func writeCanonicalExclusive(path string, contents []byte, mode fs.FileMode) err if err := f.Sync(); err != nil { return ErrUnsafeFile } + var opened, named, parentAfter unix.Stat_t + if err := unix.Fstat(fd, &opened); err != nil || opened.Nlink != 1 { + return ErrUnsafeFile + } + if err := unix.Fstatat(dir, components[len(components)-1], &named, unix.AT_SYMLINK_NOFOLLOW); err != nil || named.Nlink != 1 || named.Ino != opened.Ino || named.Dev != opened.Dev { + return ErrUnsafeFile + } + if err := unix.Fstat(dir, &parentAfter); err != nil || parentAfter.Ino != parentBefore.Ino || parentAfter.Dev != parentBefore.Dev || !recheckUnixParentPath(components[:len(components)-1], dir) { + return ErrUnsafeFile + } return nil } @@ -140,3 +164,46 @@ func validateCanonicalOutputPath(path string) error { } return nil } + +func recheckUnixParents(components []string, retained []int) bool { + if len(retained) != len(components) { + return false + } + dir, err := unix.Open("/", unix.O_RDONLY|unix.O_DIRECTORY|unix.O_CLOEXEC, 0) + if err != nil { + return false + } + defer 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 { + return false + } + var got, want unix.Stat_t + if unix.Fstat(next, &got) != nil || unix.Fstat(retained[i+1], &want) != nil || got.Ino != want.Ino || got.Dev != want.Dev { + unix.Close(next) + return false + } + unix.Close(dir) + dir = next + } + return true +} + +func recheckUnixParentPath(components []string, retained int) bool { + dir, err := unix.Open("/", unix.O_RDONLY|unix.O_DIRECTORY|unix.O_CLOEXEC, 0) + if err != nil { + return false + } + defer 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 { + return false + } + unix.Close(dir) + dir = next + } + var got, want unix.Stat_t + return unix.Fstat(dir, &got) == nil && unix.Fstat(retained, &want) == nil && got.Ino == want.Ino && got.Dev == want.Dev +} diff --git a/tools/thothctl/internal/safeio/files_windows.go b/tools/thothctl/internal/safeio/files_windows.go index c67df68e..115e0ed0 100644 --- a/tools/thothctl/internal/safeio/files_windows.go +++ b/tools/thothctl/internal/safeio/files_windows.go @@ -54,7 +54,29 @@ func ReadCanonicalRegular(path string, maximum int64) ([]byte, error) { return nil, ErrUnsafeFile } defer file.Close() - return readBoundedRegularFile(file, maximum) + var before, after windows.ByHandleFileInformation + if windows.GetFileInformationByHandle(handle, &before) != nil || before.NumberOfLinks > 1 { + return nil, ErrUnsafeFile + } + contents, err := readBoundedRegularFile(file, maximum) + if err != nil { + return nil, ErrUnsafeFile + } + if windows.GetFileInformationByHandle(handle, &after) != nil || after.NumberOfLinks > 1 || !sameWindowsFile(before, after) { + return nil, ErrUnsafeFile + } + // Re-opened pathname identity is checked as well as the retained handle. This + // catches replacement by an ordinary local actor where share/delete policy permits it. + check, err := openWindowsComponent(current, false) + if err != nil { + return nil, ErrUnsafeFile + } + defer windows.CloseHandle(check) + var named windows.ByHandleFileInformation + if windows.GetFileInformationByHandle(check, &named) != nil || named.NumberOfLinks > 1 || !sameWindowsFile(before, named) { + return nil, ErrUnsafeFile + } + return contents, nil } func openWindowsComponent(path string, directory bool) (windows.Handle, error) { @@ -81,7 +103,7 @@ func openWindowsComponent(path string, directory bool) (windows.Handle, error) { windows.CloseHandle(handle) return 0, err } - if information.FileAttributes&windows.FILE_ATTRIBUTE_REPARSE_POINT != 0 || + if information.NumberOfLinks > 1 || information.FileAttributes&windows.FILE_ATTRIBUTE_REPARSE_POINT != 0 || (directory && information.FileAttributes&windows.FILE_ATTRIBUTE_DIRECTORY == 0) || (!directory && information.FileAttributes&windows.FILE_ATTRIBUTE_DIRECTORY != 0) { windows.CloseHandle(handle) @@ -113,7 +135,22 @@ func writeCanonicalExclusive(path string, contents []byte, mode fs.FileMode) err if _, err := f.Write(contents); err != nil { return ErrUnsafeFile } - return f.Sync() + if err := f.Sync(); err != nil { + return ErrUnsafeFile + } + var opened, named windows.ByHandleFileInformation + if windows.GetFileInformationByHandle(h, &opened) != nil || opened.NumberOfLinks != 1 { + return ErrUnsafeFile + } + check, err := openWindowsComponent(path, false) + if err != nil { + return ErrUnsafeFile + } + defer windows.CloseHandle(check) + if windows.GetFileInformationByHandle(check, &named) != nil || named.NumberOfLinks != 1 || !sameWindowsFile(opened, named) { + return ErrUnsafeFile + } + return nil } func validateCanonicalOutputPath(path string) error { @@ -140,3 +177,7 @@ func validateCanonicalOutputPath(path string) error { } return nil } + +func sameWindowsFile(a, b windows.ByHandleFileInformation) bool { + return a.VolumeSerialNumber == b.VolumeSerialNumber && a.FileIndexHigh == b.FileIndexHigh && a.FileIndexLow == b.FileIndexLow +} diff --git a/tools/thothctl/internal/serverops/operations_test.go b/tools/thothctl/internal/serverops/operations_test.go index d1f6906f..4e6bf7d6 100644 --- a/tools/thothctl/internal/serverops/operations_test.go +++ b/tools/thothctl/internal/serverops/operations_test.go @@ -294,7 +294,10 @@ func TestRemoveRejectsConfirmationForDifferentContainerIDs(t *testing.T) { func testInstallation(t *testing.T) config.Installation { t.Helper() - root := t.TempDir() + root, err := filepath.EvalSymlinks(t.TempDir()) + if err != nil { + t.Fatal(err) + } project := filepath.Join(root, "project") if err := os.Mkdir(project, 0o700); err != nil { t.Fatal(err) diff --git a/tools/thothctl/internal/workspaceops/operations.go b/tools/thothctl/internal/workspaceops/operations.go index ebe61e0f..459e9cd4 100644 --- a/tools/thothctl/internal/workspaceops/operations.go +++ b/tools/thothctl/internal/workspaceops/operations.go @@ -5,17 +5,21 @@ import ( "bytes" "context" "crypto/sha256" + "encoding/base64" "encoding/hex" "encoding/json" "errors" "fmt" "io" + "reflect" "regexp" "strings" + "unicode/utf8" "github.com/aritmolab/thothii/tools/thothctl/internal/compose" "github.com/aritmolab/thothii/tools/thothctl/internal/config" "github.com/aritmolab/thothii/tools/thothctl/internal/safeio" + "gopkg.in/yaml.v3" ) const CodeRegistryBootstrapRecoveryConflict = "registry_bootstrap_recovery_conflict" @@ -26,21 +30,23 @@ const ( maxAssumptionBytes = 256 maxResult = 1 << 20 maxCandidate = 700 << 10 + maxAnnotations = 16 << 20 ) var workspaceIDPattern = regexp.MustCompile(`^[a-z][a-z0-9-]{2,62}$`) var runIDPattern = regexp.MustCompile(`^[0-9a-f]{32}$`) +var revisionPattern = regexp.MustCompile(`^[0-9a-f]{40}$`) var digestPattern = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`) +// Command is intentionally closed: callers cannot provide a child argv or bootstrap run ID. type Command interface{ workspaceCommand() } type InspectCommand struct { WorkspaceID string JSON bool } type DwhRequest struct { - WorkspaceID string - Resume string - JSON bool + WorkspaceID, Resume string + JSON bool } type SuggestFksRequest struct { WorkspaceID string @@ -50,11 +56,8 @@ type SuggestFksRequest struct { JSON bool } type CheckSchemaRequest struct { - WorkspaceID string - Resume string - Annotations string - ReviewedCandidates string - JSON bool + WorkspaceID, Resume, Annotations, ReviewedCandidates string + JSON bool } type IndexSchemaRequest struct { WorkspaceID string @@ -67,9 +70,8 @@ type EvidenceRequest struct { JSON bool } type RunRequest struct { - WorkspaceID string - Resume string - JSON bool + WorkspaceID, Resume string + JSON bool } func (InspectCommand) workspaceCommand() {} @@ -99,6 +101,33 @@ type ArtifactIdentity struct { Kind string `json:"kind"` Digest string `json:"digest"` } +type hostExport struct { + MediaType string `json:"mediaType"` + SHA256 string `json:"sha256"` + ContentBase64 string `json:"contentBase64"` +} + +type sqlInput struct { + Basename string `json:"basename"` + ContentBase64 string `json:"contentBase64"` + SHA256 string `json:"sha256"` +} +type annotationInput struct { + Basename string `json:"basename"` + ContentBase64 string `json:"contentBase64"` + SHA256 string `json:"sha256"` +} +type inputEnvelope struct { + SchemaVersion int `json:"schemaVersion"` + Operation string `json:"operation"` + WorkspaceID string `json:"workspaceId"` + Resume string `json:"resume,omitempty"` + SQL []sqlInput `json:"sql,omitempty"` + Assume []string `json:"assume,omitempty"` + Annotations *annotationInput `json:"annotations,omitempty"` + ReviewedCandidates string `json:"reviewedCandidates,omitempty"` + DryRun bool `json:"dryRun,omitempty"` +} func invalid(msg string) (Command, error) { return nil, errors.New(msg) } func requireWorkspace(v string) error { @@ -117,46 +146,77 @@ func one(args []string, i *int, flag string) (string, error) { if *i+1 >= len(args) || strings.HasPrefix(args[*i+1], "--") { return "", fmt.Errorf("%s requires a value", flag) } - *i = *i + 1 + *i++ return args[*i], nil } + func ParseWorkspaceCommand(args []string) (Command, error) { if len(args) < 2 || args[0] != "workspace" { return invalid("workspace command is required") } - area, action := args[1], "" - rest := args[2:] - if area == "preprocess" || area == "schema" { - if len(rest) == 0 { - return invalid("workspace group requires an operation") - } - action = rest[0] - rest = rest[1:] - } else { - action = area - } - if action != "inspect" && action != "dwh" && action != "suggest-fks" && action != "check" && action != "index-schema" && action != "evidence" && action != "run" { - return invalid("unknown workspace command") + // Parse the hierarchy before options. This prevents aliases such as workspace schema dwh. + path := strings.Join(args[:min(3, len(args))], " ") + var action string + var rest []string + switch { + case args[1] == "inspect": + action = "inspect" + rest = args[2:] + case len(args) >= 3 && args[1] == "preprocess" && (args[2] == "dwh" || args[2] == "evidence" || args[2] == "run"): + action = "preprocess " + args[2] + rest = args[3:] + case len(args) >= 3 && args[1] == "schema" && (args[2] == "suggest-fks" || args[2] == "check"): + action = "schema " + args[2] + rest = args[3:] + case args[1] == "index-schema": + action = "index-schema" + rest = args[2:] + default: + return invalid("unknown workspace command: " + path) } var ws, resume, annotations, reviewed, output string var jsonOut, dry bool var sql, assume []string seen := map[string]bool{} + allowed := func(flag string) bool { + switch action { + case "inspect": + return flag == "--workspace" || flag == "--json" + case "preprocess dwh", "preprocess run": + return flag == "--workspace" || flag == "--resume" || flag == "--json" + case "preprocess evidence": + return flag == "--workspace" || flag == "--resume" || flag == "--dry-run" || flag == "--json" + case "schema suggest-fks": + return flag == "--workspace" || flag == "--from-sql" || flag == "--assume" || flag == "--output" || flag == "--json" + case "schema check": + return flag == "--workspace" || flag == "--resume" || flag == "--annotations" || flag == "--reviewed-candidates" || flag == "--json" + case "index-schema": + return flag == "--workspace" || flag == "--json" + } + return false + } for i := 0; i < len(rest); i++ { f := rest[i] - if f == "--json" { + if !allowed(f) { + return invalid("unknown or invalid workspace option") + } + if f == "--json" || f == "--dry-run" { if seen[f] { - return invalid("duplicate --json") + return invalid("duplicate " + f) } seen[f] = true - jsonOut = true + if f == "--json" { + jsonOut = true + } else { + dry = true + } continue } + if seen[f] && f != "--from-sql" && f != "--assume" { + return invalid("duplicate " + f) + } switch f { case "--workspace": - if seen[f] { - return invalid("duplicate --workspace") - } seen[f] = true v, e := one(rest, &i, f) if e != nil { @@ -164,9 +224,6 @@ func ParseWorkspaceCommand(args []string) (Command, error) { } ws = v case "--resume": - if seen[f] { - return invalid("duplicate --resume") - } seen[f] = true v, e := one(rest, &i, f) if e != nil { @@ -174,9 +231,6 @@ func ParseWorkspaceCommand(args []string) (Command, error) { } resume = v case "--from-sql": - if action != "suggest-fks" { - return invalid("--from-sql is valid only for schema suggest-fks") - } v, e := one(rest, &i, f) if e != nil { return nil, e @@ -186,9 +240,6 @@ func ParseWorkspaceCommand(args []string) (Command, error) { } sql = append(sql, v) case "--assume": - if action != "suggest-fks" { - return invalid("--assume is valid only for schema suggest-fks") - } v, e := one(rest, &i, f) if e != nil { return nil, e @@ -196,14 +247,11 @@ func ParseWorkspaceCommand(args []string) (Command, error) { if len(assume) >= maxAssumptions || len([]byte(v)) > maxAssumptionBytes { return invalid("--assume exceeds limit") } + if !regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)?=[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)?$`).MatchString(v) { + return invalid("--assume must be column=table") + } assume = append(assume, v) case "--output": - if action != "suggest-fks" { - return invalid("--output is valid only for schema suggest-fks") - } - if seen[f] { - return invalid("duplicate --output") - } seen[f] = true v, e := one(rest, &i, f) if e != nil { @@ -211,12 +259,6 @@ func ParseWorkspaceCommand(args []string) (Command, error) { } output = v case "--annotations": - if action != "check" { - return invalid("--annotations is valid only for schema check") - } - if seen[f] { - return invalid("duplicate --annotations") - } seen[f] = true v, e := one(rest, &i, f) if e != nil { @@ -224,46 +266,26 @@ func ParseWorkspaceCommand(args []string) (Command, error) { } annotations = v case "--reviewed-candidates": - if action != "check" { - return invalid("--reviewed-candidates is valid only for schema check") - } - if seen[f] { - return invalid("duplicate --reviewed-candidates") - } seen[f] = true v, e := one(rest, &i, f) if e != nil { return nil, e } reviewed = v - case "--dry-run": - if action != "evidence" { - return invalid("--dry-run is valid only for evidence") - } - if seen[f] { - return invalid("duplicate --dry-run") - } - seen[f] = true - dry = true - default: - return invalid("unknown workspace option") } } if err := requireWorkspace(ws); err != nil { return nil, err } - if action != "inspect" && resume != "" { + if resume != "" { if err := requireRun(resume); err != nil { return nil, err } } - if resume != "" && action != "dwh" && action != "evidence" && action != "run" && action != "check" { - return invalid("--resume is not valid for this workspace command") - } if action == "inspect" && resume != "" { return invalid("inspect does not accept --resume") } - if action == "check" { + if action == "schema check" { if resume == "" { return invalid("schema check requires --resume") } @@ -274,43 +296,32 @@ func ParseWorkspaceCommand(args []string) (Command, error) { return invalid("--reviewed-candidates must be sha256:") } } - if action != "suggest-fks" && (len(sql) > 0 || len(assume) > 0 || output != "") { - return invalid("schema options are valid only for suggest-fks") + if action != "schema check" && (annotations != "" || reviewed != "") { + return invalid("annotation options are valid only for schema check") } switch action { case "inspect": return InspectCommand{ws, jsonOut}, nil - case "dwh": + case "preprocess dwh": return DwhRequest{ws, resume, jsonOut}, nil - case "suggest-fks": + case "schema suggest-fks": return SuggestFksRequest{ws, sql, assume, output, jsonOut}, nil - case "check": + case "schema check": return CheckSchemaRequest{ws, resume, annotations, reviewed, jsonOut}, nil case "index-schema": return IndexSchemaRequest{ws, jsonOut}, nil - case "evidence": + case "preprocess evidence": return EvidenceRequest{ws, dry, resume, jsonOut}, nil default: return RunRequest{ws, resume, jsonOut}, nil } } - -type inputEnvelope struct { - SchemaVersion int `json:"schemaVersion"` - Operation string `json:"operation"` - WorkspaceID string `json:"workspaceId"` - Resume string `json:"resume,omitempty"` - SQL []sqlInput `json:"sql,omitempty"` - Assume []string `json:"assume,omitempty"` - Annotations string `json:"annotations,omitempty"` - ReviewedCandidates string `json:"reviewedCandidates,omitempty"` - DryRun bool `json:"dryRun,omitempty"` +func min(a, b int) int { + if a < b { + return a + } + return b } -type sqlInput struct { - Name string `json:"name"` - Content string `json:"content"` -} - func operationName(c Command) string { switch c.(type) { case InspectCommand: @@ -329,114 +340,275 @@ func operationName(c Command) string { return "run" } } -func Run(ctx context.Context, installation config.Installation, runner compose.Runner, command Command, stdin io.Reader) (Result, error) { - env := inputEnvelope{SchemaVersion: 1, Operation: operationName(command)} - var outputPath string - switch c := command.(type) { + +func makeInput(c Command) (inputEnvelope, string, error) { + env := inputEnvelope{SchemaVersion: 1, Operation: operationName(c)} + var out string + switch x := c.(type) { case InspectCommand: - env.WorkspaceID = c.WorkspaceID + env.WorkspaceID = x.WorkspaceID case DwhRequest: - env.WorkspaceID, env.Resume = c.WorkspaceID, c.Resume + env.WorkspaceID, env.Resume = x.WorkspaceID, x.Resume case SuggestFksRequest: - env.WorkspaceID, env.Assume, outputPath = c.WorkspaceID, c.Assume, c.Output - if outputPath != "" { - if err := safeio.ValidateCanonicalOutputPath(outputPath); err != nil { - return Result{}, errors.New("unsafe output file") + env.WorkspaceID, env.Assume, out = x.WorkspaceID, x.Assume, x.Output + if out != "" { + if err := safeio.ValidateCanonicalOutputPath(out); err != nil { + return env, "", errors.New("unsafe output file") } } var total int64 - for _, path := range c.FromSQL { + for _, path := range x.FromSQL { b, e := safeio.ReadCanonicalUTF8(path, maxSQLFile) if e != nil { - return Result{}, errors.New("unsafe SQL input") + return env, "", errors.New("unsafe SQL input") } total += int64(len(b)) if total > maxSQLTotal { - return Result{}, errors.New("SQL input exceeds limit") - } - env.SQL = append(env.SQL, sqlInput{path, string(b)}) - } - if len(env.SQL) > 0 { - var n int - for _, x := range env.SQL { - n += len(x.Content) - } - if n > maxCandidate { - return Result{}, errors.New("candidate input exceeds limit") + return env, "", errors.New("SQL input exceeds limit") } + base := path[strings.LastIndexAny(path, "/\\")+1:] + env.SQL = append(env.SQL, sqlInput{base, base64.StdEncoding.EncodeToString(b), DigestBytes(b)}) } case CheckSchemaRequest: - env.WorkspaceID, env.Resume, env.Annotations, env.ReviewedCandidates = c.WorkspaceID, c.Resume, c.Annotations, c.ReviewedCandidates - if c.Annotations != "" { - b, e := safeio.ReadCanonicalUTF8(c.Annotations, 16<<20) + env.WorkspaceID, env.Resume, env.ReviewedCandidates = x.WorkspaceID, x.Resume, x.ReviewedCandidates + if x.Annotations != "" { + b, e := safeio.ReadCanonicalUTF8(x.Annotations, maxAnnotations) if e != nil { - return Result{}, errors.New("unsafe annotation input") + return env, "", errors.New("unsafe annotation input") } - env.Annotations = string(b) + base := x.Annotations[strings.LastIndexAny(x.Annotations, "/\\")+1:] + env.Annotations = &annotationInput{base, base64.StdEncoding.EncodeToString(b), DigestBytes(b)} } case IndexSchemaRequest: - env.WorkspaceID = c.WorkspaceID + env.WorkspaceID = x.WorkspaceID case EvidenceRequest: - env.WorkspaceID, env.Resume, env.DryRun = c.WorkspaceID, c.Resume, c.DryRun + env.WorkspaceID, env.Resume, env.DryRun = x.WorkspaceID, x.Resume, x.DryRun case RunRequest: - env.WorkspaceID, env.Resume = c.WorkspaceID, c.Resume + env.WorkspaceID, env.Resume = x.WorkspaceID, x.Resume default: - return Result{}, errors.New("unsupported workspace command") + return env, "", errors.New("unsupported workspace command") } payload, e := json.Marshal(env) + if e != nil { + return env, "", e + } + if len(payload) > maxResult { + return env, "", errors.New("request exceeds limit") + } + return env, string(payload), nil +} + +var commonIngress = map[string]bool{"schemaVersion": true, "operation": true, "workspaceId": true} + +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 { + return errors.New("invalid workspace request") + } + var extra any + if d.Decode(&extra) != io.EOF { + return errors.New("invalid workspace request") + } + var supplied inputEnvelope + dec := json.NewDecoder(bytes.NewReader(payload)) + dec.DisallowUnknownFields() + if dec.Decode(&supplied) != nil || supplied.SchemaVersion != 1 || supplied.Operation != expected.Operation || supplied.WorkspaceID != expected.WorkspaceID { + return errors.New("invalid workspace request") + } + if !reflect.DeepEqual(supplied, expected) { + return errors.New("workspace request does not match command") + } + allowed := map[string]bool{} + for k := range commonIngress { + allowed[k] = true + } + switch expected.Operation { + case "dwh", "run": + if expected.Resume != "" { + allowed["resume"] = true + } + case "evidence": + allowed["resume"] = true + allowed["dryRun"] = true + case "suggest-fks": + allowed["sql"] = true + allowed["assume"] = true + case "check": + allowed["resume"] = true + allowed["annotations"] = true + allowed["reviewedCandidates"] = true + } + for k := range raw { + if !allowed[k] { + return errors.New("invalid workspace request") + } + } + return nil +} + +func Run(ctx context.Context, installation config.Installation, runner compose.Runner, command Command, stdin io.Reader) (Result, error) { + env, generated, e := makeInput(command) if e != nil { return Result{}, e } + payload := []byte(generated) if stdin != nil { - payload, e = io.ReadAll(io.LimitReader(stdin, 1<<20+1)) - if e != nil { - return Result{}, e - } - if len(payload) > 1<<20 { + payload, e = io.ReadAll(io.LimitReader(stdin, maxResult+1)) + if e != nil || len(payload) > maxResult { return Result{}, errors.New("request exceeds limit") } - var supplied inputEnvelope - d := json.NewDecoder(bytes.NewReader(payload)) - d.DisallowUnknownFields() - if e = d.Decode(&supplied); e != nil || supplied.SchemaVersion != 1 || supplied.Operation != env.Operation || supplied.WorkspaceID != env.WorkspaceID { - return Result{}, errors.New("invalid workspace request") + if e = validateIngress(payload, env); e != nil { + return Result{}, e } } - - args := installation.ComposeArgs("run", "--rm", "--no-deps", "workspace-maintenance", "--operation", operationName(command), "--workspace", env.WorkspaceID) - cr, runErr := runner.RunBounded(ctx, args, bytes.NewReader(payload), compose.CaptureLimits{StdoutBytes: 1 << 20, StderrBytes: 64 << 10}) - if runErr != nil && (errors.Is(runErr, compose.ErrOutputLimit) || cr.Stdout == "") { + args := installation.ComposeArgs("run", "--rm", "--no-deps", "--no-TTY", "workspace-maintenance", "--operation", operationName(command), "--workspace", env.WorkspaceID) + cr, runErr := runner.RunBounded(ctx, args, bytes.NewReader(payload), compose.CaptureLimits{StdoutBytes: maxResult, StderrBytes: 64 << 10}) + if errors.Is(runErr, compose.ErrOutputLimit) { return Result{}, runErr } + if len(cr.Stdout) == 0 { + return Result{}, runErrOr(runErr, "workspace maintenance returned no result") + } if len(cr.Stdout) > maxResult { return Result{}, errors.New("invalid workspace result") } - var result Result - dec := json.NewDecoder(strings.NewReader(cr.Stdout)) - dec.DisallowUnknownFields() - if e := dec.Decode(&result); e != nil { + var raw map[string]json.RawMessage + d := json.NewDecoder(strings.NewReader(cr.Stdout)) + if d.Decode(&raw) != nil { return Result{}, errors.New("invalid workspace result") } var extra any - if e := dec.Decode(&extra); e != io.EOF { + if d.Decode(&extra) != io.EOF { return Result{}, errors.New("invalid workspace result") } - if result.SchemaVersion != 1 || result.WorkspaceID != env.WorkspaceID || result.Operation != operationName(command) || !validStatus(result.Status) || !validCode(result.Code) { + var export *hostExport + hasExport := false + if h, ok := raw["hostExport"]; ok { + hasExport = true + if operationName(command) != "suggest-fks" { + return Result{}, errors.New("invalid workspace result") + } + if json.Unmarshal(h, &export) != nil { + return Result{}, errors.New("invalid host export") + } + delete(raw, "hostExport") + } + b, _ := json.Marshal(raw) + var result Result + dec := json.NewDecoder(bytes.NewReader(b)) + dec.DisallowUnknownFields() + if dec.Decode(&result) != nil { return Result{}, errors.New("invalid workspace result") } - if outputPath != "" { - if len(cr.Stdout) > maxCandidate { - return Result{}, errors.New("candidate export exceeds limit") - } - if err := safeio.WriteCanonicalExclusive(outputPath, []byte(cr.Stdout), 0o600); err != nil { - return Result{}, errors.New("unsafe output file") + if e = validateResult(result, env.WorkspaceID, operationName(command)); e != nil { + return Result{}, e + } + if runErr != nil { + if result.Status == "blocked" && cr.ExitCode == 3 { + } else if result.Status == "failed" && cr.ExitCode == 1 { + } else { + return Result{}, runErr } } - + if hasExport { + if export == nil { + return Result{}, errors.New("invalid candidate export") + } + if !candidateBoundToResult(*export, result) { + return Result{}, errors.New("invalid candidate identity") + } + if err := publishCandidate(export, result, outPath(command)); err != nil { + return Result{}, err + } + } else if outPath(command) != "" { + return Result{}, errors.New("candidate export is required") + } return result, nil } -func DigestBytes(b []byte) string { s := sha256.Sum256(b); return "sha256:" + hex.EncodeToString(s[:]) } +func runErrOr(e error, msg string) error { + if e != nil { + return e + } + return errors.New(msg) +} +func outPath(c Command) string { + if x, ok := c.(SuggestFksRequest); ok { + return x.Output + } + return "" +} +func candidateBoundToResult(x hostExport, result Result) bool { + if result.RunID == "" { + return false + } + for _, artifact := range result.ArtifactIdentities { + if artifact.Kind == "fk-candidates" && artifact.Digest == x.SHA256 { + return true + } + } + return false +} +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") + } + if !digestPattern.MatchString(x.SHA256) { + return errors.New("invalid candidate export") + } + b, e := base64.StdEncoding.DecodeString(x.ContentBase64) + if e != nil || len(b) > maxCandidate || !utf8.Valid(b) { + return errors.New("invalid candidate export") + } + if DigestBytes(b) != x.SHA256 { + return errors.New("invalid candidate export") + } + var doc any + if yaml.Unmarshal(b, &doc) != nil { + return errors.New("invalid candidate export") + } + if result.RunID == "" || !runIDPattern.MatchString(result.RunID) { + return errors.New("invalid candidate identity") + } + if e = safeio.WriteCanonicalExclusive(path, b, 0o600); e != nil { + return errors.New("unsafe output file") + } + return nil +} +func validateResult(r Result, workspace, operation string) error { + if r.SchemaVersion != 1 || r.WorkspaceID != workspace || r.Operation != operation || !validStatus(r.Status) || !validCode(r.Code) || !revisionPattern.MatchString(r.WorkspaceRevision) || !revisionPattern.MatchString(r.DescriptorBlob) { + return errors.New("invalid workspace result") + } + if r.RunID != "" && !runIDPattern.MatchString(r.RunID) { + return errors.New("invalid workspace result") + } + for k, v := range r.ChildRuns { + if k == "" || !runIDPattern.MatchString(v) { + return errors.New("invalid workspace result") + } + } + for _, a := range r.ArtifactIdentities { + if a.Kind == "" || !digestPattern.MatchString(a.Digest) { + return errors.New("invalid workspace result") + } + } + if r.CompletedStages == nil { + return errors.New("invalid workspace result") + } + if r.Status == "blocked" && (r.Code != "manual_review_required" && r.Code != "evidence_materialization_required" && r.Code != "preprocessing_conflict" && r.Code != "preprocessing_resume_mismatch" && r.Code != CodeRegistryBootstrapRecoveryConflict) { + return errors.New("invalid workspace result") + } + if r.Status != "blocked" && r.Code == CodeRegistryBootstrapRecoveryConflict { + return errors.New("invalid workspace result") + } + return nil +} +func DigestBytes(b []byte) string { s := sha256.Sum256(b); return "sha256:" + hex.EncodeToString(s[:]) } func validStatus(v string) bool { switch v { case "succeeded", "unchanged", "dry_run", "blocked", "failed": diff --git a/tools/thothctl/internal/workspaceops/operations_test.go b/tools/thothctl/internal/workspaceops/operations_test.go index 528043b0..3a11b977 100644 --- a/tools/thothctl/internal/workspaceops/operations_test.go +++ b/tools/thothctl/internal/workspaceops/operations_test.go @@ -1,8 +1,16 @@ package workspaceops import ( + "encoding/base64" + "os" + "path/filepath" "reflect" + "strings" "testing" + + "context" + "github.com/aritmolab/thothii/tools/thothctl/internal/compose" + "github.com/aritmolab/thothii/tools/thothctl/internal/config" ) func TestParseWorkspaceCommands(t *testing.T) { @@ -48,3 +56,79 @@ func TestParseWorkspaceRejectsUnsafeOrAmbiguousOptions(t *testing.T) { } } } + +func TestParseWorkspaceRejectsNonContractHierarchyAndAssumeShell(t *testing.T) { + for _, argv := range [][]string{ + {"workspace", "schema", "dwh", "--workspace", "psd"}, + {"workspace", "preprocess", "check", "--workspace", "psd"}, + {"workspace", "suggest-fks", "--workspace", "psd"}, + {"workspace", "inspect", "--workspace", "psd", "--bootstrap-run-id", strings.Repeat("a", 32)}, + {"workspace", "schema", "suggest-fks", "--workspace", "psd", "--assume", "a"}, + {"workspace", "schema", "suggest-fks", "--workspace", "psd", "--assume", "a=$(id)"}, + } { + if _, err := ParseWorkspaceCommand(argv); err == nil { + t.Errorf("accepted non-contract argv %v", argv) + } + } +} + +func TestRunUsesBase64BasenameIngressAndNoTTY(t *testing.T) { + root, err := filepath.EvalSymlinks(t.TempDir()) + if err != nil { + t.Fatal(err) + } + sqlPath := filepath.Join(root, "schema.sql") + if err := os.WriteFile(sqlPath, []byte("select 1"), 0o600); err != nil { + t.Fatal(err) + } + out := filepath.Join(root, "candidate.yaml") + resultJSON := `{"schemaVersion":1,"status":"succeeded","code":"ok","workspaceId":"psd","workspaceRevision":"0123456789012345678901234567890123456789","descriptorBlob":"abcdefabcdefabcdefabcdefabcdefabcdefabcd","operation":"suggest-fks","runId":"0123456789abcdef0123456789abcdef","completedStages":[]}` + script := "#!/bin/sh\ncat >/dev/null\nprintf '%s' '" + resultJSON + "'\n" + fake := filepath.Join(root, "docker") + if err := os.WriteFile(fake, []byte(script), 0o700); err != nil { + t.Fatal(err) + } + cmd, err := ParseWorkspaceCommand([]string{"workspace", "schema", "suggest-fks", "--workspace", "psd", "--from-sql", sqlPath}) + if err != nil { + t.Fatal(err) + } + got, err := Run(context.Background(), config.Installation{ProjectDirectory: root, EnvFile: filepath.Join(root, "env")}, compose.NewRunner(fake), cmd, nil) + if err != nil { + t.Fatal(err) + } + if got.Code != "ok" { + t.Fatalf("result = %#v", got) + } + _ = out +} + +func TestRunPublishesOnlyVerifiedCandidateExport(t *testing.T) { + root, err := filepath.EvalSymlinks(t.TempDir()) + if err != nil { + t.Fatal(err) + } + candidate := []byte("candidates: []\n") + digest := DigestBytes(candidate) + encoded := base64.StdEncoding.EncodeToString(candidate) + resultJSON := `{"schemaVersion":1,"status":"succeeded","code":"ok","workspaceId":"psd","workspaceRevision":"0123456789012345678901234567890123456789","descriptorBlob":"abcdefabcdefabcdefabcdefabcdefabcdefabcd","operation":"suggest-fks","runId":"0123456789abcdef0123456789abcdef","completedStages":[],"artifactIdentities":[{"kind":"fk-candidates","digest":"` + digest + `"}],"hostExport":{"mediaType":"application/yaml","sha256":"` + digest + `","contentBase64":"` + encoded + `"}}` + fake := filepath.Join(root, "docker") + if err := os.WriteFile(fake, []byte("#!/bin/sh\ncat >/dev/null\nprintf '%s' '"+resultJSON+"'\n"), 0o700); err != nil { + t.Fatal(err) + } + path := filepath.Join(root, "out.yaml") + cmd, err := ParseWorkspaceCommand([]string{"workspace", "schema", "suggest-fks", "--workspace", "psd", "--output", path}) + if err != nil { + t.Fatal(err) + } + got, err := Run(context.Background(), config.Installation{ProjectDirectory: root, EnvFile: filepath.Join(root, "env")}, compose.NewRunner(fake), cmd, nil) + if err != nil { + t.Fatal(err) + } + if got.RunID == "" { + t.Fatal("missing run identity") + } + b, err := os.ReadFile(path) + if err != nil || string(b) != string(candidate) { + t.Fatalf("candidate = %q, %v", b, err) + } +}