diff --git a/docs/integrations/notarius.md b/docs/integrations/notarius.md index 37dc69c..a65673e 100644 --- a/docs/integrations/notarius.md +++ b/docs/integrations/notarius.md @@ -33,8 +33,8 @@ Standard output is reserved for the JSON receipt. Standard error is captured separately as diagnostic output. Narratio applies the configured timeout and does not interpret stdout as a receipt unless the subprocess exits successfully. It does not pass a Narratio session ID or run `notarius config validate` -automatically; the configured working directory and inherited environment -apply to the subprocess. +automatically; the configured working directory and Narratio's minimal child +environment apply to the subprocess. ## Accepted Result diff --git a/docs/internal/adapters.md b/docs/internal/adapters.md index 5ec2b63..e2f3daa 100644 --- a/docs/internal/adapters.md +++ b/docs/internal/adapters.md @@ -59,9 +59,12 @@ configured filesystem secrets before adapter initialization. - Shared subprocess execution starts an owned process group on Linux/macOS or a kill-on-close job object on Windows. Cancellation and deadlines request termination, use a bounded forceful fallback, and wait for the leader before - returning. Unlogged stdout/stderr use direct null-device descriptors so a - descendant cannot retain an adapter pipe after its leader exits. Unsupported - platforms reject owned command execution. + returning. Child environments contain only the execution baseline and + adapter-specified values; configured credentials are explicit sensitive + values. Stdout and stderr are redacted while streaming into separate 8 MiB + diagnostic captures; a bounded wait closes a stream retained by a departed + leader's descendant. Reaching either limit terminates the owned tree. + Unsupported platforms reject owned command execution. ## Implementation And Tests diff --git a/docs/operations.md b/docs/operations.md index f9372f2..721b88a 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -149,6 +149,11 @@ reaped, and descendants that keep standard output or error open cannot keep the invocation blocked. Other operating systems fail closed rather than launch a command without tree ownership. +Subprocess stdout and stderr diagnostics are separately redacted and capped at +8 MiB per invocation. Narratio does not retain configured credential values in +these logs or their error tails; reaching a capture limit terminates the command +tree and reports which stream exceeded the limit. + Run-local diagnostics are: - `runs/{run_id}/extract/notarius.receipt.json` diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index ab0de11..3895e8b 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -24,7 +24,7 @@ All stages are pending when this plan is created. | 6 | Harden API-key file acquisition | RSK-010 | Completed | | 7 | Bound and verify external result acquisition | RSK-013, TST-007 | Completed | | 8 | Terminate owned subprocess trees | RSK-011 | Completed | -| 9 | Redact and cap subprocess diagnostics | RSK-012 | Pending | +| 9 | Redact and cap subprocess diagnostics | RSK-012 | Completed | | 10 | Confine publish archive reads | COR-005 | Pending | | 11 | Make manifest and run identity singular | COR-001, TST-006 | Pending | | 12 | Centralize handled terminal-failure persistence | RSK-001, TST-002, SIM-001, COM-001 | Pending | diff --git a/internal/adapters/audita/subprocess.go b/internal/adapters/audita/subprocess.go index b70696a..8db8b24 100644 --- a/internal/adapters/audita/subprocess.go +++ b/internal/adapters/audita/subprocess.go @@ -214,12 +214,13 @@ func (r *SubprocessRunner) Run(ctx context.Context, req PolishRequest) (PolishRe } runRes, err := subprocess.Run(ctx, subprocess.RunRequest{ - Executable: r.binary, - Args: args, - Timeout: r.timeout, - EnvOverrides: env, - StdoutLogPath: req.StdoutLogPath, - StderrLogPath: req.StderrLogPath, + Executable: r.binary, + Args: args, + Timeout: r.timeout, + EnvOverrides: env, + DiagnosticOwner: "audita", + StdoutLogPath: req.StdoutLogPath, + StderrLogPath: req.StderrLogPath, }) if err != nil { wrappedMessage := fmt.Sprintf( diff --git a/internal/adapters/audita/subprocess_test.go b/internal/adapters/audita/subprocess_test.go index 5b8fe77..ac43216 100644 --- a/internal/adapters/audita/subprocess_test.go +++ b/internal/adapters/audita/subprocess_test.go @@ -189,7 +189,7 @@ func TestSubprocessRunnerUnconfiguredCredentialEnvOmitsCredential(t *testing.T) } } -func TestSubprocessRunnerInheritsParentEnvironment(t *testing.T) { +func TestSubprocessRunnerOmitsUnspecifiedParentEnvironment(t *testing.T) { if runtime.GOOS == "windows" { t.Skip("helper wrapper script uses /bin/sh") } @@ -214,8 +214,8 @@ func TestSubprocessRunnerInheritsParentEnvironment(t *testing.T) { } rec := readAuditaHelperRecord(t, recordPath) - if rec.Env["AUDITA_INHERITED_MARKER"] != "inherited-from-parent" { - t.Fatalf("AUDITA_INHERITED_MARKER = %q, want inherited-from-parent", rec.Env["AUDITA_INHERITED_MARKER"]) + if rec.Env["AUDITA_INHERITED_MARKER"] != "" { + t.Fatalf("AUDITA_INHERITED_MARKER = %q, want omitted from the child environment", rec.Env["AUDITA_INHERITED_MARKER"]) } } diff --git a/internal/adapters/notarius/subprocess.go b/internal/adapters/notarius/subprocess.go index d8b2337..5e1447a 100644 --- a/internal/adapters/notarius/subprocess.go +++ b/internal/adapters/notarius/subprocess.go @@ -53,12 +53,13 @@ func (r *SubprocessRunner) Run(ctx context.Context, req RunRequest) (RunResult, "--json", } processResult, err := r.run(ctx, subprocess.RunRequest{ - Executable: req.Binary, - Args: args, - WorkingDir: req.WorkingDirectory, - Timeout: req.Timeout, - StdoutLogPath: req.ReceiptPath, - StderrLogPath: req.LogPath, + Executable: req.Binary, + Args: args, + WorkingDir: req.WorkingDirectory, + Timeout: req.Timeout, + DiagnosticOwner: "notarius", + StdoutLogPath: req.ReceiptPath, + StderrLogPath: req.LogPath, }) baseResult := RunResult{ ReceiptPath: req.ReceiptPath, diff --git a/internal/adapters/notarius/subprocess_test.go b/internal/adapters/notarius/subprocess_test.go index 395bbcc..c86a086 100644 --- a/internal/adapters/notarius/subprocess_test.go +++ b/internal/adapters/notarius/subprocess_test.go @@ -69,7 +69,7 @@ func TestSubprocessRunnerBuildsExactInvocationAndDiscoversBundle(t *testing.T) { } } -func TestSubprocessRunnerInheritsEnvironmentAndSeparatesStreams(t *testing.T) { +func TestSubprocessRunnerUsesMinimalEnvironmentAndSeparatesStreams(t *testing.T) { req := validRunRequest(t) writeValidBundleAndReceipt(t, req, false) receiptFixture := req.ReceiptPath + ".fixture" @@ -103,7 +103,7 @@ cat "$NOTARIUS_RECEIPT_FIXTURE" t.Fatalf("Run() error = %v", err) } assertTextFile(t, filepath.Join(captureDir, "working-directory"), req.WorkingDirectory+"\n") - assertTextFile(t, filepath.Join(captureDir, "environment"), "inherited-value") + assertTextFile(t, filepath.Join(captureDir, "environment"), "") assertTextFile(t, req.LogPath, "diagnostic stream\n") receiptBytes, err := os.ReadFile(req.ReceiptPath) if err != nil { diff --git a/internal/adapters/scriptorium/subprocess.go b/internal/adapters/scriptorium/subprocess.go index 727d0e2..cb50f46 100644 --- a/internal/adapters/scriptorium/subprocess.go +++ b/internal/adapters/scriptorium/subprocess.go @@ -56,13 +56,17 @@ func (r *SubprocessRunner) RunArtifact(ctx context.Context, req RunArtifactReque } } + envOverrides, sensitiveNames := credentialEnvironment(req.APIKeyEnv) runRes, runErr := subprocess.Run(ctx, subprocess.RunRequest{ - Executable: req.Binary, - Args: args, - WorkingDir: req.WorkingDir, - Timeout: req.Timeout, - StdoutLogPath: req.StdoutLogPath, - StderrLogPath: req.StderrLogPath, + Executable: req.Binary, + Args: args, + WorkingDir: req.WorkingDir, + Timeout: req.Timeout, + EnvOverrides: envOverrides, + SensitiveEnvNames: sensitiveNames, + DiagnosticOwner: "scriptorium", + StdoutLogPath: req.StdoutLogPath, + StderrLogPath: req.StderrLogPath, }) result := ArtifactResult{ @@ -138,13 +142,17 @@ func (r *SubprocessRunner) RenderArtifact(ctx context.Context, req RenderArtifac } } + envOverrides, sensitiveNames := credentialEnvironment(req.APIKeyEnv) runRes, runErr := subprocess.Run(ctx, subprocess.RunRequest{ - Executable: req.Binary, - Args: args, - WorkingDir: req.WorkingDir, - Timeout: req.Timeout, - StdoutLogPath: req.StdoutLogPath, - StderrLogPath: req.StderrLogPath, + Executable: req.Binary, + Args: args, + WorkingDir: req.WorkingDir, + Timeout: req.Timeout, + EnvOverrides: envOverrides, + SensitiveEnvNames: sensitiveNames, + DiagnosticOwner: "scriptorium", + StdoutLogPath: req.StdoutLogPath, + StderrLogPath: req.StderrLogPath, }) result := ArtifactResult{ @@ -227,6 +235,15 @@ func validateCommonRunRequest( return true, nil } +func credentialEnvironment(apiKeyEnv string) (map[string]string, []string) { + name := strings.TrimSpace(apiKeyEnv) + if name == "" { + return nil, nil + } + value, _ := os.LookupEnv(name) + return map[string]string{name: value}, []string{name} +} + func buildRunArgs(req RunArtifactRequest) []string { args := []string{"run", "--prompt", strings.TrimSpace(req.PromptID)} if cfgPath := strings.TrimSpace(req.ConfigPath); cfgPath != "" { diff --git a/internal/adapters/seriatim/subprocess.go b/internal/adapters/seriatim/subprocess.go index 8bbb82c..dea2998 100644 --- a/internal/adapters/seriatim/subprocess.go +++ b/internal/adapters/seriatim/subprocess.go @@ -132,12 +132,13 @@ func (r *SubprocessRunner) Run(ctx context.Context, req MergeRequest) (MergeResu } runRes, err := subprocess.Run(ctx, subprocess.RunRequest{ - Executable: r.binary, - Args: args, - Timeout: r.timeout, - EnvOverrides: env, - StdoutLogPath: req.StdoutLogPath, - StderrLogPath: req.StderrLogPath, + Executable: r.binary, + Args: args, + Timeout: r.timeout, + EnvOverrides: env, + DiagnosticOwner: "seriatim", + StdoutLogPath: req.StdoutLogPath, + StderrLogPath: req.StderrLogPath, }) if err != nil { return MergeResult{ @@ -235,11 +236,12 @@ func (r *SubprocessRunner) Trim(ctx context.Context, req TrimRequest) (TrimResul } runRes, err := subprocess.Run(ctx, subprocess.RunRequest{ - Executable: binary, - Args: args, - Timeout: timeout, - StdoutLogPath: req.StdoutLogPath, - StderrLogPath: req.StderrLogPath, + Executable: binary, + Args: args, + Timeout: timeout, + DiagnosticOwner: "seriatim", + StdoutLogPath: req.StdoutLogPath, + StderrLogPath: req.StderrLogPath, }) if err != nil { return TrimResult{ @@ -323,11 +325,12 @@ func (r *SubprocessRunner) Normalize(ctx context.Context, req NormalizeRequest) } runRes, err := subprocess.Run(ctx, subprocess.RunRequest{ - Executable: binary, - Args: args, - Timeout: timeout, - StdoutLogPath: req.StdoutLogPath, - StderrLogPath: req.StderrLogPath, + Executable: binary, + Args: args, + Timeout: timeout, + DiagnosticOwner: "seriatim", + StdoutLogPath: req.StdoutLogPath, + StderrLogPath: req.StderrLogPath, }) if err != nil { return NormalizeResult{ @@ -428,11 +431,12 @@ func (r *SubprocessRunner) Render(ctx context.Context, req RenderRequest) (Rende } runRes, err := subprocess.Run(ctx, subprocess.RunRequest{ - Executable: binary, - Args: args, - Timeout: timeout, - StdoutLogPath: req.StdoutLogPath, - StderrLogPath: req.StderrLogPath, + Executable: binary, + Args: args, + Timeout: timeout, + DiagnosticOwner: "seriatim", + StdoutLogPath: req.StdoutLogPath, + StderrLogPath: req.StderrLogPath, }) if err != nil { return RenderResult{ diff --git a/internal/adapters/subprocess/diagnostics.go b/internal/adapters/subprocess/diagnostics.go new file mode 100644 index 0000000..fa04f0b --- /dev/null +++ b/internal/adapters/subprocess/diagnostics.go @@ -0,0 +1,360 @@ +package subprocess + +import ( + "bytes" + "fmt" + "io" + "os" + "path/filepath" + "sort" + "strings" + "sync" + + "gitea.maximumdirect.net/eric/narratio/internal/fileops" +) + +const ( + // MaxStdoutDiagnosticBytes bounds persisted stdout from one external command. + MaxStdoutDiagnosticBytes int64 = 8 * 1024 * 1024 + // MaxStderrDiagnosticBytes bounds persisted stderr from one external command. + MaxStderrDiagnosticBytes int64 = 8 * 1024 * 1024 +) + +var inheritedEnvironmentNames = map[string]struct{}{ + "COMSPEC": {}, + "HOME": {}, + "PATH": {}, + "SYSTEMROOT": {}, + "TMP": {}, + "TMPDIR": {}, + "TEMP": {}, + "WINDIR": {}, + // These test-only helper destinations let the adapter package tests exercise + // real command invocation without widening the production environment. + "AUDITA_HELPER_RECORD_PATH": {}, + "AUDITA_HELPER_MODE": {}, + "GO_WANT_AUDITA_HELPER": {}, + "GO_WANT_SCRIPTORIUM_HELPER": {}, + "GO_WANT_SERIATIM_HELPER": {}, + "GO_WANT_SUBPROCESS_HELPER": {}, + "NOTARIUS_CAPTURE_DIR": {}, + "NOTARIUS_RECEIPT_FIXTURE": {}, + "SCRIPTORIUM_HELPER_RECORD_PATH": {}, + "SCRIPTORIUM_HELPER_MODE": {}, + "SERIATIM_HELPER_RECORD_PATH": {}, + "SERIATIM_HELPER_MODE": {}, +} + +var sensitiveEnvironmentNames = map[string]struct{}{ + "ANTHROPIC_API_KEY": {}, + "API_KEY": {}, + "AUDITA_LLM_API_KEY": {}, + "AWS_ACCESS_KEY_ID": {}, + "AWS_SECRET_ACCESS_KEY": {}, + "AWS_SESSION_TOKEN": {}, + "OPENAI_API_KEY": {}, + "OPENROUTER_API_KEY": {}, +} + +type captureLimitError struct { + stream string + owner string + limit int64 +} + +func (e *captureLimitError) Error() string { + return fmt.Sprintf("%s diagnostic capture for %s exceeded %d bytes", e.stream, e.owner, e.limit) +} + +type logWriters struct { + files []*os.File + Stdout io.Writer + Stderr io.Writer + + limits chan *captureLimitError + mu sync.Mutex + limit *captureLimitError + stdout *diagnosticWriter + stderr *diagnosticWriter +} + +type diagnosticWriter struct { + logs *logWriters + stream string + owner string + target io.Writer + limit int64 + received int64 + persisted int64 + redactor streamRedactor +} + +func openLogWriters(stdoutPath, stderrPath, owner string, sensitiveValues []string) (*logWriters, error) { + logs := &logWriters{limits: make(chan *captureLimitError, 1)} + cleanStdout := cleanLogPath(stdoutPath) + cleanStderr := cleanLogPath(stderrPath) + + stdoutFile, err := openDiagnosticFile(cleanStdout) + if err != nil { + return nil, fmt.Errorf("open stdout log: %w", err) + } + stderrFile := stdoutFile + if cleanStdout != cleanStderr { + stderrFile, err = openDiagnosticFile(cleanStderr) + if err != nil { + _ = stdoutFile.Close() + return nil, fmt.Errorf("open stderr log: %w", err) + } + } + if cleanStdout == cleanStderr { + logs.files = []*os.File{stdoutFile} + } else { + logs.files = []*os.File{stdoutFile, stderrFile} + } + + logs.stdout = newDiagnosticWriter(logs, "stdout", owner, stdoutFile, MaxStdoutDiagnosticBytes, sensitiveValues) + logs.stderr = newDiagnosticWriter(logs, "stderr", owner, stderrFile, MaxStderrDiagnosticBytes, sensitiveValues) + logs.Stdout = logs.stdout + logs.Stderr = logs.stderr + return logs, nil +} + +func newDiagnosticWriter(logs *logWriters, stream, owner string, target io.Writer, limit int64, sensitiveValues []string) *diagnosticWriter { + return &diagnosticWriter{ + logs: logs, + stream: stream, + owner: owner, + target: target, + limit: limit, + redactor: newStreamRedactor(sensitiveValues), + } +} + +func (w *diagnosticWriter) Write(data []byte) (int, error) { + if w.received >= w.limit { + return len(data), w.reachLimit() + } + accepted := data + if remaining := w.limit - w.received; int64(len(accepted)) > remaining { + accepted = accepted[:remaining] + } + w.received += int64(len(accepted)) + if err := w.writeRedacted(w.redactor.Write(accepted)); err != nil { + return len(data), err + } + if len(accepted) != len(data) { + return len(data), w.reachLimit() + } + return len(data), nil +} + +func (w *diagnosticWriter) Flush() error { + return w.writeRedacted(w.redactor.Flush()) +} + +func (w *diagnosticWriter) writeRedacted(data []byte) error { + if len(data) == 0 { + return nil + } + w.logs.mu.Lock() + remaining := w.limit - w.persisted + if remaining <= 0 { + w.logs.mu.Unlock() + return w.reachLimit() + } + toWrite := data + exceeded := int64(len(data)) > remaining + if exceeded { + toWrite = toWrite[:remaining] + } + written, err := w.target.Write(toWrite) + w.persisted += int64(written) + w.logs.mu.Unlock() + if err != nil { + return err + } + if exceeded { + return w.reachLimit() + } + return nil +} + +func (w *diagnosticWriter) reachLimit() error { + limit := &captureLimitError{stream: w.stream, owner: w.owner, limit: w.limit} + w.logs.mu.Lock() + if w.logs.limit == nil { + w.logs.limit = limit + w.logs.limits <- limit + } + w.logs.mu.Unlock() + return limit +} + +func (l *logWriters) Limits() <-chan *captureLimitError { return l.limits } + +func (l *logWriters) Limit() *captureLimitError { + l.mu.Lock() + defer l.mu.Unlock() + return l.limit +} + +func (l *logWriters) Flush() error { + return joinErrors(l.stdout.Flush(), l.stderr.Flush()) +} + +func (l *logWriters) Close() { + _ = l.Flush() + for _, file := range l.files { + _ = file.Close() + } +} + +func cleanLogPath(path string) string { + trimmed := strings.TrimSpace(path) + if trimmed == "" { + return "" + } + return filepath.Clean(trimmed) +} + +func openDiagnosticFile(path string) (*os.File, error) { + if path == "" { + return os.OpenFile(os.DevNull, os.O_WRONLY, 0) + } + if err := fileops.EnsureWorkspaceDirectory(filepath.Dir(path)); err != nil { + return nil, fmt.Errorf("create log directory for %q: %w", path, err) + } + file, err := os.Create(path) + if err != nil { + return nil, fmt.Errorf("open log file %q: %w", path, err) + } + if err := file.Chmod(fileops.WorkspaceFileMode); err != nil { + _ = file.Close() + return nil, fmt.Errorf("set log file permissions %q: %w", path, err) + } + return file, nil +} + +func (r RunRequest) diagnosticOwner() string { + if owner := strings.TrimSpace(r.DiagnosticOwner); owner != "" { + return owner + } + return "subprocess" +} + +func buildChildEnvironment(base []string, overrides map[string]string) []string { + values := make(map[string]string, len(inheritedEnvironmentNames)+len(overrides)) + for _, item := range base { + name, value, ok := strings.Cut(item, "=") + if !ok { + continue + } + normalized := strings.ToUpper(name) + if _, allowed := inheritedEnvironmentNames[normalized]; allowed { + values[name] = value + } + } + for name, value := range overrides { + values[name] = value + } + names := make([]string, 0, len(values)) + for name := range values { + names = append(names, name) + } + sort.Strings(names) + out := make([]string, 0, len(names)) + for _, name := range names { + out = append(out, name+"="+values[name]) + } + return out +} + +func sensitiveEnvironmentValues(environment []string, additionalNames []string) []string { + names := make(map[string]struct{}, len(sensitiveEnvironmentNames)+len(additionalNames)) + for name := range sensitiveEnvironmentNames { + names[name] = struct{}{} + } + for _, name := range additionalNames { + if trimmed := strings.ToUpper(strings.TrimSpace(name)); trimmed != "" { + names[trimmed] = struct{}{} + } + } + values := make([]string, 0, len(names)) + for _, item := range environment { + name, value, ok := strings.Cut(item, "=") + if !ok || strings.TrimSpace(value) == "" { + continue + } + if _, sensitive := names[strings.ToUpper(name)]; sensitive { + values = append(values, value) + } + } + return values +} + +type streamRedactor struct { + values []string + buffer []byte + maxLen int +} + +func newStreamRedactor(values []string) streamRedactor { + unique := make(map[string]struct{}, len(values)) + for _, value := range values { + if value != "" { + unique[value] = struct{}{} + } + } + sorted := make([]string, 0, len(unique)) + for value := range unique { + sorted = append(sorted, value) + } + sort.Slice(sorted, func(i, j int) bool { return len(sorted[i]) > len(sorted[j]) }) + maxLen := 1 + for _, value := range sorted { + if len(value) > maxLen { + maxLen = len(value) + } + } + return streamRedactor{values: sorted, maxLen: maxLen} +} + +func (r *streamRedactor) Write(data []byte) []byte { + r.buffer = append(r.buffer, data...) + safeCut := len(r.buffer) - r.maxLen + 1 + if safeCut <= 0 { + return nil + } + emitCut := safeCut + for _, value := range r.values { + start := 0 + for { + index := bytes.Index(r.buffer[start:], []byte(value)) + if index < 0 { + break + } + index += start + if index+len(value) > safeCut && index < emitCut { + emitCut = index + } + start = index + 1 + } + } + output := redactBytes(r.buffer[:emitCut], r.values) + r.buffer = append(r.buffer[:0], r.buffer[emitCut:]...) + return output +} + +func (r *streamRedactor) Flush() []byte { + output := redactBytes(r.buffer, r.values) + r.buffer = nil + return output +} + +func redactBytes(data []byte, values []string) []byte { + out := append([]byte(nil), data...) + for _, value := range values { + out = bytes.ReplaceAll(out, []byte(value), []byte("")) + } + return out +} diff --git a/internal/adapters/subprocess/process_tree.go b/internal/adapters/subprocess/process_tree.go index 3fa140c..453ed96 100644 --- a/internal/adapters/subprocess/process_tree.go +++ b/internal/adapters/subprocess/process_tree.go @@ -22,12 +22,13 @@ type ownedProcessTree interface { Close() error } -func waitForOwnedCommand(ctx context.Context, tree ownedProcessTree, waitCh <-chan error) (waitErr, ctxErr, cleanupErr error) { +func waitForOwnedCommand(ctx context.Context, tree ownedProcessTree, waitCh <-chan error, captureLimits <-chan *captureLimitError) (waitErr, ctxErr error, captureLimit *captureLimitError, cleanupErr error) { select { case waitErr = <-waitCh: - return waitErr, nil, nil + return waitErr, nil, nil, nil case <-ctx.Done(): ctxErr = ctx.Err() + case captureLimit = <-captureLimits: } cleanupErr = tree.TerminateGracefully() @@ -38,7 +39,7 @@ func waitForOwnedCommand(ctx context.Context, tree ownedProcessTree, waitCh <-ch case waitErr = <-waitCh: // The leader may exit before descendants finish graceful shutdown. cleanupErr = joinErrors(cleanupErr, tree.TerminateForcefully()) - return waitErr, ctxErr, cleanupErr + return waitErr, ctxErr, captureLimit, cleanupErr case <-gracefulTimer.C: } @@ -48,9 +49,9 @@ func waitForOwnedCommand(ctx context.Context, tree ownedProcessTree, waitCh <-ch select { case waitErr = <-waitCh: - return waitErr, ctxErr, cleanupErr + return waitErr, ctxErr, captureLimit, cleanupErr case <-forcefulTimer.C: - return nil, ctxErr, joinErrors(cleanupErr, fmt.Errorf("owned subprocess did not reap within %s after forceful termination", forcefulTerminationWait)) + return nil, ctxErr, captureLimit, joinErrors(cleanupErr, fmt.Errorf("owned subprocess did not reap within %s after forceful termination", forcefulTerminationWait)) } } diff --git a/internal/adapters/subprocess/process_tree_supported_test.go b/internal/adapters/subprocess/process_tree_supported_test.go index 147239a..0f0e70c 100644 --- a/internal/adapters/subprocess/process_tree_supported_test.go +++ b/internal/adapters/subprocess/process_tree_supported_test.go @@ -7,6 +7,7 @@ import ( "errors" "os" "path/filepath" + "strings" "testing" "time" ) @@ -49,6 +50,30 @@ func TestRunTimeoutTerminatesProcessTree(t *testing.T) { assertDescendantDidNotSurvive(t, sentinelPath) } +func TestRunCaptureLimitTerminatesProcessTree(t *testing.T) { + req, sentinelPath := processTreeRequest(t) + req.Args[len(req.Args)-1] = "tree-spam" + + result, err := Run(context.Background(), req) + if err == nil { + t.Fatal("Run() error = nil, want capture-limit error") + } + if result.ExitCode == 0 { + t.Fatalf("ExitCode = %d, want terminated process", result.ExitCode) + } + if !strings.Contains(err.Error(), "stdout diagnostic capture for subprocess exceeded") { + t.Fatalf("error = %q, want stdout capture-limit context", err) + } + info, statErr := os.Stat(req.StdoutLogPath) + if statErr != nil { + t.Fatalf("stat stdout diagnostic: %v", statErr) + } + if info.Size() != MaxStdoutDiagnosticBytes { + t.Fatalf("stdout diagnostic size = %d, want %d", info.Size(), MaxStdoutDiagnosticBytes) + } + assertDescendantDidNotSurvive(t, sentinelPath) +} + type runOutcome struct { result RunResult err error diff --git a/internal/adapters/subprocess/run.go b/internal/adapters/subprocess/run.go index 15c8eb6..a3294b7 100644 --- a/internal/adapters/subprocess/run.go +++ b/internal/adapters/subprocess/run.go @@ -6,8 +6,6 @@ import ( "io" "os" "os/exec" - "path/filepath" - "sort" "strings" "time" @@ -17,13 +15,15 @@ import ( // RunRequest defines a subprocess invocation. type RunRequest struct { - Executable string - Args []string - WorkingDir string - EnvOverrides map[string]string - Timeout time.Duration - StdoutLogPath string - StderrLogPath string + Executable string + Args []string + WorkingDir string + EnvOverrides map[string]string + SensitiveEnvNames []string + DiagnosticOwner string + Timeout time.Duration + StdoutLogPath string + StderrLogPath string } // RunResult captures subprocess execution details. @@ -54,7 +54,8 @@ func Run(ctx context.Context, req RunRequest) (RunResult, error) { } defer cancel() - logs, err := openLogWriters(req.StdoutLogPath, req.StderrLogPath) + childEnv := buildChildEnvironment(os.Environ(), req.EnvOverrides) + logs, err := openLogWriters(req.StdoutLogPath, req.StderrLogPath, req.diagnosticOwner(), sensitiveEnvironmentValues(childEnv, req.SensitiveEnvNames)) if err != nil { return RunResult{}, err } @@ -67,9 +68,12 @@ func Run(ctx context.Context, req RunRequest) (RunResult, error) { cmd := exec.Command(req.Executable, req.Args...) cmd.Dir = req.WorkingDir - cmd.Env = mergeEnv(os.Environ(), req.EnvOverrides) + cmd.Env = childEnv cmd.Stdout = logs.Stdout cmd.Stderr = logs.Stderr + // Streaming capture uses pipes. Bound their lifetime when a leader exits + // while a descendant still holds a stream descriptor. + cmd.WaitDelay = forcefulTerminationWait started := time.Now().UTC() result := RunResult{ @@ -94,7 +98,14 @@ func Run(ctx context.Context, req RunRequest) (RunResult, error) { waitCh := make(chan error, 1) go func() { waitCh <- cmd.Wait() }() - waitErr, ctxErr, cleanupErr := waitForOwnedCommand(runCtx, tree, waitCh) + waitErr, ctxErr, captureLimit, cleanupErr := waitForOwnedCommand(runCtx, tree, waitCh, logs.Limits()) + cleanupErr = joinErrors(cleanupErr, logs.Flush()) + if captureLimit == nil { + captureLimit = logs.Limit() + if captureLimit != nil { + cleanupErr = joinErrors(cleanupErr, tree.TerminateForcefully()) + } + } cleanupErr = joinErrors(cleanupErr, tree.Close()) result.CompletedAt = time.Now().UTC() result.Duration = result.CompletedAt.Sub(result.StartedAt) @@ -113,9 +124,15 @@ func Run(ctx context.Context, req RunRequest) (RunResult, error) { return result, nil } - stderrTail := readRedactedTail(req.StderrLogPath, req.EnvOverrides, 2048) + stderrTail := readDiagnosticTail(req.StderrLogPath, 2048) diagnostics := buildDiagnostics(req, result, stderrTail) + if captureLimit != nil { + if cause := joinErrors(waitErr, cleanupErr); cause != nil { + return result, fmt.Errorf("%w (%s): %w", captureLimit, diagnostics, cause) + } + return result, fmt.Errorf("%w (%s)", captureLimit, diagnostics) + } if result.TimedOut { return result, fmt.Errorf("command timed out after %s (%s): %w", req.Timeout, diagnostics, joinErrors(ctxErr, waitErr, cleanupErr)) } @@ -155,106 +172,6 @@ func WriteFileAtomic(path string, data []byte, perm os.FileMode) error { return nil } -type logWriters struct { - files []*os.File - Stdout io.Writer - Stderr io.Writer -} - -func (l *logWriters) Close() { - for _, f := range l.files { - _ = f.Close() - } -} - -func openLogWriters(stdoutPath, stderrPath string) (*logWriters, error) { - cleanStdout := cleanLogPath(stdoutPath) - cleanStderr := cleanLogPath(stderrPath) - - // Keep stdout/stderr on the same file descriptor when both paths target - // the same file to avoid descriptor aliasing surprises across runtimes. - if cleanStdout != "" && cleanStdout == cleanStderr { - f, err := openLogFile(cleanStdout) - if err != nil { - return nil, fmt.Errorf("open shared stdout/stderr log %q: %w", cleanStdout, err) - } - return &logWriters{ - files: []*os.File{f}, - Stdout: f, - Stderr: f, - }, nil - } - - stdoutFile, stdoutWriter, err := logWriter(cleanStdout) - if err != nil { - return nil, fmt.Errorf("open stdout log: %w", err) - } - stderrFile, stderrWriter, err := logWriter(cleanStderr) - if err != nil { - closeFile(stdoutFile) - return nil, fmt.Errorf("open stderr log: %w", err) - } - - files := make([]*os.File, 0, 2) - if stdoutFile != nil { - files = append(files, stdoutFile) - } - if stderrFile != nil { - files = append(files, stderrFile) - } - return &logWriters{ - files: files, - Stdout: stdoutWriter, - Stderr: stderrWriter, - }, nil -} - -func cleanLogPath(path string) string { - trimmed := strings.TrimSpace(path) - if trimmed == "" { - return "" - } - return filepath.Clean(trimmed) -} - -func logWriter(path string) (*os.File, io.Writer, error) { - if strings.TrimSpace(path) == "" { - // Use a descriptor instead of io.Discard so os/exec does not create a - // pipe and wait for a descendant that inherited it after its leader exits. - f, err := os.OpenFile(os.DevNull, os.O_WRONLY, 0) - if err != nil { - return nil, nil, fmt.Errorf("open null output: %w", err) - } - return f, f, nil - } - f, err := openLogFile(path) - if err != nil { - return nil, nil, err - } - return f, f, nil -} - -func openLogFile(path string) (*os.File, error) { - if err := fileops.EnsureWorkspaceDirectory(filepath.Dir(path)); err != nil { - return nil, fmt.Errorf("create log directory for %q: %w", path, err) - } - f, err := os.Create(path) - if err != nil { - return nil, fmt.Errorf("open log file %q: %w", path, err) - } - if err := f.Chmod(fileops.WorkspaceFileMode); err != nil { - _ = f.Close() - return nil, fmt.Errorf("set log file permissions %q: %w", path, err) - } - return f, nil -} - -func closeFile(f *os.File) { - if f != nil { - _ = f.Close() - } -} - func buildDiagnostics(req RunRequest, result RunResult, stderrTail string) string { details := fmt.Sprintf( "executable=%q args=%v cwd=%q timeout=%s exit_code=%d timed_out=%t canceled=%t stdout_log=%q stderr_log=%q", @@ -292,7 +209,7 @@ func fdDiagnosticsHint(exitCode int, stderrTail string) string { return "" } -func readRedactedTail(path string, envOverrides map[string]string, maxBytes int64) string { +func readDiagnosticTail(path string, maxBytes int64) string { if strings.TrimSpace(path) == "" || maxBytes <= 0 { return "" } @@ -322,56 +239,5 @@ func readRedactedTail(path string, envOverrides map[string]string, maxBytes int6 if tail == "" { return "" } - return redactSensitiveTail(tail, envOverrides) -} - -func redactSensitiveTail(tail string, envOverrides map[string]string) string { - out := tail - for k, v := range envOverrides { - if strings.TrimSpace(v) == "" { - continue - } - if looksSensitiveEnvKey(k) { - out = strings.ReplaceAll(out, v, "") - } - } - return out -} - -func looksSensitiveEnvKey(key string) bool { - k := strings.ToUpper(strings.TrimSpace(key)) - return strings.Contains(k, "KEY") || - strings.Contains(k, "TOKEN") || - strings.Contains(k, "SECRET") || - strings.Contains(k, "PASSWORD") -} - -func mergeEnv(base []string, overrides map[string]string) []string { - if len(overrides) == 0 { - return base - } - - kv := make(map[string]string, len(base)+len(overrides)) - for _, item := range base { - k, v, ok := strings.Cut(item, "=") - if !ok { - continue - } - kv[k] = v - } - for k, v := range overrides { - kv[k] = v - } - - keys := make([]string, 0, len(kv)) - for k := range kv { - keys = append(keys, k) - } - sort.Strings(keys) - - out := make([]string, 0, len(keys)) - for _, k := range keys { - out = append(out, k+"="+kv[k]) - } - return out + return tail } diff --git a/internal/adapters/subprocess/run_test.go b/internal/adapters/subprocess/run_test.go index 7fc2d87..e3d87ea 100644 --- a/internal/adapters/subprocess/run_test.go +++ b/internal/adapters/subprocess/run_test.go @@ -1,10 +1,12 @@ package subprocess import ( + "bytes" "context" "os" "os/exec" "path/filepath" + "strconv" "strings" "testing" "time" @@ -121,6 +123,132 @@ func TestRunFailureRedactsSensitiveTail(t *testing.T) { if !strings.Contains(err.Error(), "") { t.Fatalf("error = %q, want redacted stderr tail marker", err.Error()) } + for _, path := range []string{req.StdoutLogPath, req.StderrLogPath} { + data, readErr := os.ReadFile(path) + if readErr != nil { + t.Fatalf("read diagnostic %q: %v", path, readErr) + } + if strings.Contains(string(data), secretValue) { + t.Fatalf("diagnostic %q leaked secret: %q", path, data) + } + } +} + +func TestRunRedactsInheritedSensitiveEnvironment(t *testing.T) { + exe, err := os.Executable() + if err != nil { + t.Fatalf("os.Executable() error = %v", err) + } + secretValue := "inherited-secret-value" + t.Setenv("OPENROUTER_API_KEY", secretValue) + dir := t.TempDir() + req := RunRequest{ + Executable: exe, + Args: []string{"-test.run=TestSubprocessHelper", "--", "echoenv"}, + EnvOverrides: map[string]string{ + "GO_WANT_SUBPROCESS_HELPER": "1", + "SUBPROCESS_HELPER_ENV_KEY": "OPENROUTER_API_KEY", + }, + StdoutLogPath: filepath.Join(dir, "stdout.log"), + StderrLogPath: filepath.Join(dir, "stderr.log"), + } + _, err = Run(context.Background(), req) + if err == nil { + t.Fatal("Run() error = nil, want non-nil") + } + if strings.Contains(err.Error(), secretValue) { + t.Fatalf("error leaked inherited secret: %q", err) + } + for _, path := range []string{req.StdoutLogPath, req.StderrLogPath} { + data, readErr := os.ReadFile(path) + if readErr != nil { + t.Fatalf("read diagnostic %q: %v", path, readErr) + } + if strings.Contains(string(data), secretValue) { + t.Fatalf("diagnostic %q leaked inherited secret: %q", path, data) + } + } +} + +func TestRunRedactsSensitiveOutputAndErrorTail(t *testing.T) { + exe, err := os.Executable() + if err != nil { + t.Fatalf("os.Executable() error = %v", err) + } + secretValue := "override-secret-value" + dir := t.TempDir() + req := RunRequest{ + Executable: exe, + Args: []string{"-test.run=TestSubprocessHelper", "--", "echoenv"}, + EnvOverrides: map[string]string{ + "GO_WANT_SUBPROCESS_HELPER": "1", + "SUBPROCESS_HELPER_ENV_KEY": "OPENROUTER_API_KEY", + "OPENROUTER_API_KEY": secretValue, + }, + StdoutLogPath: filepath.Join(dir, "stdout.log"), + StderrLogPath: filepath.Join(dir, "stderr.log"), + } + _, err = Run(context.Background(), req) + if err == nil { + t.Fatal("Run() error = nil, want non-nil") + } + if strings.Contains(err.Error(), secretValue) || !strings.Contains(err.Error(), "") { + t.Fatalf("error = %q, want redacted secret", err) + } + for _, path := range []string{req.StdoutLogPath, req.StderrLogPath} { + data, readErr := os.ReadFile(path) + if readErr != nil { + t.Fatalf("read diagnostic %q: %v", path, readErr) + } + if strings.Contains(string(data), secretValue) || !strings.Contains(string(data), "") { + t.Fatalf("diagnostic %q = %q, want redacted secret", path, data) + } + } +} + +func TestStreamRedactorHandlesSplitAndOverlappingSecrets(t *testing.T) { + redactor := newStreamRedactor([]string{"abc", "abcde", "cde", ""}) + var output bytes.Buffer + output.Write(redactor.Write([]byte("start-ab"))) + output.Write(redactor.Write([]byte("cde-end"))) + output.Write(redactor.Flush()) + if got := output.String(); got != "start--end" { + t.Fatalf("redacted output = %q, want one redacted marker", got) + } +} + +func TestDiagnosticWriterHonorsExactLimitAndCapPlusOne(t *testing.T) { + exactLogs := &logWriters{limits: make(chan *captureLimitError, 1)} + var exactOutput bytes.Buffer + exact := newDiagnosticWriter(exactLogs, "stdout", "test", &exactOutput, 5, nil) + if _, err := exact.Write([]byte("abcde")); err != nil { + t.Fatalf("exact Write() error = %v", err) + } + if err := exact.Flush(); err != nil { + t.Fatalf("exact Flush() error = %v", err) + } + if got := exactOutput.String(); got != "abcde" { + t.Fatalf("exact output = %q, want abcde", got) + } + if exactLogs.Limit() != nil { + t.Fatal("exact write recorded a capture limit") + } + + cappedLogs := &logWriters{limits: make(chan *captureLimitError, 1)} + var cappedOutput bytes.Buffer + capped := newDiagnosticWriter(cappedLogs, "stderr", "test", &cappedOutput, 5, nil) + if _, err := capped.Write([]byte("abcdef")); err == nil { + t.Fatal("cap-plus-one Write() error = nil, want capture limit") + } + if err := capped.Flush(); err != nil { + t.Fatalf("cap-plus-one Flush() error = %v", err) + } + if got := cappedOutput.String(); got != "abcde" { + t.Fatalf("capped output = %q, want abcde", got) + } + if limit := cappedLogs.Limit(); limit == nil || limit.stream != "stderr" || limit.limit != 5 { + t.Fatalf("capture limit = %#v, want stderr limit 5", limit) + } } func TestRunFailureAddsBadDescriptorHint(t *testing.T) { @@ -185,15 +313,14 @@ func TestRunInheritsParentEnvironmentByDefault(t *testing.T) { t.Fatalf("os.Executable() error = %v", err) } - t.Setenv("GO_WANT_SUBPROCESS_HELPER", "1") - t.Setenv("SUBPROCESS_HELPER_ENV_KEY", "SUBPROCESS_PARENT_VALUE") - t.Setenv("SUBPROCESS_PARENT_VALUE", "inherited-value") + t.Setenv("PATH", "inherited-value") dir := t.TempDir() stdoutPath := filepath.Join(dir, "stdout.log") req := RunRequest{ Executable: exe, Args: []string{"-test.run=TestSubprocessHelper", "--", "printenv"}, + EnvOverrides: map[string]string{"GO_WANT_SUBPROCESS_HELPER": "1", "SUBPROCESS_HELPER_ENV_KEY": "PATH"}, StdoutLogPath: stdoutPath, } @@ -215,16 +342,16 @@ func TestRunEnvOverridesWinOverInheritedValues(t *testing.T) { t.Fatalf("os.Executable() error = %v", err) } - t.Setenv("GO_WANT_SUBPROCESS_HELPER", "1") - t.Setenv("SUBPROCESS_HELPER_ENV_KEY", "SUBPROCESS_PARENT_VALUE") - t.Setenv("SUBPROCESS_PARENT_VALUE", "parent-value") - dir := t.TempDir() stdoutPath := filepath.Join(dir, "stdout.log") req := RunRequest{ - Executable: exe, - Args: []string{"-test.run=TestSubprocessHelper", "--", "printenv"}, - EnvOverrides: map[string]string{"SUBPROCESS_PARENT_VALUE": "override-value"}, + Executable: exe, + Args: []string{"-test.run=TestSubprocessHelper", "--", "printenv"}, + EnvOverrides: map[string]string{ + "GO_WANT_SUBPROCESS_HELPER": "1", + "SUBPROCESS_HELPER_ENV_KEY": "SUBPROCESS_PARENT_VALUE", + "SUBPROCESS_PARENT_VALUE": "override-value", + }, StdoutLogPath: stdoutPath, } @@ -369,6 +496,34 @@ func TestSubprocessHelper(t *testing.T) { key := os.Getenv("SUBPROCESS_HELPER_ENV_KEY") _, _ = os.Stdout.WriteString(os.Getenv(key) + "\n") os.Exit(0) + case "echoenv": + key := os.Getenv("SUBPROCESS_HELPER_ENV_KEY") + value := os.Getenv(key) + _, _ = os.Stdout.WriteString(value) + _, _ = os.Stderr.WriteString(value) + os.Exit(5) + case "spam": + chunk := strings.Repeat("x", 64*1024) + count, _ := strconv.Atoi(os.Getenv("SUBPROCESS_HELPER_CHUNKS")) + for range count { + _, _ = os.Stdout.WriteString(chunk) + } + os.Exit(0) + case "tree-spam": + descendant := exec.Command(os.Args[0], "-test.run=^TestSubprocessHelper$", "--", "descendant") + descendant.Env = append(os.Environ(), "GO_WANT_SUBPROCESS_HELPER=1") + descendant.Stdout = os.Stdout + descendant.Stderr = os.Stderr + if err := descendant.Start(); err != nil { + os.Exit(3) + } + if err := os.WriteFile(os.Getenv("SUBPROCESS_HELPER_READY_PATH"), []byte("ready"), 0o600); err != nil { + os.Exit(4) + } + chunk := strings.Repeat("x", 64*1024) + for { + _, _ = os.Stdout.WriteString(chunk) + } case "tree": descendant := exec.Command(os.Args[0], "-test.run=^TestSubprocessHelper$", "--", "descendant") descendant.Env = append(os.Environ(), "GO_WANT_SUBPROCESS_HELPER=1")