diff --git a/docs/contracts/workspace-preprocessing-cli.md b/docs/contracts/workspace-preprocessing-cli.md new file mode 100644 index 00000000..1c3e1e04 --- /dev/null +++ b/docs/contracts/workspace-preprocessing-cli.md @@ -0,0 +1,19 @@ +# 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. + +```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 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. + +`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 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. diff --git a/tools/thothctl/cmd/thothctl/main.go b/tools/thothctl/cmd/thothctl/main.go index 69d02447..74d41af6 100644 --- a/tools/thothctl/cmd/thothctl/main.go +++ b/tools/thothctl/cmd/thothctl/main.go @@ -19,6 +19,7 @@ import ( "github.com/aritmolab/thothii/tools/thothctl/internal/output" "github.com/aritmolab/thothii/tools/thothctl/internal/pi" "github.com/aritmolab/thothii/tools/thothctl/internal/serverops" + "github.com/aritmolab/thothii/tools/thothctl/internal/workspaceops" ) const usage = `Usage: thothctl --installation /thothii-installation.yaml @@ -50,6 +51,10 @@ Commands: pi maintenance recover --yes Verify a terminal installation, remove stale lifecycle files, and clear maintenance. pi logs Show the latest 200 sanitized core log lines (bounded; no follow mode). + workspace inspect --workspace ID [--json] + workspace preprocess dwh|evidence|run --workspace ID [options] + workspace schema suggest-fks|check --workspace ID [options] + workspace index-schema --workspace ID [--json] ` func main() { @@ -83,6 +88,14 @@ 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()) } + 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 + } var result compose.Result switch command { case "status": diff --git a/tools/thothctl/internal/compose/process_unix.go b/tools/thothctl/internal/compose/process_unix.go new file mode 100644 index 00000000..872d6c42 --- /dev/null +++ b/tools/thothctl/internal/compose/process_unix.go @@ -0,0 +1,5 @@ +//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() } } diff --git a/tools/thothctl/internal/compose/process_windows.go b/tools/thothctl/internal/compose/process_windows.go new file mode 100644 index 00000000..3a8fc1b6 --- /dev/null +++ b/tools/thothctl/internal/compose/process_windows.go @@ -0,0 +1,5 @@ +//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() } } diff --git a/tools/thothctl/internal/compose/runner.go b/tools/thothctl/internal/compose/runner.go index 5d34e273..8e8148fe 100644 --- a/tools/thothctl/internal/compose/runner.go +++ b/tools/thothctl/internal/compose/runner.go @@ -7,11 +7,17 @@ import ( "errors" "fmt" "io" + "sync" "os" "os/exec" ) // 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 Result struct { Stdout string Stderr string @@ -54,3 +60,58 @@ func (r Runner) Run(ctx context.Context, args []string, stdin io.Reader) (Result } return result, 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. +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 + overflow := make(chan struct{}, 1) + var wg sync.WaitGroup; wg.Add(2) + read := func(rd io.Reader, max int64, target *stream) { + defer wg.Done() + buf := make([]byte, 32*1024) + for { + n, readErr := rd.Read(buf) + 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]...) } + if int64(n) > remain { + target.overflow = true + select { case overflow <- struct{}{}: default: } + return + } + } + if readErr != nil { return } + } + } + go read(stdoutPipe, limits.StdoutBytes, &out); go read(stderrPipe, limits.StderrBytes, &er) + finished := make(chan error, 1); go func() { finished <- command.Wait() }() + var waitErr error + select { + case waitErr = <-finished: + // The parent may exit while a descendant inherited stdout/stderr; tear down the owned group. + terminateOwnedProcess(command) + case <-overflow: + terminateOwnedProcess(command); waitErr = <-finished + case <-ctx.Done(): + terminateOwnedProcess(command); waitErr = <-finished + } + 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 +} diff --git a/tools/thothctl/internal/compose/runner_test.go b/tools/thothctl/internal/compose/runner_test.go index f1467e9c..72b4c325 100644 --- a/tools/thothctl/internal/compose/runner_test.go +++ b/tools/thothctl/internal/compose/runner_test.go @@ -64,3 +64,19 @@ 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) } +} + +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) } + } +} diff --git a/tools/thothctl/internal/safeio/files.go b/tools/thothctl/internal/safeio/files.go index b83339cd..5d853bc8 100644 --- a/tools/thothctl/internal/safeio/files.go +++ b/tools/thothctl/internal/safeio/files.go @@ -4,9 +4,11 @@ package safeio import ( "errors" "io" + "io/fs" "os" "path/filepath" "strings" + "unicode/utf8" ) var ErrUnsafeFile = errors.New("unsafe file") @@ -33,3 +35,21 @@ func readBoundedRegularFile(file *os.File, maximum int64) ([]byte, error) { } return contents, nil } + +// ReadCanonicalUTF8 reads a canonical regular file with a strict byte bound and UTF-8 validation. +func ReadCanonicalUTF8(path string, maximum int64) ([]byte, error) { + contents, err := ReadCanonicalRegular(path, maximum) + if err != nil || !utf8.Valid(contents) { + return nil, ErrUnsafeFile + } + return contents, nil +} + +// WriteCanonicalExclusive creates a canonical regular file without following links or replacing +// an existing leaf. The file is private to the caller and is never opened in truncate mode. +func WriteCanonicalExclusive(path string, contents []byte, mode fs.FileMode) error { + return writeCanonicalExclusive(path, contents, mode) +} + +// ValidateCanonicalOutputPath verifies every parent directory without creating the leaf. +func ValidateCanonicalOutputPath(path string) error { return validateCanonicalOutputPath(path) } diff --git a/tools/thothctl/internal/safeio/files_test.go b/tools/thothctl/internal/safeio/files_test.go index 67d4d0fa..2b34c253 100644 --- a/tools/thothctl/internal/safeio/files_test.go +++ b/tools/thothctl/internal/safeio/files_test.go @@ -41,3 +41,37 @@ func TestReadCanonicalRegularRejectsFinalAndParentSymlinks(t *testing.T) { t.Fatalf("final symlink error = %v, want ErrUnsafeFile", err) } } + +func TestReadCanonicalUTF8RejectsNonUTF8AndBounds(t *testing.T) { + path := filepath.Join(t.TempDir(), "input.sql") + if err := os.WriteFile(path, []byte("\xff"), 0o600); err != nil { + t.Fatal(err) + } + if _, err := ReadCanonicalUTF8(path, 1024); !errors.Is(err, ErrUnsafeFile) { + t.Fatalf("ReadCanonicalUTF8 invalid UTF-8 = %v", err) + } + if err := os.WriteFile(path, []byte("12345"), 0o600); err != nil { + t.Fatal(err) + } + if _, err := ReadCanonicalUTF8(path, 4); !errors.Is(err, ErrUnsafeFile) { + t.Fatalf("ReadCanonicalUTF8 oversized = %v", err) + } +} + +func TestWriteCanonicalExclusiveRejectsExistingAndCreatesPrivateFile(t *testing.T) { + root, err := filepath.EvalSymlinks(t.TempDir()) + if err != nil { + t.Fatal(err) + } + path := filepath.Join(root, "candidate.yaml") + if err := WriteCanonicalExclusive(path, []byte("ok"), 0o600); err != nil { + t.Fatal(err) + } + contents, err := os.ReadFile(path) + if err != nil || string(contents) != "ok" { + t.Fatalf("output = %q, %v", contents, err) + } + if err := WriteCanonicalExclusive(path, []byte("replace"), 0o600); !errors.Is(err, ErrUnsafeFile) { + t.Fatalf("replacement = %v", err) + } +} diff --git a/tools/thothctl/internal/safeio/files_unix.go b/tools/thothctl/internal/safeio/files_unix.go index c3a745b9..2a493b30 100644 --- a/tools/thothctl/internal/safeio/files_unix.go +++ b/tools/thothctl/internal/safeio/files_unix.go @@ -3,6 +3,7 @@ package safeio import ( + "io/fs" "os" "strings" @@ -47,7 +48,18 @@ func ReadCanonicalRegular(path string, maximum int64) ([]byte, error) { return nil, ErrUnsafeFile } defer file.Close() - return readBoundedRegularFile(file, maximum) + var before, after unix.Stat_t + if err := unix.Fstat(int(file.Fd()), &before); err != nil || before.Nlink > 1 || before.Mode&unix.S_IFMT != unix.S_IFREG { + return nil, ErrUnsafeFile + } + contents, err := readBoundedRegularFile(file, maximum) + if err != nil { + return nil, ErrUnsafeFile + } + 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 + } + return contents, nil } func closeUnixDescriptors(descriptors []int) { @@ -55,3 +67,76 @@ func closeUnixDescriptors(descriptors []int) { unix.Close(descriptor) } } + +func writeCanonicalExclusive(path string, contents []byte, mode fs.FileMode) error { + if err := ValidateCanonicalPath(path); err != nil || len(contents) > 16<<20 || mode.Perm() == 0 || mode.Perm()&0o077 != 0 { + return ErrUnsafeFile + } + components := strings.Split(strings.TrimPrefix(path, string(os.PathSeparator)), string(os.PathSeparator)) + if len(components) == 0 || components[0] == "" { + return ErrUnsafeFile + } + dir, err := unix.Open("/", unix.O_RDONLY|unix.O_DIRECTORY|unix.O_CLOEXEC, 0) + if err != nil { + return ErrUnsafeFile + } + defer 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 { + return ErrUnsafeFile + } + unix.Close(dir) + dir = next + } + 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 + } + f := os.NewFile(uintptr(fd), "thothctl-safeio-output") + if f == nil { + unix.Close(fd) + return ErrUnsafeFile + } + defer f.Close() + if err := f.Chmod(mode); err != nil { + return ErrUnsafeFile + } + if _, err := f.Write(contents); err != nil { + return ErrUnsafeFile + } + if err := f.Sync(); err != nil { + return ErrUnsafeFile + } + return nil +} + +func validateCanonicalOutputPath(path string) error { + if err := ValidateCanonicalPath(path); err != nil { + return err + } + components := strings.Split(strings.TrimPrefix(path, string(os.PathSeparator)), string(os.PathSeparator)) + if len(components) < 2 || components[0] == "" { + return ErrUnsafeFile + } + dir, err := unix.Open("/", unix.O_RDONLY|unix.O_DIRECTORY|unix.O_CLOEXEC, 0) + if err != nil { + return ErrUnsafeFile + } + defer 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 { + return ErrUnsafeFile + } + unix.Close(dir) + dir = next + } + var st unix.Stat_t + if err := unix.Fstatat(dir, components[len(components)-1], &st, unix.AT_SYMLINK_NOFOLLOW); err == nil { + return ErrUnsafeFile + } else if err != unix.ENOENT { + return ErrUnsafeFile + } + return nil +} diff --git a/tools/thothctl/internal/safeio/files_windows.go b/tools/thothctl/internal/safeio/files_windows.go index ce40ae03..c67df68e 100644 --- a/tools/thothctl/internal/safeio/files_windows.go +++ b/tools/thothctl/internal/safeio/files_windows.go @@ -3,6 +3,7 @@ package safeio import ( + "io/fs" "os" "path/filepath" "strings" @@ -94,3 +95,48 @@ func closeWindowsHandles(handles []windows.Handle) { windows.CloseHandle(handle) } } + +func writeCanonicalExclusive(path string, contents []byte, mode fs.FileMode) error { + 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) + if err != nil { + return ErrUnsafeFile + } + f := os.NewFile(uintptr(h), "thothctl-safeio-output") + if f == nil { + windows.CloseHandle(h) + return ErrUnsafeFile + } + defer f.Close() + if _, err := f.Write(contents); err != nil { + return ErrUnsafeFile + } + return f.Sync() +} + +func validateCanonicalOutputPath(path string) error { + if err := ValidateCanonicalPath(path); err != nil { + return err + } + 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 + } + current := root + for _, component := range components[:len(components)-1] { + current = filepath.Join(current, component) + h, err := openWindowsComponent(current, true) + if err != nil { + return ErrUnsafeFile + } + windows.CloseHandle(h) + } + if _, err := os.Lstat(filepath.Join(current, components[len(components)-1])); err == nil || !os.IsNotExist(err) { + return ErrUnsafeFile + } + return nil +} diff --git a/tools/thothctl/internal/workspaceops/operations.go b/tools/thothctl/internal/workspaceops/operations.go new file mode 100644 index 00000000..ebe61e0f --- /dev/null +++ b/tools/thothctl/internal/workspaceops/operations.go @@ -0,0 +1,453 @@ +// Package workspaceops defines the deliberately closed host workspace contract. +package workspaceops + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "regexp" + "strings" + + "github.com/aritmolab/thothii/tools/thothctl/internal/compose" + "github.com/aritmolab/thothii/tools/thothctl/internal/config" + "github.com/aritmolab/thothii/tools/thothctl/internal/safeio" +) + +const CodeRegistryBootstrapRecoveryConflict = "registry_bootstrap_recovery_conflict" +const ( + maxSQLFile = 1 << 20 + maxSQLTotal = 16 << 20 + maxAssumptions = 256 + maxAssumptionBytes = 256 + maxResult = 1 << 20 + maxCandidate = 700 << 10 +) + +var workspaceIDPattern = regexp.MustCompile(`^[a-z][a-z0-9-]{2,62}$`) +var runIDPattern = regexp.MustCompile(`^[0-9a-f]{32}$`) +var digestPattern = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`) + +type Command interface{ workspaceCommand() } +type InspectCommand struct { + WorkspaceID string + JSON bool +} +type DwhRequest struct { + WorkspaceID string + Resume string + JSON bool +} +type SuggestFksRequest struct { + WorkspaceID string + FromSQL []string + Assume []string + Output string + JSON bool +} +type CheckSchemaRequest struct { + WorkspaceID string + Resume string + Annotations string + ReviewedCandidates string + JSON bool +} +type IndexSchemaRequest struct { + WorkspaceID string + JSON bool +} +type EvidenceRequest struct { + WorkspaceID string + DryRun bool + Resume string + JSON bool +} +type RunRequest struct { + WorkspaceID string + Resume string + JSON bool +} + +func (InspectCommand) workspaceCommand() {} +func (DwhRequest) workspaceCommand() {} +func (SuggestFksRequest) workspaceCommand() {} +func (CheckSchemaRequest) workspaceCommand() {} +func (IndexSchemaRequest) workspaceCommand() {} +func (EvidenceRequest) workspaceCommand() {} +func (RunRequest) workspaceCommand() {} + +type Result struct { + SchemaVersion int `json:"schemaVersion"` + Status string `json:"status"` + Code string `json:"code"` + WorkspaceID string `json:"workspaceId"` + WorkspaceRevision string `json:"workspaceRevision"` + DescriptorBlob string `json:"descriptorBlob"` + Operation string `json:"operation"` + RunID string `json:"runId,omitempty"` + ChildRuns map[string]string `json:"childRuns,omitempty"` + CompletedStages []string `json:"completedStages"` + Counts map[string]int `json:"counts,omitempty"` + ArtifactIdentities []ArtifactIdentity `json:"artifactIdentities,omitempty"` + Warnings []string `json:"warnings,omitempty"` +} +type ArtifactIdentity struct { + Kind string `json:"kind"` + Digest string `json:"digest"` +} + +func invalid(msg string) (Command, error) { return nil, errors.New(msg) } +func requireWorkspace(v string) error { + if !workspaceIDPattern.MatchString(v) { + return errors.New("--workspace must be a valid workspace ID") + } + return nil +} +func requireRun(v string) error { + if !runIDPattern.MatchString(v) { + return errors.New("--resume must be a lowercase 32-hex run ID") + } + return nil +} +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 + 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") + } + var ws, resume, annotations, reviewed, output string + var jsonOut, dry bool + var sql, assume []string + seen := map[string]bool{} + for i := 0; i < len(rest); i++ { + f := rest[i] + if f == "--json" { + if seen[f] { + return invalid("duplicate --json") + } + seen[f] = true + jsonOut = true + continue + } + switch f { + case "--workspace": + if seen[f] { + return invalid("duplicate --workspace") + } + seen[f] = true + v, e := one(rest, &i, f) + if e != nil { + return nil, e + } + ws = v + case "--resume": + if seen[f] { + return invalid("duplicate --resume") + } + seen[f] = true + v, e := one(rest, &i, f) + if e != nil { + return nil, e + } + 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 + } + if len(sql) >= 32 { + return invalid("too many --from-sql files") + } + 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 + } + if len(assume) >= maxAssumptions || len([]byte(v)) > maxAssumptionBytes { + return invalid("--assume exceeds limit") + } + 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 { + return nil, e + } + 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 { + return nil, e + } + 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 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 resume == "" { + return invalid("schema check requires --resume") + } + if (annotations == "") != (reviewed == "") { + return invalid("--annotations and --reviewed-candidates must be supplied together") + } + if reviewed != "" && !digestPattern.MatchString(reviewed) { + 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") + } + switch action { + case "inspect": + return InspectCommand{ws, jsonOut}, nil + case "dwh": + return DwhRequest{ws, resume, jsonOut}, nil + case "suggest-fks": + return SuggestFksRequest{ws, sql, assume, output, jsonOut}, nil + case "check": + return CheckSchemaRequest{ws, resume, annotations, reviewed, jsonOut}, nil + case "index-schema": + return IndexSchemaRequest{ws, jsonOut}, nil + case "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"` +} +type sqlInput struct { + Name string `json:"name"` + Content string `json:"content"` +} + +func operationName(c Command) string { + switch c.(type) { + case InspectCommand: + return "inspect" + case DwhRequest: + return "dwh" + case SuggestFksRequest: + return "suggest-fks" + case CheckSchemaRequest: + return "check" + case IndexSchemaRequest: + return "index-schema" + case EvidenceRequest: + return "evidence" + default: + 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) { + case InspectCommand: + env.WorkspaceID = c.WorkspaceID + case DwhRequest: + env.WorkspaceID, env.Resume = c.WorkspaceID, c.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") + } + } + var total int64 + for _, path := range c.FromSQL { + b, e := safeio.ReadCanonicalUTF8(path, maxSQLFile) + if e != nil { + return Result{}, 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") + } + } + 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) + if e != nil { + return Result{}, errors.New("unsafe annotation input") + } + env.Annotations = string(b) + } + case IndexSchemaRequest: + env.WorkspaceID = c.WorkspaceID + case EvidenceRequest: + env.WorkspaceID, env.Resume, env.DryRun = c.WorkspaceID, c.Resume, c.DryRun + case RunRequest: + env.WorkspaceID, env.Resume = c.WorkspaceID, c.Resume + default: + return Result{}, errors.New("unsupported workspace command") + } + payload, e := json.Marshal(env) + if e != nil { + return Result{}, e + } + if stdin != nil { + payload, e = io.ReadAll(io.LimitReader(stdin, 1<<20+1)) + if e != nil { + return Result{}, e + } + if len(payload) > 1<<20 { + 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") + } + } + + 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 == "") { + return Result{}, runErr + } + 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 { + return Result{}, errors.New("invalid workspace result") + } + var extra any + if e := dec.Decode(&extra); e != 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) { + 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") + } + } + + return result, 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": + return true + } + return false +} +func validCode(v string) bool { + switch v { + case "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", CodeRegistryBootstrapRecoveryConflict: + return true + } + return false +} diff --git a/tools/thothctl/internal/workspaceops/operations_test.go b/tools/thothctl/internal/workspaceops/operations_test.go new file mode 100644 index 00000000..528043b0 --- /dev/null +++ b/tools/thothctl/internal/workspaceops/operations_test.go @@ -0,0 +1,50 @@ +package workspaceops + +import ( + "reflect" + "testing" +) + +func TestParseWorkspaceCommands(t *testing.T) { + cases := []struct { + name string + argv []string + want any + }{ + {"inspect", []string{"workspace", "inspect", "--workspace", "psd", "--json"}, InspectCommand{WorkspaceID: "psd", JSON: true}}, + {"dwh", []string{"workspace", "preprocess", "dwh", "--workspace", "psd", "--resume", "0123456789abcdef0123456789abcdef"}, DwhRequest{WorkspaceID: "psd", Resume: "0123456789abcdef0123456789abcdef"}}, + {"suggest", []string{"workspace", "schema", "suggest-fks", "--workspace", "psd", "--assume", "a=b"}, SuggestFksRequest{WorkspaceID: "psd", Assume: []string{"a=b"}}}, + {"check", []string{"workspace", "schema", "check", "--workspace", "psd", "--resume", "0123456789abcdef0123456789abcdef"}, CheckSchemaRequest{WorkspaceID: "psd", Resume: "0123456789abcdef0123456789abcdef"}}, + {"index", []string{"workspace", "index-schema", "--workspace", "psd"}, IndexSchemaRequest{WorkspaceID: "psd"}}, + {"evidence", []string{"workspace", "preprocess", "evidence", "--workspace", "psd", "--dry-run"}, EvidenceRequest{WorkspaceID: "psd", DryRun: true}}, + {"run", []string{"workspace", "preprocess", "run", "--workspace", "psd"}, RunRequest{WorkspaceID: "psd"}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, err := ParseWorkspaceCommand(tc.argv) + if err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(got, tc.want) { + t.Errorf("got %#v want %#v", got, tc.want) + } + }) + } +} + +func TestParseWorkspaceRejectsUnsafeOrAmbiguousOptions(t *testing.T) { + bad := [][]string{ + {"workspace", "inspect", "--workspace", "psd", "--resume", "0123456789abcdef0123456789abcdef"}, + {"workspace", "inspect", "--workspace", "psd", "--bootstrap-run-id", "0123456789abcdef0123456789abcdef"}, + {"workspace", "inspect", "--workspace", "psd", "--passthrough"}, + {"workspace", "preprocess", "run", "--workspace", "PSd"}, + {"workspace", "preprocess", "run", "--workspace", "psd", "--resume", "bad"}, + {"workspace", "schema", "check", "--workspace", "psd"}, + {"workspace", "schema", "check", "--workspace", "psd", "--resume", "0123456789abcdef0123456789abcdef", "--annotations", "a"}, + } + for _, argv := range bad { + if _, err := ParseWorkspaceCommand(argv); err == nil { + t.Errorf("accepted unsafe argv %v", argv) + } + } +}