diff --git a/docs/contracts/workspace-preprocessing-cli.md b/docs/contracts/workspace-preprocessing-cli.md new file mode 100644 index 00000000..baa0fffd --- /dev/null +++ b/docs/contracts/workspace-preprocessing-cli.md @@ -0,0 +1,89 @@ +# Workspace preprocessing CLI contract + +`thothctl` is the only supported host entrypoint for workspace preprocessing. + +## Invocation + +```text +thothctl --installation /thothii-installation.yaml workspace inspect + --workspace [--json] + +thothctl --installation /thothii-installation.yaml workspace preprocess dwh + --workspace [--resume <32hex>] [--json] + +thothctl --installation /thothii-installation.yaml workspace schema suggest-fks + --workspace + [--from-sql ]... [--assume ]... + [--output ] [--json] + +thothctl --installation /thothii-installation.yaml workspace schema check + --workspace + [--annotations --reviewed-candidates ] + [--json] + +thothctl --installation /thothii-installation.yaml workspace index-schema + --workspace [--json] + +thothctl --installation /thothii-installation.yaml workspace preprocess evidence + --workspace [--dry-run] [--resume <32hex>] [--json] + +thothctl --installation /thothii-installation.yaml workspace preprocess run + --workspace [--resume <32hex>] [--json] +``` + +## Validation + +- `--installation` is mandatory and absolute. +- `--workspace` is mandatory exactly once and must match `[a-z][a-z0-9-]{2,62}`. +- `--resume` values must be 32 lowercase hex characters. +- `--json` may be supplied once. +- `schema suggest-fks` + - allows at most 32 `--from-sql` files; + - each SQL file must be a canonical regular file, UTF-8, non-symlink, max 1 MiB; + - total SQL ingress must not exceed 16 MiB; + - allows at most 256 `--assume` values, each `column=table`, max 256 bytes; + - `--output` must name a new canonical path; existing targets are refused. +- `schema check` + - `--annotations` and `--reviewed-candidates` are all-or-nothing; + - annotations must be UTF-8, canonical, non-symlink, max 16 MiB; + - `--reviewed-candidates` must match `sha256:<64 lowercase hex>`. +- Unknown flags, passthrough separators, and shell fragments are rejected before Docker runs. + +## Container boundary + +`thothctl` resolves the selected `core` image from the rendered installation, converts it to an immutable local image ID, writes a one-shot final override that pins both `core` and `workspace-maintenance` to that ID with `pull_policy: never`, and runs only: + +```text +docker compose run --rm --no-deps --no-TTY --name workspace-maintenance +``` + +The request is streamed as one schema-versioned JSON document over stdin. Public stdout is always one schema-versioned JSON result; human mode is rendered from an allowlisted subset of that same result. + +## Public JSON result + +```json +{ + "schemaVersion": 1, + "status": "succeeded|unchanged|dry_run|blocked|failed", + "code": "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", + "workspaceId": "abc", + "workspaceRevision": "1234567890abcdef1234567890abcdef12345678", + "descriptorBlob": "sha256:<64 lowercase hex>", + "operation": "inspect|preprocess-dwh|schema-suggest-fks|schema-check|index-schema|preprocess-evidence|preprocess-run", + "runId": "", + "childRuns": {"stage": ""}, + "completedStages": ["stage"], + "counts": {"name": 1}, + "artifactIdentities": [{"kind": "fk_candidates", "digest": "sha256:<64 lowercase hex>"}], + "warnings": ["safe warning"] +} +``` + +`thothctl --json` parses the operator stdout strictly and re-encodes only the public fields above. + +## Exit codes + +- `0`: `succeeded`, `unchanged`, or `dry_run` +- `3`: `blocked` +- `2`: host-side grammar or local file safety failure +- `1`: operational failure or operator-reported `failed` diff --git a/tools/thothctl/cmd/thothctl/main.go b/tools/thothctl/cmd/thothctl/main.go index 69d02447..0a364ebb 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,14 @@ 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] + Inspect the active registry snapshot for one workspace. + 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 [--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] ` func main() { @@ -154,6 +163,8 @@ func run(ctx context.Context, args []string, stdout, stderr io.Writer) int { } fmt.Fprintf(stdout, "Removed %d stopped app containers; verified %d preserved paths.\n", len(removal.Targets), removal.Preserved) return 0 + case "workspace": + return workspaceCommand(ctx, installation, runner, commandArgs, secretValues, stdout, stderr) default: return commandUsageError(stderr, fmt.Sprintf("unknown command %q", command)) } @@ -171,6 +182,47 @@ func writeRemovalTargets(outputWriter io.Writer, project string, targets []serve } } +func workspaceCommand(ctx context.Context, installation config.Installation, runner compose.Runner, args []string, secretValues []string, stdout, stderr io.Writer) int { + request, err := workspaceops.Parse(args) + if err != nil { + return commandUsageError(stderr, err.Error()) + } + result, err := workspaceops.Execute(ctx, installation, runner, request) + if err != nil { + return workspaceFailure(stderr, err, secretValues) + } + if request.JSONMode() { + encoder := json.NewEncoder(stdout) + encoder.SetEscapeHTML(false) + if encodeErr := encoder.Encode(result); encodeErr != nil { + fmt.Fprintln(stderr, "thothctl: workspace result could not be written") + return 1 + } + } else { + fmt.Fprint(stdout, workspaceops.Human(result)) + } + switch result.Status { + case "blocked": + return 3 + case "failed": + return 1 + default: + return 0 + } +} + +func workspaceFailure(stderr io.Writer, err error, secretValues []string) int { + message := output.Sanitize(err.Error(), secretValues) + var operationErr *workspaceops.OperationError + if errors.As(err, &operationErr) && operationErr.Detail() != "" { + detail := output.SanitizeDetail(operationErr.Detail(), secretValues) + fmt.Fprintf(stderr, "thothctl: %s: %s\n", message, detail) + } else { + fmt.Fprintf(stderr, "thothctl: %s\n", message) + } + return 1 +} + func serverOperationFailure(stderr io.Writer, err error, secretValues []string) int { message := output.Sanitize(err.Error(), secretValues) var operationErr *serverops.OperationError diff --git a/tools/thothctl/cmd/thothctl/main_test.go b/tools/thothctl/cmd/thothctl/main_test.go index 630abc1d..a956c662 100644 --- a/tools/thothctl/cmd/thothctl/main_test.go +++ b/tools/thothctl/cmd/thothctl/main_test.go @@ -664,6 +664,128 @@ func TestRunPiMaintenanceStatusAndRecoverConfirmationContract(t *testing.T) { assertDockerNotInvoked(t, second) } +func TestRunWorkspaceInspectDispatchesThroughTheMaintenanceService(t *testing.T) { + fixture := newCLIFixture(t, "SAFE_VALUE=1\n") + fixture.setEnvironment(t) + t.Setenv("THOTHCTL_FAKE_WORKSPACE_RESULT", `{"schemaVersion":1,"status":"succeeded","code":"ok","workspaceId":"abc","workspaceRevision":"1234567890abcdef1234567890abcdef12345678","descriptorBlob":"sha256:`+strings.Repeat("a", 64)+`","operation":"inspect","completedStages":[]}`) + + var stdout, stderr bytes.Buffer + exitCode := run(context.Background(), []string{"--installation", fixture.installationPath, "workspace", "inspect", "--workspace", "abc", "--json"}, &stdout, &stderr) + + if exitCode != 0 { + t.Fatalf("run() exit code = %d, stderr = %s", exitCode, stderr.String()) + } + if !strings.Contains(stdout.String(), `"workspaceId":"abc"`) || !strings.Contains(stdout.String(), `"operation":"inspect"`) { + t.Fatalf("stdout = %q", stdout.String()) + } + assertInvocationContains(t, fixture.invocations(t), "run", "--rm", "--no-deps", "--no-TTY", "--name") + assertInvocationContains(t, fixture.invocations(t), "workspace-maintenance", "inspect") +} + +func TestRunWorkspaceBlockedResultsExitThreeAndRenderHumanOutput(t *testing.T) { + fixture := newCLIFixture(t, "SAFE_VALUE=1\n") + fixture.setEnvironment(t) + t.Setenv("THOTHCTL_FAKE_WORKSPACE_RESULT", `{"schemaVersion":1,"status":"blocked","code":"manual_review_required","workspaceId":"abc","workspaceRevision":"1234567890abcdef1234567890abcdef12345678","descriptorBlob":"sha256:`+strings.Repeat("b", 64)+`","operation":"schema-suggest-fks","completedStages":["dwh"],"warnings":["review required"]}`) + + var stdout, stderr bytes.Buffer + exitCode := run(context.Background(), []string{"--installation", fixture.installationPath, "workspace", "schema", "suggest-fks", "--workspace", "abc"}, &stdout, &stderr) + + if exitCode != 3 { + t.Fatalf("run() exit code = %d, want 3", exitCode) + } + for _, expected := range []string{"workspace: abc", "status: blocked", "warning: review required"} { + if !strings.Contains(stdout.String(), expected) { + t.Fatalf("stdout = %q, missing %q", stdout.String(), expected) + } + } + if strings.Contains(stdout.String(), "descriptorBlob") || strings.Contains(stdout.String(), strings.Repeat("b", 64)) { + t.Fatalf("stdout leaked non-allowlisted fields: %q", stdout.String()) + } + if stderr.Len() != 0 { + t.Fatalf("stderr = %q", stderr.String()) + } +} + +func TestRunWorkspaceRequiresWorkspaceFlagBeforeDocker(t *testing.T) { + fixture := newCLIFixture(t, "SAFE_VALUE=1\n") + fixture.setEnvironment(t) + + var stdout, stderr bytes.Buffer + exitCode := run(context.Background(), []string{ + "--installation", fixture.installationPath, + "workspace", "inspect", + }, &stdout, &stderr) + + if exitCode != 2 { + t.Fatalf("run() exit code = %d, want 2", exitCode) + } + if !strings.Contains(stderr.String(), "--workspace") || !strings.Contains(stderr.String(), "required") { + t.Fatalf("stderr = %q", stderr.String()) + } + assertDockerNotInvoked(t, fixture) +} + +func TestRunWorkspaceRejectsDuplicateWorkspaceFlagsBeforeDocker(t *testing.T) { + fixture := newCLIFixture(t, "SAFE_VALUE=1\n") + fixture.setEnvironment(t) + + var stdout, stderr bytes.Buffer + exitCode := run(context.Background(), []string{ + "--installation", fixture.installationPath, + "workspace", "inspect", "--workspace", "abc", "--workspace", "def", + }, &stdout, &stderr) + + if exitCode != 2 { + t.Fatalf("run() exit code = %d, want 2", exitCode) + } + if !strings.Contains(stderr.String(), "--workspace") || !strings.Contains(stderr.String(), "exactly once") { + t.Fatalf("stderr = %q", stderr.String()) + } + assertDockerNotInvoked(t, fixture) +} + +func TestRunWorkspaceRejectsInvalidResumeRunIDsBeforeDocker(t *testing.T) { + fixture := newCLIFixture(t, "SAFE_VALUE=1\n") + fixture.setEnvironment(t) + + var stdout, stderr bytes.Buffer + exitCode := run(context.Background(), []string{ + "--installation", fixture.installationPath, + "workspace", "preprocess", "dwh", "--workspace", "abc", "--resume", "not-a-run-id", + }, &stdout, &stderr) + + if exitCode != 2 { + t.Fatalf("run() exit code = %d, want 2", exitCode) + } + if !strings.Contains(stderr.String(), "--resume") || !strings.Contains(stderr.String(), "32 lowercase hex") { + t.Fatalf("stderr = %q", stderr.String()) + } + assertDockerNotInvoked(t, fixture) +} + +func TestRunWorkspaceSchemaCheckRequiresReviewedCandidatesWithAnnotations(t *testing.T) { + fixture := newCLIFixture(t, "SAFE_VALUE=1\n") + fixture.setEnvironment(t) + annotations := filepath.Join(fixture.root, "annotations.yaml") + if err := os.WriteFile(annotations, []byte("reviewed: []\n"), 0o600); err != nil { + t.Fatal(err) + } + + var stdout, stderr bytes.Buffer + exitCode := run(context.Background(), []string{ + "--installation", fixture.installationPath, + "workspace", "schema", "check", "--workspace", "abc", "--annotations", annotations, + }, &stdout, &stderr) + + if exitCode != 2 { + t.Fatalf("run() exit code = %d, want 2", exitCode) + } + if !strings.Contains(stderr.String(), "--reviewed-candidates") { + t.Fatalf("stderr = %q", stderr.String()) + } + assertDockerNotInvoked(t, fixture) +} + func TestRunPiStatusPreservesDockerExitCodeAndRedactsDiagnostics(t *testing.T) { fixture := newCLIFixture(t, "PI_TOKEN_FILE=%s\n") secretPath := filepath.Join(fixture.root, "pi-secret") @@ -744,6 +866,7 @@ case " $* " in fi printf '%s\n' '{"applied":[],"drifted":[],"pending":[]}' ;; *" ps --format json "*) printf '%s\n' '[{"Service":"core","State":"running","Health":"healthy"},{"Service":"frontend","State":"running","Health":"healthy"}]' ;; + *" image inspect --format {{.Id}} "*) printf '%s\n' "${THOTHCTL_FAKE_IMAGE_ID:-sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb}" ;; *"io.thothii.pi.version"*) printf '%s\n' '0.80.3' ;; *"PI_VERSION"*) printf '%s\n' '0.80.3' ;; *" pi --version "*) printf '%s\n' '0.80.3' ;; @@ -752,6 +875,7 @@ case " $* " in *"/settings "*) printf '%s\n' '{"provider":"provider","model":"model","thinking":"medium"}' ;; *"/internal/maintenance/status "*) printf '%s\n' '{"active":true,"admissions":0}' ;; *" logs "*) printf '%s\n' "$THOTHCTL_FAKE_LOG" ;; + *" run --rm --no-deps --no-TTY "*" workspace-maintenance "*) printf '%s\n' "$THOTHCTL_FAKE_WORKSPACE_RESULT" ;; esac if [ "${THOTHCTL_FAKE_FAIL_ON:-}" = "version" ]; then printf '%s\n' "${THOTHCTL_FAKE_FAILURE:-fake Docker failure}" >&2 @@ -792,6 +916,8 @@ func (f cliFixture) setEnvContents(t *testing.T, env string) { t.Setenv("THOTHCTL_FAKE_CONFIG", "") t.Setenv("THOTHCTL_FAKE_MIGRATION_FAILURE", "") t.Setenv("THOTHCTL_FAKE_MIGRATION_EXIT", "0") + t.Setenv("THOTHCTL_FAKE_WORKSPACE_RESULT", "") + t.Setenv("THOTHCTL_FAKE_IMAGE_ID", "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb") } func (f cliFixture) setProfile(t *testing.T, profile string) { diff --git a/tools/thothctl/internal/config/installation_test.go b/tools/thothctl/internal/config/installation_test.go index d12e918d..4be55207 100644 --- a/tools/thothctl/internal/config/installation_test.go +++ b/tools/thothctl/internal/config/installation_test.go @@ -178,7 +178,16 @@ 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") + temporaryRoot, err := filepath.EvalSymlinks(os.TempDir()) + if err != nil { + t.Fatal(err) + } + physicalRoot, err := os.MkdirTemp(temporaryRoot, "thothctl-config-") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = os.RemoveAll(physicalRoot) }) + root := filepath.Join(physicalRoot, "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/pi/update.go b/tools/thothctl/internal/pi/update.go index 01515532..2f5405af 100644 --- a/tools/thothctl/internal/pi/update.go +++ b/tools/thothctl/internal/pi/update.go @@ -743,7 +743,12 @@ func writeLifecycleOverride(path, image string) error { if err != nil { return errors.New("lifecycle image override could not be encoded") } - contents := []byte("services:\n core:\n image: " + string(quoted) + "\n") + contents := []byte(`services: + core: + image: ` + string(quoted) + ` + workspace-maintenance: + image: ` + string(quoted) + ` +`) if err := writeFileDurably(path, ".pi-lifecycle-", contents); err != nil { return errors.New("lifecycle image override could not be written durably") } diff --git a/tools/thothctl/internal/pi/update_test.go b/tools/thothctl/internal/pi/update_test.go index 2e36d434..9f95966a 100644 --- a/tools/thothctl/internal/pi/update_test.go +++ b/tools/thothctl/internal/pi/update_test.go @@ -126,6 +126,24 @@ func TestSuccessfulUpdateAndRollbackRemainSelectedOnFreshRecreate(t *testing.T) } } +func TestWriteLifecycleOverridePinsCoreAndWorkspaceMaintenanceToTheSameImage(t *testing.T) { + path := filepath.Join(t.TempDir(), "current-image.yaml") + if err := writeLifecycleOverride(path, "sha256:"+strings.Repeat("a", 64)); err != nil { + t.Fatal(err) + } + contents, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + text := string(contents) + wantedImage := `image: "sha256:` + strings.Repeat("a", 64) + `"` + for _, expected := range []string{"services:", "core:", "workspace-maintenance:", wantedImage} { + if !strings.Contains(text, expected) { + t.Fatalf("override = %q, missing %q", text, expected) + } + } +} + func TestSelectorPromotionDoesNotMaskPostRenameDirectoryFsyncFailure(t *testing.T) { directory := t.TempDir() source := filepath.Join(directory, "candidate.yaml") diff --git a/tools/thothctl/internal/safeio/files.go b/tools/thothctl/internal/safeio/files.go index b83339cd..86553a16 100644 --- a/tools/thothctl/internal/safeio/files.go +++ b/tools/thothctl/internal/safeio/files.go @@ -1,4 +1,4 @@ -// Package safeio reads installation files without following symlinked path components. +// Package safeio reads and writes local files without following symlinked path components. package safeio import ( @@ -7,6 +7,7 @@ import ( "os" "path/filepath" "strings" + "unicode/utf8" ) var ErrUnsafeFile = errors.New("unsafe file") @@ -19,17 +20,83 @@ func ValidateCanonicalPath(path string) error { return nil } -func readBoundedRegularFile(file *os.File, maximum int64) ([]byte, error) { +func readBoundedRegularFile(path string, file *os.File, maximum int64) ([]byte, error) { if maximum < 0 || maximum == int64(^uint64(0)>>1) { return nil, ErrUnsafeFile } info, err := file.Stat() - if err != nil || !info.Mode().IsRegular() { + if err != nil || !info.Mode().IsRegular() || !hasSingleLink(info) { return nil, ErrUnsafeFile } contents, err := io.ReadAll(io.LimitReader(file, maximum+1)) if err != nil || int64(len(contents)) > maximum { return nil, ErrUnsafeFile } + after, err := file.Stat() + if err != nil || !after.Mode().IsRegular() || !hasSingleLink(after) || !os.SameFile(info, after) { + return nil, ErrUnsafeFile + } + current, err := os.Stat(path) + if err != nil || !os.SameFile(info, current) { + return nil, ErrUnsafeFile + } return contents, nil } + +func ReadCanonicalUTF8(path string, maximum int64) (string, error) { + contents, err := ReadCanonicalRegular(path, maximum) + if err != nil { + return "", err + } + if !utf8.Valid(contents) { + return "", ErrUnsafeFile + } + return string(contents), nil +} + +func WriteCanonicalNewFile(path string, contents []byte, mode os.FileMode) error { + if err := ValidateCanonicalPath(path); err != nil { + return err + } + parent := filepath.Dir(path) + if err := requireCanonicalDirectory(parent); err != nil { + return err + } + if info, err := os.Lstat(path); err == nil { + if !info.Mode().IsRegular() || info.Mode()&os.ModeSymlink != 0 || info.Mode()&os.ModeType != 0 { + return ErrUnsafeFile + } + return ErrUnsafeFile + } else if !errors.Is(err, os.ErrNotExist) { + return ErrUnsafeFile + } + file, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, mode) + if err != nil { + return ErrUnsafeFile + } + defer file.Close() + if _, err := file.Write(contents); err != nil { + _ = os.Remove(path) + return ErrUnsafeFile + } + if err := file.Sync(); err != nil { + _ = os.Remove(path) + return ErrUnsafeFile + } + return nil +} + +func requireCanonicalDirectory(path string) error { + if err := ValidateCanonicalPath(path); err != nil { + return err + } + resolved, err := filepath.EvalSymlinks(path) + if err != nil || resolved != path { + return ErrUnsafeFile + } + info, err := os.Stat(path) + if err != nil || !info.IsDir() { + return ErrUnsafeFile + } + return nil +} diff --git a/tools/thothctl/internal/safeio/files_test.go b/tools/thothctl/internal/safeio/files_test.go index 67d4d0fa..8c510113 100644 --- a/tools/thothctl/internal/safeio/files_test.go +++ b/tools/thothctl/internal/safeio/files_test.go @@ -41,3 +41,55 @@ func TestReadCanonicalRegularRejectsFinalAndParentSymlinks(t *testing.T) { t.Fatalf("final symlink error = %v, want ErrUnsafeFile", err) } } + +func TestReadCanonicalUTF8RejectsNonUTF8AndHardlinks(t *testing.T) { + temporaryRoot, err := filepath.EvalSymlinks(os.TempDir()) + if err != nil { + t.Fatal(err) + } + root, err := os.MkdirTemp(temporaryRoot, "thothctl-safeio-") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = os.RemoveAll(root) }) + + nonUTF8 := filepath.Join(root, "annotations.yaml") + if err := os.WriteFile(nonUTF8, []byte{0xff, 0xfe, 0xfd}, 0o600); err != nil { + t.Fatal(err) + } + if _, err := ReadCanonicalUTF8(nonUTF8, 1024); !errors.Is(err, ErrUnsafeFile) { + t.Fatalf("ReadCanonicalUTF8(nonUTF8) error = %v, want ErrUnsafeFile", err) + } + + target := filepath.Join(root, "regular.txt") + if err := os.WriteFile(target, []byte("linked"), 0o600); err != nil { + t.Fatal(err) + } + link := filepath.Join(root, "hardlink.txt") + if err := os.Link(target, link); err != nil { + t.Fatal(err) + } + if _, err := ReadCanonicalUTF8(link, 1024); !errors.Is(err, ErrUnsafeFile) { + t.Fatalf("ReadCanonicalUTF8(hardlink) error = %v, want ErrUnsafeFile", err) + } +} + +func TestWriteCanonicalNewFileRejectsExistingTargets(t *testing.T) { + temporaryRoot, err := filepath.EvalSymlinks(os.TempDir()) + if err != nil { + t.Fatal(err) + } + root, err := os.MkdirTemp(temporaryRoot, "thothctl-safeio-") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = os.RemoveAll(root) }) + + path := filepath.Join(root, "artifact.yaml") + if err := os.WriteFile(path, []byte("existing"), 0o600); err != nil { + t.Fatal(err) + } + if err := WriteCanonicalNewFile(path, []byte("new"), 0o600); !errors.Is(err, ErrUnsafeFile) { + t.Fatalf("WriteCanonicalNewFile(existing) error = %v, want ErrUnsafeFile", err) + } +} diff --git a/tools/thothctl/internal/safeio/files_unix.go b/tools/thothctl/internal/safeio/files_unix.go index c3a745b9..05e999d2 100644 --- a/tools/thothctl/internal/safeio/files_unix.go +++ b/tools/thothctl/internal/safeio/files_unix.go @@ -47,7 +47,7 @@ func ReadCanonicalRegular(path string, maximum int64) ([]byte, error) { return nil, ErrUnsafeFile } defer file.Close() - return readBoundedRegularFile(file, maximum) + return readBoundedRegularFile(path, file, maximum) } func closeUnixDescriptors(descriptors []int) { diff --git a/tools/thothctl/internal/safeio/files_windows.go b/tools/thothctl/internal/safeio/files_windows.go index ce40ae03..8227c71e 100644 --- a/tools/thothctl/internal/safeio/files_windows.go +++ b/tools/thothctl/internal/safeio/files_windows.go @@ -53,7 +53,7 @@ func ReadCanonicalRegular(path string, maximum int64) ([]byte, error) { return nil, ErrUnsafeFile } defer file.Close() - return readBoundedRegularFile(file, maximum) + return readBoundedRegularFile(path, file, maximum) } func openWindowsComponent(path string, directory bool) (windows.Handle, error) { diff --git a/tools/thothctl/internal/safeio/linkcount_unix.go b/tools/thothctl/internal/safeio/linkcount_unix.go new file mode 100644 index 00000000..ab4e8854 --- /dev/null +++ b/tools/thothctl/internal/safeio/linkcount_unix.go @@ -0,0 +1,13 @@ +//go:build !windows + +package safeio + +import ( + "os" + "syscall" +) + +func hasSingleLink(info os.FileInfo) bool { + stat, ok := info.Sys().(*syscall.Stat_t) + return ok && stat.Nlink == 1 +} diff --git a/tools/thothctl/internal/safeio/linkcount_windows.go b/tools/thothctl/internal/safeio/linkcount_windows.go new file mode 100644 index 00000000..dc414609 --- /dev/null +++ b/tools/thothctl/internal/safeio/linkcount_windows.go @@ -0,0 +1,9 @@ +//go:build windows + +package safeio + +import "os" + +func hasSingleLink(info os.FileInfo) bool { + return true +} diff --git a/tools/thothctl/internal/serverops/operations_test.go b/tools/thothctl/internal/serverops/operations_test.go index d1f6906f..22e219b2 100644 --- a/tools/thothctl/internal/serverops/operations_test.go +++ b/tools/thothctl/internal/serverops/operations_test.go @@ -294,7 +294,15 @@ func TestRemoveRejectsConfirmationForDifferentContainerIDs(t *testing.T) { func testInstallation(t *testing.T) config.Installation { t.Helper() - root := t.TempDir() + temporaryRoot, err := filepath.EvalSymlinks(os.TempDir()) + if err != nil { + t.Fatal(err) + } + root, err := os.MkdirTemp(temporaryRoot, "thothctl-serverops-") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = os.RemoveAll(root) }) 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 new file mode 100644 index 00000000..8a2ca241 --- /dev/null +++ b/tools/thothctl/internal/workspaceops/operations.go @@ -0,0 +1,772 @@ +// Package workspaceops implements the closed host-side workspace preprocessing contract. +package workspaceops + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "regexp" + "sort" + "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" +) + +var ( + workspacePattern = regexp.MustCompile(`^[a-z][a-z0-9-]{2,62}$`) + runIDPattern = regexp.MustCompile(`^[0-9a-f]{32}$`) + reviewedCandidatesDigest = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`) +) + +const ( + maxFromSQLFiles = 32 + maxAssumptions = 256 +) + +type Runner interface { + Run(context.Context, []string, io.Reader) (compose.Result, error) +} + +type Request interface { + workspaceRequest() + workspaceID() string + JSONMode() bool + operatorCommand() string + stdinEnvelope() (requestEnvelope, error) +} + +type baseRequest struct { + Workspace string + JSON bool +} + +func (b baseRequest) workspaceID() string { return b.Workspace } +func (b baseRequest) JSONMode() bool { return b.JSON } + +type InspectRequest struct{ baseRequest } + +type DwhRequest struct { + baseRequest + Resume string +} + +type SuggestFksRequest struct { + baseRequest + FromSQL []string + Assume []string + Output string +} + +type CheckSchemaRequest struct { + baseRequest + Annotations string + ReviewedCandidates string +} + +type IndexSchemaRequest struct{ baseRequest } + +type EvidenceRequest struct { + baseRequest + DryRun bool + Resume string +} + +type RunRequest struct { + baseRequest + Resume string +} + +func (InspectRequest) workspaceRequest() {} +func (DwhRequest) workspaceRequest() {} +func (SuggestFksRequest) workspaceRequest() {} +func (CheckSchemaRequest) workspaceRequest() {} +func (IndexSchemaRequest) workspaceRequest() {} +func (EvidenceRequest) workspaceRequest() {} +func (RunRequest) workspaceRequest() {} + +func (InspectRequest) operatorCommand() string { return "inspect" } +func (DwhRequest) operatorCommand() string { return "preprocess-dwh" } +func (SuggestFksRequest) operatorCommand() string { return "schema-suggest-fks" } +func (CheckSchemaRequest) operatorCommand() string { + return "schema-check" +} +func (IndexSchemaRequest) operatorCommand() string { return "index-schema" } +func (EvidenceRequest) operatorCommand() string { return "preprocess-evidence" } +func (RunRequest) operatorCommand() string { return "preprocess-run" } + +func (r InspectRequest) stdinEnvelope() (requestEnvelope, error) { + return requestEnvelope{SchemaVersion: 1, Operation: r.operatorCommand(), WorkspaceID: r.Workspace}, nil +} + +func (r DwhRequest) stdinEnvelope() (requestEnvelope, error) { + return requestEnvelope{SchemaVersion: 1, Operation: r.operatorCommand(), WorkspaceID: r.Workspace, Resume: r.Resume}, nil +} + +func (r SuggestFksRequest) stdinEnvelope() (requestEnvelope, error) { + envelope := requestEnvelope{SchemaVersion: 1, Operation: r.operatorCommand(), WorkspaceID: r.Workspace, Assume: append([]string(nil), r.Assume...)} + totalBytes := 0 + for _, path := range r.FromSQL { + contents, err := safeio.ReadCanonicalUTF8(path, 1<<20) + if err != nil { + return requestEnvelope{}, errors.New("SQL input could not be read safely") + } + totalBytes += len(contents) + if totalBytes > 16<<20 { + return requestEnvelope{}, errors.New("SQL input total exceeds 16 MiB") + } + envelope.SQLFiles = append(envelope.SQLFiles, inputFile{Path: path, Contents: contents}) + } + return envelope, nil +} + +func (r CheckSchemaRequest) stdinEnvelope() (requestEnvelope, error) { + annotations, err := safeio.ReadCanonicalUTF8(r.Annotations, 16<<20) + if err != nil { + return requestEnvelope{}, errors.New("annotation file could not be read safely") + } + return requestEnvelope{ + SchemaVersion: 1, + Operation: r.operatorCommand(), + WorkspaceID: r.Workspace, + Annotations: annotations, + ReviewedCandidates: r.ReviewedCandidates, + }, nil +} + +func (r IndexSchemaRequest) stdinEnvelope() (requestEnvelope, error) { + return requestEnvelope{SchemaVersion: 1, Operation: r.operatorCommand(), WorkspaceID: r.Workspace}, nil +} + +func (r EvidenceRequest) stdinEnvelope() (requestEnvelope, error) { + return requestEnvelope{SchemaVersion: 1, Operation: r.operatorCommand(), WorkspaceID: r.Workspace, Resume: r.Resume, DryRun: r.DryRun}, nil +} + +func (r RunRequest) stdinEnvelope() (requestEnvelope, error) { + return requestEnvelope{SchemaVersion: 1, Operation: r.operatorCommand(), WorkspaceID: r.Workspace, Resume: r.Resume}, nil +} + +type requestEnvelope struct { + SchemaVersion int `json:"schemaVersion"` + Operation string `json:"operation"` + WorkspaceID string `json:"workspaceId"` + Resume string `json:"resume,omitempty"` + DryRun bool `json:"dryRun,omitempty"` + Assume []string `json:"assume,omitempty"` + SQLFiles []inputFile `json:"sqlFiles,omitempty"` + Annotations string `json:"annotations,omitempty"` + ReviewedCandidates string `json:"reviewedCandidates,omitempty"` +} + +type inputFile struct { + Path string `json:"path"` + Contents string `json:"contents"` +} + +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"` +} + +type operationResponse struct { + Result + SuggestedFksYAML string `json:"suggestedFksYaml,omitempty"` +} + +type Stage string + +type ExitClass string + +const ( + StageRenderedConfig Stage = "rendered-config" + StageImageInspect Stage = "image-inspect" + StageComposeRun Stage = "compose-run" + + ExitClassNonzero ExitClass = "nonzero-exit" + ExitClassUnavailable ExitClass = "unavailable" + ExitClassTimeout ExitClass = "timeout" + ExitClassInvocation ExitClass = "invocation-failure" +) + +type OperationError struct { + stage Stage + class ExitClass + detail string +} + +func (e *OperationError) Error() string { + return fmt.Sprintf("stage=%s class=%s", e.stage, e.class) +} + +func (e *OperationError) Stage() Stage { return e.stage } +func (e *OperationError) Class() ExitClass { return e.class } +func (e *OperationError) Detail() string { return e.detail } + +func Parse(args []string) (Request, error) { + if len(args) == 0 { + return nil, errors.New("workspace requires a subcommand") + } + switch args[0] { + case "inspect": + parsed, err := parseInspect(args[1:]) + if err != nil { + return nil, err + } + return parsed, nil + case "preprocess": + return parsePreprocess(args[1:]) + case "schema": + return parseSchema(args[1:]) + case "index-schema": + parsed, err := parseIndexSchema(args[1:]) + if err != nil { + return nil, err + } + return parsed, nil + default: + return nil, fmt.Errorf("unknown workspace command %q", args[0]) + } +} + +func Execute(ctx context.Context, installation config.Installation, runner Runner, request Request) (Result, error) { + envelope, err := request.stdinEnvelope() + if err != nil { + return Result{}, err + } + rendered, err := runDocker(ctx, runner, StageRenderedConfig, installation.ComposeArgs("config", "--format", "json")) + if err != nil { + return Result{}, err + } + imageReference, err := selectedCoreImage(rendered.Stdout) + if err != nil { + return Result{}, err + } + imageID, err := immutableImageID(ctx, runner, imageReference) + if err != nil { + return Result{}, err + } + override, cleanup, err := maintenanceOverride(installation, imageID) + if err != nil { + return Result{}, err + } + defer cleanup() + stdin, err := encodeEnvelope(envelope) + if err != nil { + return Result{}, err + } + args, err := installation.ComposeArgsWithFinalOverride( + override, + "run", "--rm", "--no-deps", "--no-TTY", "--name", ownedContainerName(installation, request), "workspace-maintenance", request.operatorCommand(), + ) + if err != nil { + return Result{}, err + } + result, err := runDocker(ctx, runner, StageComposeRun, args, bytes.NewReader(stdin)) + if err != nil { + return Result{}, err + } + response, err := parseResponse(result.Stdout) + if err != nil { + return Result{}, err + } + if suggest, ok := request.(SuggestFksRequest); ok && suggest.Output != "" { + if response.SuggestedFksYAML == "" { + return Result{}, errors.New("workspace maintenance did not return the requested FK artifact") + } + if digest := suggestedArtifactDigest(response); digest != "" { + sum := sha256.Sum256([]byte(response.SuggestedFksYAML)) + if digest != "sha256:"+fmt.Sprintf("%x", sum[:]) { + return Result{}, errors.New("workspace maintenance returned an FK artifact with a mismatched digest") + } + } + if err := safeio.WriteCanonicalNewFile(suggest.Output, []byte(response.SuggestedFksYAML), 0o600); err != nil { + return Result{}, errors.New("workspace FK output file could not be created safely") + } + } + return response.Result, nil +} + +func encodeEnvelope(envelope requestEnvelope) ([]byte, error) { + encoded, err := json.Marshal(envelope) + if err != nil { + return nil, errors.New("workspace request could not be encoded") + } + if len(encoded) > 1<<20 { + return nil, errors.New("workspace request exceeds the bounded stdin contract") + } + return append(encoded, '\n'), nil +} + +func parseResponse(document string) (operationResponse, error) { + decoder := json.NewDecoder(strings.NewReader(document)) + decoder.DisallowUnknownFields() + var response operationResponse + if err := decoder.Decode(&response); err != nil { + return operationResponse{}, errors.New("workspace maintenance returned invalid JSON") + } + var extra any + if err := decoder.Decode(&extra); !errors.Is(err, io.EOF) { + return operationResponse{}, errors.New("workspace maintenance returned trailing output") + } + if err := validateResult(response.Result); err != nil { + return operationResponse{}, err + } + return response, nil +} + +func validateResult(result Result) error { + if result.SchemaVersion != 1 { + return errors.New("workspace maintenance returned an unsupported schema version") + } + if !workspacePattern.MatchString(result.WorkspaceID) { + return errors.New("workspace maintenance returned an invalid workspace identity") + } + if len(result.WorkspaceRevision) != 40 || !isLowerHex(result.WorkspaceRevision) { + return errors.New("workspace maintenance returned an invalid workspace revision") + } + if !strings.HasPrefix(result.DescriptorBlob, "sha256:") || len(result.DescriptorBlob) != len("sha256:")+64 || !isLowerHex(strings.TrimPrefix(result.DescriptorBlob, "sha256:")) { + return errors.New("workspace maintenance returned an invalid descriptor digest") + } + validStatuses := map[string]struct{}{"succeeded": {}, "unchanged": {}, "dry_run": {}, "blocked": {}, "failed": {}} + if _, ok := validStatuses[result.Status]; !ok { + return errors.New("workspace maintenance returned an invalid status") + } + if strings.TrimSpace(result.Code) == "" || strings.TrimSpace(result.Operation) == "" || result.CompletedStages == nil { + return errors.New("workspace maintenance omitted required fields") + } + for _, digest := range result.ArtifactIdentities { + if strings.TrimSpace(digest.Kind) == "" || !strings.HasPrefix(digest.Digest, "sha256:") { + return errors.New("workspace maintenance returned an invalid artifact identity") + } + } + return nil +} + +func isLowerHex(value string) bool { + for _, r := range value { + if !(r >= '0' && r <= '9' || r >= 'a' && r <= 'f') { + return false + } + } + return value != "" +} + +func parseInspect(args []string) (InspectRequest, error) { + base, err := parseBaseFlags(args, false) + if err != nil { + return InspectRequest{}, err + } + return InspectRequest{baseRequest: base}, nil +} + +func parsePreprocess(args []string) (Request, error) { + if len(args) == 0 { + return nil, errors.New("workspace preprocess requires dwh, evidence, or run") + } + switch args[0] { + case "dwh": + base, resume, dryRun, err := parseResumeFlags(args[1:], false) + if err != nil { + return nil, err + } + if dryRun { + return nil, errors.New("workspace preprocess dwh does not accept --dry-run") + } + return DwhRequest{baseRequest: base, Resume: resume}, nil + case "evidence": + base, resume, dryRun, err := parseResumeFlags(args[1:], true) + if err != nil { + return nil, err + } + return EvidenceRequest{baseRequest: base, Resume: resume, DryRun: dryRun}, nil + case "run": + base, resume, dryRun, err := parseResumeFlags(args[1:], false) + if err != nil { + return nil, err + } + if dryRun { + return nil, errors.New("workspace preprocess run does not accept --dry-run") + } + return RunRequest{baseRequest: base, Resume: resume}, nil + default: + return nil, fmt.Errorf("unknown workspace preprocess command %q", args[0]) + } +} + +func parseSchema(args []string) (Request, error) { + if len(args) == 0 { + return nil, errors.New("workspace schema requires suggest-fks or check") + } + switch args[0] { + case "suggest-fks": + return parseSuggestFks(args[1:]) + case "check": + return parseSchemaCheck(args[1:]) + default: + return nil, fmt.Errorf("unknown workspace schema command %q", args[0]) + } +} + +func parseIndexSchema(args []string) (IndexSchemaRequest, error) { + base, err := parseBaseFlags(args, false) + if err != nil { + return IndexSchemaRequest{}, err + } + return IndexSchemaRequest{baseRequest: base}, nil +} + +func parseResumeFlags(args []string, allowDryRun bool) (baseRequest, string, bool, error) { + var resume string + var dryRun bool + base, seen, err := parseSharedFlags(args, map[string]func(string) error{ + "--resume": func(value string) error { + if resume != "" { + return errors.New("--resume may be supplied once") + } + if !runIDPattern.MatchString(value) { + return errors.New("--resume must be 32 lowercase hex characters") + } + resume = value + return nil + }, + }, map[string]func() error{ + "--dry-run": func() error { + if !allowDryRun { + return errors.New("--dry-run is not accepted here") + } + if dryRun { + return errors.New("--dry-run may be supplied once") + } + dryRun = true + return nil + }, + }) + if err != nil { + return baseRequest{}, "", false, err + } + if !seen.workspace { + return baseRequest{}, "", false, errors.New("--workspace is required") + } + return base, resume, dryRun, nil +} + +func parseSuggestFks(args []string) (SuggestFksRequest, error) { + request := SuggestFksRequest{} + base, seen, err := parseSharedFlags(args, map[string]func(string) error{ + "--from-sql": func(value string) error { + if len(request.FromSQL) >= maxFromSQLFiles { + return fmt.Errorf("--from-sql may be supplied at most %d times", maxFromSQLFiles) + } + request.FromSQL = append(request.FromSQL, value) + return nil + }, + "--assume": func(value string) error { + if len(request.Assume) >= maxAssumptions { + return fmt.Errorf("--assume may be supplied at most %d times", maxAssumptions) + } + if len(value) > 256 || !strings.Contains(value, "=") { + return errors.New("--assume values must be column=table entries up to 256 bytes") + } + left, right, _ := strings.Cut(value, "=") + if strings.TrimSpace(left) == "" || strings.TrimSpace(right) == "" { + return errors.New("--assume values must be column=table entries up to 256 bytes") + } + request.Assume = append(request.Assume, value) + return nil + }, + "--output": func(value string) error { + if request.Output != "" { + return errors.New("--output may be supplied once") + } + request.Output = value + return nil + }, + }, nil) + if err != nil { + return SuggestFksRequest{}, err + } + if !seen.workspace { + return SuggestFksRequest{}, errors.New("--workspace is required") + } + request.baseRequest = base + return request, nil +} + +func parseSchemaCheck(args []string) (CheckSchemaRequest, error) { + request := CheckSchemaRequest{} + base, seen, err := parseSharedFlags(args, map[string]func(string) error{ + "--annotations": func(value string) error { + if request.Annotations != "" { + return errors.New("--annotations may be supplied once") + } + request.Annotations = value + return nil + }, + "--reviewed-candidates": func(value string) error { + if request.ReviewedCandidates != "" { + return errors.New("--reviewed-candidates may be supplied once") + } + if !reviewedCandidatesDigest.MatchString(value) { + return errors.New("--reviewed-candidates must be sha256:<64 lowercase hex>") + } + request.ReviewedCandidates = value + return nil + }, + }, nil) + if err != nil { + return CheckSchemaRequest{}, err + } + if !seen.workspace { + return CheckSchemaRequest{}, errors.New("--workspace is required") + } + if (request.Annotations == "") != (request.ReviewedCandidates == "") { + return CheckSchemaRequest{}, errors.New("--annotations and --reviewed-candidates must be supplied together") + } + request.baseRequest = base + return request, nil +} + +func parseBaseFlags(args []string, allowDryRun bool) (baseRequest, error) { + base, seen, err := parseSharedFlags(args, nil, nil) + if err != nil { + return baseRequest{}, err + } + if !seen.workspace { + return baseRequest{}, errors.New("--workspace is required") + } + return base, nil +} + +type seenFlags struct { + workspace bool + json bool +} + +func parseSharedFlags(args []string, valueHandlers map[string]func(string) error, boolHandlers map[string]func() error) (baseRequest, seenFlags, error) { + request := baseRequest{} + seen := seenFlags{} + valueHandlers = cloneValueHandlers(valueHandlers) + boolHandlers = cloneBoolHandlers(boolHandlers) + for len(args) > 0 { + flag := args[0] + if flag == "--" { + return baseRequest{}, seenFlags{}, errors.New("passthrough separators are not supported") + } + switch flag { + case "--workspace": + if len(args) < 2 { + return baseRequest{}, seenFlags{}, errors.New("--workspace requires a value") + } + if seen.workspace { + return baseRequest{}, seenFlags{}, errors.New("--workspace must be supplied exactly once") + } + workspace := args[1] + if !workspacePattern.MatchString(workspace) { + return baseRequest{}, seenFlags{}, errors.New("--workspace must match [a-z][a-z0-9-]{2,62}") + } + request.Workspace, seen.workspace, args = workspace, true, args[2:] + case "--json": + if seen.json { + return baseRequest{}, seenFlags{}, errors.New("--json may be supplied once") + } + request.JSON, seen.json, args = true, true, args[1:] + default: + if handler, ok := boolHandlers[flag]; ok { + if err := handler(); err != nil { + return baseRequest{}, seenFlags{}, err + } + args = args[1:] + continue + } + handler, ok := valueHandlers[flag] + if !ok { + return baseRequest{}, seenFlags{}, fmt.Errorf("unknown workspace option %q", flag) + } + if len(args) < 2 { + return baseRequest{}, seenFlags{}, fmt.Errorf("%s requires a value", flag) + } + if err := handler(args[1]); err != nil { + return baseRequest{}, seenFlags{}, err + } + args = args[2:] + } + } + return request, seen, nil +} + +func cloneValueHandlers(source map[string]func(string) error) map[string]func(string) error { + if len(source) == 0 { + return map[string]func(string) error{} + } + clone := make(map[string]func(string) error, len(source)) + for key, handler := range source { + clone[key] = handler + } + return clone +} + +func cloneBoolHandlers(source map[string]func() error) map[string]func() error { + if len(source) == 0 { + return map[string]func() error{} + } + clone := make(map[string]func() error, len(source)) + for key, handler := range source { + clone[key] = handler + } + return clone +} + +func selectedCoreImage(document string) (string, error) { + var rendered struct { + Services map[string]struct { + Image string `json:"image"` + } `json:"services"` + } + if err := json.Unmarshal([]byte(document), &rendered); err != nil { + return "", errors.New("rendered Compose configuration is invalid") + } + core, exists := rendered.Services["core"] + if !exists || strings.TrimSpace(core.Image) == "" { + return "", errors.New("selected core image is unavailable") + } + return core.Image, nil +} + +func immutableImageID(ctx context.Context, runner Runner, reference string) (string, error) { + result, err := runDocker(ctx, runner, StageImageInspect, []string{"image", "inspect", "--format", "{{.Id}}", reference}) + if err != nil { + return "", err + } + id := strings.TrimSpace(result.Stdout) + if !strings.HasPrefix(id, "sha256:") || len(id) != len("sha256:")+64 || !isLowerHex(strings.TrimPrefix(id, "sha256:")) { + return "", errors.New("selected core image did not resolve to an immutable sha256 image id") + } + return id, nil +} + +func maintenanceOverride(installation config.Installation, imageID string) (string, func(), error) { + control := installation.ControlDirectory() + if err := os.MkdirAll(control, 0o700); err != nil { + return "", func() {}, errors.New("workspace maintenance control directory could not be created") + } + info, err := os.Lstat(control) + if err != nil || !info.IsDir() || info.Mode()&os.ModeSymlink != 0 { + return "", func() {}, errors.New("workspace maintenance control directory is unsafe") + } + directory, err := os.MkdirTemp(control, "workspace-maintenance-") + if err != nil { + return "", func() {}, errors.New("workspace maintenance override directory could not be created") + } + cleanup := func() { + _ = os.Remove(filepath.Join(directory, "override.yaml")) + _ = os.Remove(directory) + } + path := filepath.Join(directory, "override.yaml") + contents := []string{ + "services:", + " core:", + " image: " + strconvQuote(imageID), + " pull_policy: never", + " workspace-maintenance:", + " image: " + strconvQuote(imageID), + " pull_policy: never", + "", + } + if err := os.WriteFile(path, []byte(strings.Join(contents, "\n")), 0o600); err != nil { + cleanup() + return "", func() {}, errors.New("workspace maintenance override could not be written") + } + return path, cleanup, nil +} + +func strconvQuote(value string) string { + encoded, _ := json.Marshal(value) + return string(encoded) +} + +func ownedContainerName(installation config.Installation, request Request) string { + parts := []string{installation.ProjectName(), request.workspaceID(), request.operatorCommand()} + for index, value := range parts { + parts[index] = strings.NewReplacer("/", "-", ":", "-", "@", "-", "_", "-").Replace(value) + } + return strings.Join(parts, "-") +} + +func runDocker(ctx context.Context, runner Runner, stage Stage, args []string, stdin ...io.Reader) (compose.Result, error) { + var input io.Reader + if len(stdin) > 0 { + input = stdin[0] + } + result, err := runner.Run(ctx, args, input) + if err != nil { + class := ExitClassInvocation + switch { + case errors.Is(ctx.Err(), context.DeadlineExceeded): + class = ExitClassTimeout + case result.ExitCode == 127: + class = ExitClassUnavailable + case result.ExitCode != 0: + class = ExitClassNonzero + } + detail := result.Stderr + if strings.TrimSpace(detail) == "" { + detail = err.Error() + } + return result, &OperationError{stage: stage, class: class, detail: detail} + } + return result, nil +} + +func suggestedArtifactDigest(response operationResponse) string { + for _, artifact := range response.ArtifactIdentities { + if strings.HasPrefix(artifact.Digest, "sha256:") { + return artifact.Digest + } + } + return "" +} + +func Human(result Result) string { + lines := []string{ + fmt.Sprintf("workspace: %s", result.WorkspaceID), + fmt.Sprintf("operation: %s", result.Operation), + fmt.Sprintf("status: %s", result.Status), + fmt.Sprintf("code: %s", result.Code), + fmt.Sprintf("revision: %s", result.WorkspaceRevision), + } + if result.RunID != "" { + lines = append(lines, fmt.Sprintf("run: %s", result.RunID)) + } + if len(result.CompletedStages) > 0 { + stages := append([]string(nil), result.CompletedStages...) + sort.Strings(stages) + lines = append(lines, fmt.Sprintf("completed: %s", strings.Join(stages, ", "))) + } + for _, warning := range result.Warnings { + lines = append(lines, fmt.Sprintf("warning: %s", warning)) + } + return strings.Join(lines, "\n") + "\n" +} diff --git a/tools/thothctl/internal/workspaceops/operations_test.go b/tools/thothctl/internal/workspaceops/operations_test.go new file mode 100644 index 00000000..1f734fc3 --- /dev/null +++ b/tools/thothctl/internal/workspaceops/operations_test.go @@ -0,0 +1,203 @@ +package workspaceops + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "os" + "path/filepath" + "reflect" + "strings" + "testing" + + "github.com/aritmolab/thothii/tools/thothctl/internal/compose" + "github.com/aritmolab/thothii/tools/thothctl/internal/config" +) + +type fakeRunner struct { + run func(args []string, stdin string) (compose.Result, error) + all [][]string + stdins []string +} + +func (r *fakeRunner) Run(_ context.Context, args []string, stdin io.Reader) (compose.Result, error) { + payload := "" + if stdin != nil { + bytes, err := io.ReadAll(stdin) + if err != nil { + return compose.Result{}, err + } + payload = string(bytes) + } + r.all = append(r.all, append([]string(nil), args...)) + r.stdins = append(r.stdins, payload) + return r.run(args, payload) +} + +func TestExecuteSuggestFksStreamsSQLFileContentsOnStdin(t *testing.T) { + installation := testInstallation(t) + sqlPath := filepath.Join(filepath.Dir(installation.Path), "query.sql") + if err := os.WriteFile(sqlPath, []byte("select 1;\n"), 0o600); err != nil { + t.Fatal(err) + } + runner := &fakeRunner{run: func(args []string, stdin string) (compose.Result, error) { + switch { + case contains(args, "config", "--format", "json"): + return compose.Result{Stdout: `{"services":{"core":{"image":"thothii-core:local"}}}`}, nil + case reflect.DeepEqual(args, []string{"image", "inspect", "--format", "{{.Id}}", "thothii-core:local"}): + return compose.Result{Stdout: "sha256:" + strings.Repeat("a", 64)}, nil + case contains(args, "workspace-maintenance", "schema-suggest-fks"): + var envelope map[string]any + if err := json.Unmarshal([]byte(stdin), &envelope); err != nil { + t.Fatalf("stdin JSON = %q, err=%v", stdin, err) + } + sqlFiles, ok := envelope["sqlFiles"].([]any) + if !ok || len(sqlFiles) != 1 { + t.Fatalf("sqlFiles = %#v", envelope["sqlFiles"]) + } + file, ok := sqlFiles[0].(map[string]any) + if !ok || file["contents"] != "select 1;\n" { + t.Fatalf("sql file envelope = %#v", sqlFiles[0]) + } + return compose.Result{Stdout: successResult("schema-suggest-fks")}, nil + default: + t.Fatalf("unexpected Docker invocation: %#v", args) + return compose.Result{}, nil + } + }} + + _, err := Execute(context.Background(), installation, runner, SuggestFksRequest{baseRequest: baseRequest{Workspace: "abc"}, FromSQL: []string{sqlPath}}) + if err != nil { + t.Fatalf("Execute() error = %v", err) + } +} + +func TestExecuteSuggestFksRejectsTotalSQLIngressOverSixteenMiB(t *testing.T) { + installation := testInstallation(t) + paths := make([]string, 0, 17) + for index := 0; index < 17; index++ { + path := filepath.Join(filepath.Dir(installation.Path), fmt.Sprintf("query-%02d.sql", index)) + if err := os.WriteFile(path, bytes.Repeat([]byte("x"), 1<<20), 0o600); err != nil { + t.Fatal(err) + } + paths = append(paths, path) + } + runner := &fakeRunner{run: func(args []string, stdin string) (compose.Result, error) { + t.Fatalf("Docker should not run when total SQL ingress exceeds the bound: %#v", args) + return compose.Result{}, nil + }} + + _, err := Execute(context.Background(), installation, runner, SuggestFksRequest{baseRequest: baseRequest{Workspace: "abc"}, FromSQL: paths}) + if err == nil || !strings.Contains(err.Error(), "total") { + t.Fatalf("Execute() error = %v, want total-size failure", err) + } +} + +func TestExecuteSchemaCheckStreamsAnnotationContentOnStdin(t *testing.T) { + installation := testInstallation(t) + annotationsPath := filepath.Join(filepath.Dir(installation.Path), "annotations.yaml") + if err := os.WriteFile(annotationsPath, []byte("reviewed: []\n"), 0o600); err != nil { + t.Fatal(err) + } + runner := &fakeRunner{run: func(args []string, stdin string) (compose.Result, error) { + switch { + case contains(args, "config", "--format", "json"): + return compose.Result{Stdout: `{"services":{"core":{"image":"thothii-core:local"}}}`}, nil + case reflect.DeepEqual(args, []string{"image", "inspect", "--format", "{{.Id}}", "thothii-core:local"}): + return compose.Result{Stdout: "sha256:" + strings.Repeat("b", 64)}, nil + case contains(args, "workspace-maintenance", "schema-check"): + var envelope map[string]any + if err := json.Unmarshal([]byte(stdin), &envelope); err != nil { + t.Fatalf("stdin JSON = %q, err=%v", stdin, err) + } + if envelope["annotations"] != "reviewed: []\n" || envelope["reviewedCandidates"] != "sha256:"+strings.Repeat("c", 64) { + t.Fatalf("annotation envelope = %#v", envelope) + } + if _, exists := envelope["annotationsPath"]; exists { + t.Fatalf("annotation path leaked into stdin: %#v", envelope) + } + return compose.Result{Stdout: successResult("schema-check")}, nil + default: + t.Fatalf("unexpected Docker invocation: %#v", args) + return compose.Result{}, nil + } + }} + + _, err := Execute(context.Background(), installation, runner, CheckSchemaRequest{baseRequest: baseRequest{Workspace: "abc"}, Annotations: annotationsPath, ReviewedCandidates: "sha256:" + strings.Repeat("c", 64)}) + if err != nil { + t.Fatalf("Execute() error = %v", err) + } +} + +func TestExecuteSuggestFksWritesTheReturnedCandidateArtifact(t *testing.T) { + installation := testInstallation(t) + outputPath := filepath.Join(filepath.Dir(installation.Path), "candidates.yaml") + runner := &fakeRunner{run: func(args []string, stdin string) (compose.Result, error) { + switch { + case contains(args, "config", "--format", "json"): + return compose.Result{Stdout: `{"services":{"core":{"image":"thothii-core:local"}}}`}, nil + case reflect.DeepEqual(args, []string{"image", "inspect", "--format", "{{.Id}}", "thothii-core:local"}): + return compose.Result{Stdout: "sha256:" + strings.Repeat("d", 64)}, nil + case contains(args, "workspace-maintenance", "schema-suggest-fks"): + return compose.Result{Stdout: `{"schemaVersion":1,"status":"blocked","code":"manual_review_required","workspaceId":"abc","workspaceRevision":"1234567890abcdef1234567890abcdef12345678","descriptorBlob":"sha256:` + strings.Repeat("e", 64) + `","operation":"schema-suggest-fks","completedStages":[],"suggestedFksYaml":"reviewed: []\n"}`}, nil + default: + t.Fatalf("unexpected Docker invocation: %#v", args) + return compose.Result{}, nil + } + }} + + _, err := Execute(context.Background(), installation, runner, SuggestFksRequest{baseRequest: baseRequest{Workspace: "abc"}, Output: outputPath}) + if err != nil { + t.Fatalf("Execute() error = %v", err) + } + contents, err := os.ReadFile(outputPath) + if err != nil { + t.Fatalf("output artifact missing: %v", err) + } + if string(contents) != "reviewed: []\n" { + t.Fatalf("output contents = %q", contents) + } +} + +func testInstallation(t *testing.T) config.Installation { + t.Helper() + temporaryRoot, err := filepath.EvalSymlinks(os.TempDir()) + if err != nil { + t.Fatal(err) + } + root, err := os.MkdirTemp(temporaryRoot, "thothctl-workspaceops-") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = os.RemoveAll(root) }) + project := filepath.Join(root, "project") + if err := os.MkdirAll(filepath.Join(project, "deploy"), 0o700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(project, "compose.yaml"), []byte("services: {}\n"), 0o600); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(project, "deploy", "compose.local.yaml"), []byte("services: {}\n"), 0o600); err != nil { + t.Fatal(err) + } + envFile := filepath.Join(root, "installation.env") + if err := os.WriteFile(envFile, []byte("SAFE_VALUE=1\n"), 0o600); err != nil { + t.Fatal(err) + } + return config.Installation{Path: filepath.Join(root, "thothii-installation.yaml"), Profile: "local", ProjectDirectory: project, EnvFile: envFile} +} + +func contains(values []string, sequence ...string) bool { + for start := range values { + if start+len(sequence) <= len(values) && reflect.DeepEqual(values[start:start+len(sequence)], sequence) { + return true + } + } + return false +} + +func successResult(operation string) string { + return `{"schemaVersion":1,"status":"succeeded","code":"ok","workspaceId":"abc","workspaceRevision":"1234567890abcdef1234567890abcdef12345678","descriptorBlob":"sha256:` + strings.Repeat("f", 64) + `","operation":"` + operation + `","completedStages":[]}` +}