From def8159ded4ac431bf1af1290c970f07a8ea66b4 Mon Sep 17 00:00:00 2001 From: Jon Langevin Date: Wed, 14 Oct 2026 19:16:47 -0400 Subject: [PATCH 1/2] fix(wfctl): retain safe record-host failure evidence --- cmd/wfctl/pipeline_docker_unix_test.go | 19 +- cmd/wfctl/pipeline_record_child_unix.go | 2 +- cmd/wfctl/pipeline_record_diagnostic_unix.go | 207 ++++++ .../pipeline_record_diagnostic_unix_test.go | 636 ++++++++++++++++++ cmd/wfctl/pipeline_record_host_test.go | 10 + cmd/wfctl/pipeline_record_live_test.go | 13 +- cmd/wfctl/pipeline_record_unix.go | 123 +++- docs/WFCTL.md | 14 + 8 files changed, 1000 insertions(+), 24 deletions(-) create mode 100644 cmd/wfctl/pipeline_record_diagnostic_unix.go create mode 100644 cmd/wfctl/pipeline_record_diagnostic_unix_test.go diff --git a/cmd/wfctl/pipeline_docker_unix_test.go b/cmd/wfctl/pipeline_docker_unix_test.go index 2d2a53da..42e2576f 100644 --- a/cmd/wfctl/pipeline_docker_unix_test.go +++ b/cmd/wfctl/pipeline_docker_unix_test.go @@ -37,6 +37,7 @@ type pipelineDockerFixture struct { HelperBlocks bool `json:"helper_blocks"` Group pipelineProcessIdentity `json:"group"` LateCreate bool `json:"late_create"` + WaitStatus string `json:"wait_status,omitempty"` } type pipelineDockerInvocation struct { @@ -154,6 +155,8 @@ func TestPipelineDockerCLIHelper(t *testing.T) { os.Exit(0) } switch command { + case "pull": + // Synthetic setup transition only; no image or container is fetched. case "context": fmt.Print(fixture.Context) case "info": @@ -192,7 +195,11 @@ func TestPipelineDockerCLIHelper(t *testing.T) { _ = os.WriteFile(filepath.Join(dir, "fixture.json"), data, 0600) } case "wait": - fmt.Println("0") + if fixture.WaitStatus != "" { + fmt.Println(fixture.WaitStatus) + } else { + fmt.Println("0") + } case "create": id := strings.Repeat("a", 64) for i, arg := range commandArgs { @@ -216,6 +223,16 @@ func TestPipelineDockerCLIHelper(t *testing.T) { default: os.Exit(94) } + // Rotate only after the preceding command has succeeded. The next real + // client's identity preflight then fails before it can start another CLI. + if (command == "create" && fixture.Fail == "after-create-server-change") || + (command == "start" && fixture.Fail == "after-start-server-change") { + fixture.ServerID = "synthetic-rotated-daemon" + updated, err := json.Marshal(fixture) + if err != nil || os.WriteFile(filepath.Join(dir, "fixture.json"), updated, 0600) != nil { + os.Exit(99) + } + } os.Exit(0) } diff --git a/cmd/wfctl/pipeline_record_child_unix.go b/cmd/wfctl/pipeline_record_child_unix.go index 0226e874..dc09fe4d 100644 --- a/cmd/wfctl/pipeline_record_child_unix.go +++ b/cmd/wfctl/pipeline_record_child_unix.go @@ -22,7 +22,7 @@ func pipelineRecordChildEnvironment(source []string, toolDirectory string) []str env := make([]string, 0, len(source)+3) for _, variable := range source { name, _, _ := strings.Cut(variable, "=") - if strings.HasPrefix(name, "DOCKER_") || name == "PATH" { + if strings.HasPrefix(name, "DOCKER_") || name == "PATH" || name == pipelineRecordDiagnosticEnv { continue } env = append(env, variable) diff --git a/cmd/wfctl/pipeline_record_diagnostic_unix.go b/cmd/wfctl/pipeline_record_diagnostic_unix.go new file mode 100644 index 00000000..dee6ca42 --- /dev/null +++ b/cmd/wfctl/pipeline_record_diagnostic_unix.go @@ -0,0 +1,207 @@ +//go:build linux || darwin + +package main + +import ( + "context" + "encoding/json" + "io" + "os" + "sync" + "syscall" +) + +const pipelineRecordDiagnosticEnv = "WFCTL_RECORD_DIAGNOSTICS" +const pipelineRecordDiagnosticPrefix = "WFCTL_RECORD_DIAGNOSTIC_V1 " +const maxPipelineRecordDiagnosticBytes = 1024 + +// Diagnostic values describe observations, never workload strings or errors. +// Unknown exits are -1; a Docker CLI exit is not a confirmed container exit. +type pipelineRecordDiagnosticReport struct { + Phase string `json:"phase"` + Event string `json:"event"` + ChildObserved bool `json:"child_observed"` + ChildExit int `json:"child_exit"` + ChildSignal int `json:"child_signal"` + Operation string `json:"docker_operation"` + DockerExit int `json:"docker_cli_exit"` + ContainerExit int `json:"container_exit"` + Cleanup string `json:"cleanup"` + Cancelled bool `json:"cancelled"` + Deadline bool `json:"deadline"` +} + +var pipelineRecordDiagnosticPhases = map[string]bool{ + "unknown": true, "state_root": true, "state_open": true, "state_entries": true, + "retained_processes": true, "docker_resolve": true, "retained_cleanup": true, + "config_load": true, "plugin_inspect": true, "plugin_resolve": true, + "parent_identity": true, "request_encode": true, "nonce": true, "state_save": true, + "child_setup": true, "child_start": true, "child_identity": true, "child_journal": true, + "broker_closed": true, "cancelled": true, "child_quarantine": true, "child_exit": true, + "child_pipe": true, "child_input": true, "broker_stop": true, "child_cleanup": true, + "result_frame": true, "child_result": true, "container_args": true, + "container_create": true, "container_id": true, "container_attach": true, + "container_wait": true, "container_status": true, "container_command": true, + "cleanup": true, "pipeline_result": true, "record_write": true, +} + +type pipelineRecordDiagnostic struct { + mu sync.Mutex + report pipelineRecordDiagnosticReport +} + +type pipelineRecordDiagnosticKey struct{} + +func newPipelineRecordDiagnostic() *pipelineRecordDiagnostic { + return &pipelineRecordDiagnostic{report: pipelineRecordDiagnosticReport{ + Event: "unknown", ChildExit: -1, Operation: "unknown", DockerExit: -1, ContainerExit: -1, Cleanup: "not_started", + }} +} + +func pipelineRecordDiagnosticFromContext(ctx context.Context) *pipelineRecordDiagnostic { + diagnostic, _ := ctx.Value(pipelineRecordDiagnosticKey{}).(*pipelineRecordDiagnostic) + return diagnostic +} + +func (d *pipelineRecordDiagnostic) fail(phase string) { + if d == nil { + return + } + if !pipelineRecordDiagnosticPhases[phase] { + phase = "unknown" + } + d.mu.Lock() + defer d.mu.Unlock() + if d.report.Phase == "" { + d.report.Phase = phase + } +} + +func (d *pipelineRecordDiagnostic) event(event string) { + if d == nil { + return + } + switch event { + case "result", "broker", "context", "quarantine": + default: + event = "unknown" + } + d.mu.Lock() + d.report.Event = event + d.mu.Unlock() +} + +func (d *pipelineRecordDiagnostic) child(state *os.ProcessState) { + if d == nil || state == nil { + return + } + d.mu.Lock() + defer d.mu.Unlock() + d.report.ChildObserved = true + d.report.ChildExit = pipelineRecordDiagnosticExit(state.ExitCode()) + if status, ok := state.Sys().(syscall.WaitStatus); ok && status.Signaled() { + signal := int(status.Signal()) + if signal > 0 && signal <= 64 { + d.report.ChildSignal = signal + } + } +} + +func pipelineRecordDiagnosticExit(code int) int { + if code < 0 || code > 255 { + return -1 + } + return code +} + +func (d *pipelineRecordDiagnostic) beginOperation(operation string) { + if d == nil { + return + } + switch operation { + case "create", "start", "wait": + default: + operation = "unknown" + } + d.mu.Lock() + defer d.mu.Unlock() + if d.report.Phase == "" { + d.report.Operation = operation + d.report.DockerExit, d.report.ContainerExit = -1, -1 + } +} + +func (d *pipelineRecordDiagnostic) dockerExit(code int) { + if d == nil { + return + } + d.mu.Lock() + if d.report.Phase == "" { + d.report.DockerExit = pipelineRecordDiagnosticExit(code) + } + d.mu.Unlock() +} + +func (d *pipelineRecordDiagnostic) containerExit(code int) { + if d == nil { + return + } + d.mu.Lock() + if d.report.Phase == "" { + d.report.ContainerExit = pipelineRecordDiagnosticExit(code) + } + d.mu.Unlock() +} + +func (d *pipelineRecordDiagnostic) cleanup(outcome string) { + if d == nil { + return + } + switch outcome { + case "complete", "failed", "incomplete", "not_started": + default: + outcome = "unknown" + } + d.mu.Lock() + d.report.Cleanup = outcome + d.mu.Unlock() +} + +func (d *pipelineRecordDiagnostic) result(frame []byte) { + if d == nil { + return + } + // The caller has observed resultDone. This only classifies existing bounded + // bytes before teardown; the original decoder still makes every decision. + envelope, err := decodePipelineResultFrame(frame) + if err != nil { + d.fail("result_frame") + return + } + _, code, err := selectPipelineRecordEnvelope(envelope) + if err != nil { + d.fail("result_frame") + } else if code != "" { + d.fail("child_result") + } +} + +func (d *pipelineRecordDiagnostic) emit(ctx context.Context, writer io.Writer) { + if d == nil { + return + } + d.mu.Lock() + report := d.report + d.mu.Unlock() + if report.Phase == "" { + report.Phase = "unknown" + } + report.Cancelled = ctx.Err() == context.Canceled + report.Deadline = ctx.Err() == context.DeadlineExceeded + data, err := json.Marshal(report) + if err != nil || len(data)+len(pipelineRecordDiagnosticPrefix)+1 > maxPipelineRecordDiagnosticBytes { + return + } + // Diagnostic delivery cannot change the original execution result. + _, _ = io.WriteString(writer, pipelineRecordDiagnosticPrefix+string(data)+"\n") +} diff --git a/cmd/wfctl/pipeline_record_diagnostic_unix_test.go b/cmd/wfctl/pipeline_record_diagnostic_unix_test.go new file mode 100644 index 00000000..5954c532 --- /dev/null +++ b/cmd/wfctl/pipeline_record_diagnostic_unix_test.go @@ -0,0 +1,636 @@ +//go:build linux || darwin + +package main + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "io" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/GoCodeAlone/workflow/sandbox" + "golang.org/x/sys/unix" +) + +// Parse only the closed status schema. Unrelated stderr is never reflected. +func safePipelineRecordDiagnostic(stderr []byte) (pipelineRecordDiagnosticReport, string) { + var report pipelineRecordDiagnosticReport + if len(stderr) > maxPipelineRecordBytes { + return report, "oversize" + } + var payload []byte + for _, line := range bytes.Split(stderr, []byte{'\n'}) { + if !bytes.HasPrefix(line, []byte(pipelineRecordDiagnosticPrefix)) { + continue + } + if payload != nil { + return report, "invalid" + } + if len(line)+1 > maxPipelineRecordDiagnosticBytes { + return report, "oversize" + } + payload = line[len(pipelineRecordDiagnosticPrefix):] + } + if payload == nil { + return report, "missing" + } + if validatePipelinePrivateJSON(payload) != nil { + return report, "invalid" + } + var properties map[string]json.RawMessage + required := []string{"phase", "event", "child_observed", "child_exit", "child_signal", "docker_operation", "docker_cli_exit", "container_exit", "cleanup", "cancelled", "deadline"} + if json.Unmarshal(payload, &properties) != nil || len(properties) != len(required) { + return report, "invalid" + } + // encoding/json accepts case aliases and null scalars. Reject both before + // typed decoding can turn an absent/unknown observed status into zero. + for _, name := range required { + value, present := properties[name] + if !present || bytes.Equal(bytes.TrimSpace(value), []byte("null")) { + return report, "invalid" + } + } + decoder := json.NewDecoder(bytes.NewReader(payload)) + decoder.DisallowUnknownFields() + if decoder.Decode(&report) != nil || decoder.Decode(new(any)) != io.EOF { + return pipelineRecordDiagnosticReport{}, "invalid" + } + if !pipelineRecordDiagnosticPhases[report.Phase] || + pipelineRecordDiagnosticExit(report.ChildExit) != report.ChildExit || + pipelineRecordDiagnosticExit(report.DockerExit) != report.DockerExit || + pipelineRecordDiagnosticExit(report.ContainerExit) != report.ContainerExit || + (report.Operation == "unknown" && (report.DockerExit != -1 || report.ContainerExit != -1)) || + (report.ContainerExit != -1 && report.Operation != "wait") || + report.ChildSignal < 0 || report.ChildSignal > 64 || + (!report.ChildObserved && (report.ChildExit != -1 || report.ChildSignal != 0)) || + (report.ChildSignal != 0 && report.ChildExit != -1) || (report.Cancelled && report.Deadline) { + return pipelineRecordDiagnosticReport{}, "invalid" + } + switch report.Event { + case "unknown", "result", "broker", "context", "quarantine": + default: + return pipelineRecordDiagnosticReport{}, "invalid" + } + switch report.Cleanup { + case "unknown", "not_started", "complete", "failed", "incomplete": + default: + return pipelineRecordDiagnosticReport{}, "invalid" + } + switch report.Operation { + case "unknown", "create", "start", "wait": + default: + return pipelineRecordDiagnosticReport{}, "invalid" + } + return report, "present" +} + +func logSafePipelineRecordDiagnostic(t *testing.T, stderr []byte) { + t.Helper() + report, status := safePipelineRecordDiagnostic(stderr) + if status != "present" { + t.Logf("record diagnostic: status=%s", status) + return + } + data, err := json.Marshal(report) + if err != nil { + t.Log("record diagnostic: status=invalid") + return + } + t.Logf("record diagnostic: %s", data) +} + +func pipelineRecordHostExit(err error) int { + if err == nil { + return 0 + } + var exit *exec.ExitError + if errors.As(err, &exit) && exit.ProcessState != nil { + return pipelineRecordDiagnosticExit(exit.ProcessState.ExitCode()) + } + return -1 +} + +func pipelineRecordLiveBinaryDigest(path string) string { + info, err := os.Lstat(path) + if err != nil || !info.Mode().IsRegular() { + return "unavailable" + } + file, err := os.Open(path) + if err != nil { + return "unavailable" + } + defer file.Close() + opened, err := file.Stat() + if err != nil || !os.SameFile(info, opened) || !opened.Mode().IsRegular() { + return "unavailable" + } + digest := sha256.New() + const limit = 512 << 20 + n, err := io.Copy(digest, io.LimitReader(file, limit+1)) + if err != nil || n > limit { + return "unavailable" + } + return hex.EncodeToString(digest.Sum(nil)) +} + +func pipelineRecordLiveCleanupSummary(root string) (string, int, int) { + file, err := os.Open(filepath.Join(root, "state", "wfctl", "pipeline-cleanup")) + if errors.Is(err, os.ErrNotExist) { + return "missing", 0, 0 + } + if err != nil { + return "read_error", 0, 0 + } + defer file.Close() + entries, err := file.ReadDir(65) + if err != nil && !errors.Is(err, io.EOF) { + return "read_error", 0, 0 + } + status := "observed" + if len(entries) > 64 { + entries, status = entries[:64], "capped" + } + journals := 0 + for _, entry := range entries { + if strings.HasSuffix(entry.Name(), ".json") { + journals++ + } + } + return status, len(entries), journals +} + +func logPipelineRecordLiveFailure(t *testing.T, root, binary, imageDigest string) { + t.Helper() + if !t.Failed() { + return + } + // Digest/count observations run before TempDir cleanup and only on failure. + // No journal contents, names, paths, image name or workload output is read out. + status, entries, journals := pipelineRecordLiveCleanupSummary(root) + t.Logf("live fixture identity: executable_sha256=%s image_sha256=%s", pipelineRecordLiveBinaryDigest(binary), imageDigest) + t.Logf("live fixture cleanup: status=%s entries=%d journals=%d", status, entries, journals) +} + +func diagnosticTestReport(t *testing.T, diagnostic *pipelineRecordDiagnostic) (pipelineRecordDiagnosticReport, []byte) { + t.Helper() + var output bytes.Buffer + diagnostic.emit(context.Background(), &output) + report, status := safePipelineRecordDiagnostic(output.Bytes()) + if status != "present" { + t.Fatalf("diagnostic did not satisfy its bounded closed schema: status=%s", status) + } + return report, output.Bytes() +} + +func TestPipelineRecordDiagnosticsBrokerInjectedFailures(t *testing.T) { + for _, tc := range []struct { + name, fail, wait, phase, operation string + cli, container int + wantError bool + }{ + {name: "create", fail: "create", phase: "container_create", operation: "create", cli: 7, container: -1, wantError: true}, + {name: "attach", fail: "start", phase: "container_attach", operation: "start", cli: 7, container: -1, wantError: true}, + {name: "wait", fail: "wait", phase: "container_wait", operation: "wait", cli: 7, container: -1, wantError: true}, + {name: "attach-before-command", fail: "after-create-server-change", phase: "container_attach", operation: "start", cli: -1, container: -1, wantError: true}, + {name: "wait-before-command", fail: "after-start-server-change", phase: "container_wait", operation: "wait", cli: -1, container: -1, wantError: true}, + {name: "invalid-status", wait: "secret-invalid-status", phase: "container_status", operation: "wait", cli: 0, container: -1, wantError: true}, + {name: "nonzero-container", wait: "23", phase: "container_command", operation: "wait", cli: 0, container: 23}, + {name: "success", wait: "0", phase: "unknown", operation: "wait", cli: 0, container: 0}, + } { + t.Run(tc.name, func(t *testing.T) { + dir, fixture := newPipelineDockerFixture(t) + client := resolvePipelineDockerFixture(t) + fixture.Fail, fixture.WaitStatus = tc.fail, tc.wait + writePipelineDockerFixture(t, dir, fixture) + state, err := openPipelineCleanupState(filepath.Join(pipelineRecordHostTempDir(t), "state")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = state.Close() }) + parent, err := pipelineProcessStartToken(os.Getpid()) + if err != nil { + t.Fatal(err) + } + entry := pipelineCleanupEntry{Version: 1, Label: pipelineDockerTestLabel, + Parent: pipelineProcessIdentity{PID: os.Getpid(), Start: parent}, + Child: newPipelineDockerProcessGroup(t), Docker: client.Identity, + Deadline: time.Now().Add(time.Minute).UTC().Format(time.RFC3339Nano)} + diagnostic := newPipelineRecordDiagnostic() + ctx := context.WithValue(context.Background(), pipelineRecordDiagnosticKey{}, diagnostic) + result, err := executePipelineBrokerSandbox(ctx, pipelineBrokerRequest{ + Config: sandbox.DefaultSecureSandboxConfig("alpine:latest"), Command: []string{"true"}, + }, state, client, &entry) + if (err != nil) != tc.wantError || (!tc.wantError && (result == nil || result.ExitCode != tc.container)) { + t.Fatalf("injected operation did not produce expected outcome: error=%t", err != nil) + } + report, output := diagnosticTestReport(t, diagnostic) + if report.Phase != tc.phase || report.Operation != tc.operation || report.DockerExit != tc.cli || report.ContainerExit != tc.container { + t.Fatalf("diagnostic lost operation/status origin: %+v", report) + } + for _, secret := range []string{"TEST_SECRET_TOKEN", "secret-invalid-status", "raw daemon stderr", dir, parent} { + if bytes.Contains(output, []byte(secret)) { + t.Fatal("diagnostic reflected private operation data") + } + } + if tc.name == "create" { + // A later successful real broker execution cannot rewrite the + // operation/status attached to the first failed execution. + fixture.Fail = "" + writePipelineDockerFixture(t, dir, fixture) + if _, err := executePipelineBrokerSandbox(ctx, pipelineBrokerRequest{ + Config: sandbox.DefaultSecureSandboxConfig("alpine:latest"), Command: []string{"true"}, + }, state, client, &entry); err != nil { + t.Fatal("subsequent synthetic broker execution failed") + } + report, _ = diagnosticTestReport(t, diagnostic) + if report.Phase != "container_create" || report.Operation != "create" || report.DockerExit != 7 || report.ContainerExit != -1 { + t.Fatal("subsequent success rewrote first-failure status") + } + } + if tc.name == "nonzero-container" { + // A real cleanup failure must not mask the earlier container failure. + fixture.Containers[strings.Repeat("a", 64)] = pipelineDockerTestLabel + fixture.Fail = "rm" + writePipelineDockerFixture(t, dir, fixture) + if finishPipelineCleanupObserved(ctx, state, client, entry, "cleanup") == nil { + t.Fatal("injected cleanup failure was silently accepted") + } + report, _ = diagnosticTestReport(t, diagnostic) + if report.Phase != "container_command" || report.Cleanup != "failed" { + t.Fatalf("cleanup masked first failure: %+v", report) + } + } + }) + } +} + +func TestPipelineRecordDiagnosticsChildStatusAndConcurrency(t *testing.T) { + for _, tc := range []struct { + name, script string + exit, signal int + }{ + {name: "exit", script: "exit 23", exit: 23}, + {name: "signal", script: "kill -TERM $$", exit: -1, signal: 15}, + } { + t.Run(tc.name, func(t *testing.T) { + command := exec.Command("/bin/sh", "-c", tc.script) + if command.Run() == nil { + t.Fatal("synthetic child did not fail") + } + diagnostic := newPipelineRecordDiagnostic() + diagnostic.fail("child_exit") + diagnostic.child(command.ProcessState) + var workers sync.WaitGroup + for range 8 { + workers.Go(func() { + for range 20 { + diagnostic.beginOperation("create") + diagnostic.dockerExit(0) + diagnostic.containerExit(0) + diagnostic.fail("cleanup") + diagnostic.cleanup("failed") + var output bytes.Buffer + diagnostic.emit(context.Background(), &output) + } + }) + } + workers.Wait() + report, _ := diagnosticTestReport(t, diagnostic) + if !report.ChildObserved || report.ChildExit != tc.exit || report.ChildSignal != tc.signal || report.Phase != "child_exit" || report.Operation != "unknown" || report.DockerExit != -1 || report.ContainerExit != -1 { + t.Fatalf("lost actual child status under concurrent observations: %+v", report) + } + }) + } + for _, code := range []int{-999, -2, 256, 1 << 20} { + diagnostic := newPipelineRecordDiagnostic() + diagnostic.beginOperation("private-operation-canary") + diagnostic.dockerExit(code) + diagnostic.containerExit(code) + diagnostic.fail("private-phase-canary") + diagnostic.event("private-event-canary") + diagnostic.cleanup("private-cleanup-canary") + report, output := diagnosticTestReport(t, diagnostic) + if report.Phase != "unknown" || report.Event != "unknown" || report.Operation != "unknown" || report.Cleanup != "unknown" || report.DockerExit != -1 || report.ContainerExit != -1 || bytes.Contains(output, []byte("canary")) { + t.Fatal("diagnostic accepted arbitrary strings or out-of-range status") + } + } +} + +func TestPipelineRecordDiagnosticsRejectUntrustedReports(t *testing.T) { + diagnostic := newPipelineRecordDiagnostic() + diagnostic.fail("container_create") + _, valid := diagnosticTestReport(t, diagnostic) + aliasMissing := bytes.Replace(valid, []byte(`"container_exit":-1,`), nil, 1) + aliasMissing = bytes.Replace(aliasMissing, []byte(`"phase":`), []byte(`"PHASE":"cleanup","phase":`), 1) + for name, data := range map[string][]byte{ + "missing": []byte("private stderr canary"), + "malformed": []byte(pipelineRecordDiagnosticPrefix + "{private-canary}"), + "unknown-field": bytes.Replace(valid, []byte(`"phase":`), []byte(`"private":"canary","phase":`), 1), + "null-status": bytes.Replace(valid, []byte(`"container_exit":-1`), []byte(`"container_exit":null`), 1), + "missing-field": bytes.Replace(valid, []byte(`"deadline":false`), []byte(`"foreign":false`), 1), + "duplicate-key": bytes.Replace(valid, []byte(`"phase":`), []byte(`"phase":"cleanup","phase":`), 1), + "unknown-phase": bytes.Replace(valid, []byte(`"container_create"`), []byte(`"private-phase-canary"`), 1), + "exit-range": bytes.Replace(valid, []byte(`"docker_cli_exit":-1`), []byte(`"docker_cli_exit":999`), 1), + "exit-without-operation": bytes.Replace(valid, []byte(`"docker_cli_exit":-1`), []byte(`"docker_cli_exit":0`), 1), + "signal-range": bytes.Replace(valid, []byte(`"child_signal":0`), []byte(`"child_signal":999`), 1), + "oversize": []byte(pipelineRecordDiagnosticPrefix + strings.Repeat("x", maxPipelineRecordDiagnosticBytes)), + "duplicate-report": append(append([]byte{}, valid...), valid...), + "case-alias-missing-status": aliasMissing, + } { + t.Run(name, func(t *testing.T) { + if _, status := safePipelineRecordDiagnostic(data); status == "present" { + t.Fatal("accepted an untrusted diagnostic report") + } + }) + } + data := append([]byte("private surrounding stderr canary\n"), valid...) + if _, status := safePipelineRecordDiagnostic(data); status != "present" { + t.Fatal("unrelated stderr hid the bounded diagnostic") + } +} + +func TestPipelineRecordDiagnosticsPreserveEarlyResultFailure(t *testing.T) { + diagnostic := newPipelineRecordDiagnostic() + frame, err := encodePipelineResultFrame(map[string]any{"status": "error", "code": "pipeline_failed"}) + if err != nil { + t.Fatal(err) + } + diagnostic.result(frame) + diagnostic.fail("child_cleanup") + diagnostic.cleanup("incomplete") + report, _ := diagnosticTestReport(t, diagnostic) + if report.Phase != "child_result" || report.Cleanup != "incomplete" { + t.Fatal("teardown masked the already-drained error result") + } + diagnostic = newPipelineRecordDiagnostic() + diagnostic.result([]byte("private malformed frame canary")) + report, output := diagnosticTestReport(t, diagnostic) + if report.Phase != "result_frame" || bytes.Contains(output, []byte("canary")) { + t.Fatal("invalid private frame was reflected or unclassified") + } +} + +func TestPipelineRecordDiagnosticsActualHostFailureAndSilence(t *testing.T) { + root := pipelineRecordHostTempDir(t) + binary := filepath.Join(root, "wfctl") + // Build before fixture HOME/opt-in changes, reusing the ordinary shared pool. + buildPipelineRecordHostBinary(t, binary, ".") + for _, tc := range []struct { + name, enabled, config, phase string + fail bool + }{ + {name: "default-success", config: "builtin.yaml"}, + {name: "opted-success", enabled: "1", config: "builtin.yaml"}, + {name: "default-failure", config: "cli-print.yaml", fail: true}, + {name: "nonliteral-opt-in-failure", enabled: "true", config: "cli-print.yaml", fail: true}, + {name: "opted-failure", enabled: "1", config: "cli-print.yaml", phase: "child_quarantine", fail: true}, + } { + t.Run(tc.name, func(t *testing.T) { + newPipelineDockerFixture(t) + t.Setenv("XDG_STATE_HOME", filepath.Join(pipelineRecordHostTempDir(t), "state")) + t.Setenv(pipelineRecordDiagnosticEnv, tc.enabled) + stdout, stderr, err := runPipelineRecordHost(t, binary, filepath.Join("testdata/pipeline-record", tc.config), "") + if tc.fail { + assertPipelineRecordHostError(t, stdout, stderr, err, "pipeline_failed") + } else if err != nil || string(stdout) != "SDK_RECORD_V1 {\"ready\":true,\"source\":\"builtin\"}\n" { + t.Fatalf("diagnostics changed actual host success: exit=%d", pipelineRecordHostExit(err)) + } + report, status := safePipelineRecordDiagnostic(stderr) + if tc.phase == "" { + if status != "missing" { + t.Fatal("diagnostics emitted on success or without exact opt-in") + } + } else if status != "present" || report.Phase != tc.phase || !report.ChildObserved || report.Cleanup != "complete" { + t.Fatalf("actual host failure lost phase/child/cleanup evidence: status=%s report=%+v", status, report) + } + if bytes.Contains(stderr, []byte("fixture-cli-stdout-canary")) { + t.Fatal("actual host diagnostics reflected quarantined workload output") + } + }) + } +} + +func TestPipelineRecordDiagnosticsChildOptInNotInherited(t *testing.T) { + for _, value := range pipelineRecordChildEnvironment([]string{pipelineRecordDiagnosticEnv + "=1", "PUBLIC_VALUE=selected"}, "/fixture/tools") { + if strings.HasPrefix(value, pipelineRecordDiagnosticEnv+"=") { + t.Fatal("record child inherited the parent's diagnostic opt-in") + } + } +} + +func TestPipelineRecordDiagnosticsActualParentChildSignal(t *testing.T) { + root := pipelineRecordHostTempDir(t) + binary := filepath.Join(root, "wfctl") + buildPipelineRecordHostBinary(t, binary, ".") + pluginDir := filepath.Join(root, "plugins") + installed := filepath.Join(pluginDir, "record-fixture") + if err := os.MkdirAll(installed, 0700); err != nil { + t.Fatal(err) + } + buildPipelineRecordHostBinary(t, filepath.Join(installed, "record-fixture"), "./testdata/pipeline-record/plugin") + manifest, err := os.ReadFile("testdata/pipeline-record/plugin/plugin.json") + if err != nil || os.WriteFile(filepath.Join(installed, "plugin.json"), manifest, 0600) != nil { + t.Fatal("could not install synthetic SDK manifest") + } + newPipelineDockerFixture(t) + t.Setenv("XDG_STATE_HOME", filepath.Join(root, "state")) + t.Setenv(pipelineRecordDiagnosticEnv, "1") + audit := filepath.Join(root, "synthetic-sdk-ready") + process := startPipelineRecordHost(t, root, binary, "testdata/pipeline-record/external.yaml", pluginDir, + "--var", "mode=wait", "--var", "audit_path="+audit) + entry := readPipelineRecordHostEntry(t, root) + if entry.Parent.PID != process.command.Process.Pid { + t.Fatal("synthetic journal belongs to a different record parent") + } + waitPipelineRecordHostCondition(t, "synthetic SDK wait boundary", func() bool { + _, err := os.Stat(audit) + return err == nil + }) + token, tokenErr := pipelineProcessStartToken(entry.Child.PID) + group, groupErr := unix.Getpgid(entry.Child.PID) + if tokenErr != nil || groupErr != nil || token != entry.Child.Start || group != entry.Child.PGID || entry.Child.PGID != entry.Child.PID { + t.Fatal("synthetic child identity was not proven before fault injection") + } + // Kill only this test's recorded leader. The actual record parent retains + // group custody, reaps this status and performs its unchanged cleanup. + if unix.Kill(entry.Child.PID, unix.SIGKILL) != nil { + t.Fatal("could not inject synthetic record child failure") + } + stdout, stderr, runErr := process.finish(t) + assertPipelineRecordHostError(t, stdout, stderr, runErr, "pipeline_failed") + report, status := safePipelineRecordDiagnostic(stderr) + if status != "present" || !report.ChildObserved || report.ChildExit != -1 || report.ChildSignal != int(unix.SIGKILL) || report.Cleanup != "complete" || + (report.Phase != "result_frame" && report.Phase != "broker_closed") { + t.Fatalf("actual parent lost synthetic child wait/cleanup evidence: status=%s report=%+v", status, report) + } + assertPipelineRecordHostJournalEmpty(t, root) +} + +func TestPipelineRecordDiagnosticsCleanupAssertionHelper(t *testing.T) { + if root := os.Getenv("WFCTL_RECORD_DIAGNOSTIC_ASSERT_ROOT"); root != "" { + assertPipelineRecordHostJournalEmpty(t, root) + } +} + +func TestPipelineRecordDiagnosticsCleanupAssertionRedaction(t *testing.T) { + executable, err := os.Executable() + if err != nil { + t.Fatal(err) + } + for _, retained := range []bool{false, true} { + t.Run(map[bool]string{false: "missing", true: "retained"}[retained], func(t *testing.T) { + root := filepath.Join(pipelineRecordHostTempDir(t), "private-path-canary") + if retained { + directory := filepath.Join(root, "state", "wfctl", "pipeline-cleanup") + if err := os.MkdirAll(directory, 0700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(directory, "private-journal-canary.json"), []byte("private-content-canary"), 0600); err != nil { + t.Fatal(err) + } + } + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + command := exec.CommandContext(ctx, executable, "-test.run=^TestPipelineRecordDiagnosticsCleanupAssertionHelper$") + command.Env = append(os.Environ(), pipelineRecordDiagnosticEnv+"=1", "WFCTL_RECORD_DIAGNOSTIC_ASSERT_ROOT="+root) + output, runErr := command.CombinedOutput() + if ctx.Err() != nil || pipelineRecordHostExit(runErr) != 1 || !bytes.Contains(output, []byte("host ")) { + t.Fatal("synthetic cleanup assertion did not fail through its bounded safe path") + } + for _, secret := range []string{root, "private-path-canary", "private-journal-canary", "private-content-canary"} { + if bytes.Contains(output, []byte(secret)) { + t.Fatal("opted-in cleanup assertion reflected a private path/name/content") + } + } + }) + } +} + +func TestPipelineRecordDiagnosticsImageSetupHelper(t *testing.T) { + if os.Getenv("WFCTL_RECORD_DIAGNOSTIC_IMAGE_SETUP") == "1" { + pullPipelineRecordLiveImage(t) + } +} + +func TestPipelineRecordDiagnosticsImageSetupRedaction(t *testing.T) { + executable, err := os.Executable() + if err != nil { + t.Fatal(err) + } + for _, tc := range []struct{ command, phase string }{{"pull", "pull"}, {"image", "image_inspect"}} { + t.Run(tc.command, func(t *testing.T) { + const ambientCanary = "synthetic-image-setup-credential-canary" + t.Setenv("GITHUB_TOKEN", ambientCanary) + t.Setenv("NODE_AUTH_TOKEN", ambientCanary) + directory, fixture := newPipelineDockerFixture(t) + fixture.Fail = tc.command + writePipelineDockerFixture(t, directory, fixture) + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + command := exec.CommandContext(ctx, executable, "-test.run=^TestPipelineRecordDiagnosticsImageSetupHelper$") + // This setup path invokes Docker directly, whose test helper records + // its environment. Supply only fixture and runtime settings. + command.Env = []string{ + "PATH=" + strings.Join([]string{directory, "/usr/bin", "/bin"}, string(os.PathListSeparator)), + "HOME=" + directory, + "TMPDIR=" + directory, + "GORACE=" + os.Getenv("GORACE"), + "WFCTL_RECORD_DIAGNOSTIC_IMAGE_SETUP=1", + pipelineRecordDiagnosticEnv + "=", + } + for _, key := range []string{"GOCOVERDIR", "GOMAXPROCS", "GOMEMLIMIT"} { + if value, present := os.LookupEnv(key); present { + command.Env = append(command.Env, key+"="+value) + } + } + output, runErr := command.CombinedOutput() + want := "phase=" + tc.phase + " host_exit=7 cancelled=false deadline=false" + if ctx.Err() != nil || pipelineRecordHostExit(runErr) != 1 || !bytes.Contains(output, []byte(want)) { + t.Fatal("synthetic image setup did not expose only its fixed failure phase/status") + } + for _, secret := range []string{"TEST_SECRET_TOKEN", "raw daemon stderr", directory, ambientCanary, "GITHUB_TOKEN", "NODE_AUTH_TOKEN"} { + if bytes.Contains(output, []byte(secret)) { + t.Fatal("image setup reflected private Docker output or paths") + } + } + calls, err := os.ReadFile(filepath.Join(directory, "calls.jsonl")) + if err != nil || len(calls) == 0 { + t.Fatal("synthetic image setup did not retain its private invocation capture") + } + for _, secret := range []string{ambientCanary, "GITHUB_TOKEN", "NODE_AUTH_TOKEN"} { + if bytes.Contains(calls, []byte(secret)) { + t.Fatal("image setup copied an ambient credential into its private capture") + } + } + }) + } +} + +func TestPipelineRecordDiagnosticsLiveSummaryBoundsAndIdentity(t *testing.T) { + root := pipelineRecordHostTempDir(t) + if status, _, _ := pipelineRecordLiveCleanupSummary(root); status != "missing" { + t.Fatal("missing cleanup state was reported as observed") + } + directory := filepath.Join(root, "state", "wfctl", "pipeline-cleanup") + if err := os.MkdirAll(directory, 0700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(directory, "private-name-canary.json"), []byte("private journal contents canary"), 0600); err != nil { + t.Fatal(err) + } + if status, entries, journals := pipelineRecordLiveCleanupSummary(root); status != "observed" || entries != 1 || journals != 1 { + t.Fatal("cleanup count did not observe retained state") + } + for i := range 65 { + name := filepath.Join(directory, strings.Repeat("x", i+1)) + if err := os.WriteFile(name, nil, 0600); err != nil { + t.Fatal(err) + } + } + if status, entries, journals := pipelineRecordLiveCleanupSummary(root); status != "capped" || entries != 64 || journals > 1 { + t.Fatal("cleanup summary exceeded its directory observation bound") + } + binary := filepath.Join(root, "synthetic-binary") + const content = "synthetic executable bytes" + if err := os.WriteFile(binary, []byte(content), 0700); err != nil { + t.Fatal(err) + } + want := sha256.Sum256([]byte(content)) + if pipelineRecordLiveBinaryDigest(binary) != hex.EncodeToString(want[:]) { + t.Fatal("fixture digest did not identify the consumed bytes") + } + link := filepath.Join(root, "link") + if err := os.Symlink(binary, link); err != nil { + t.Fatal(err) + } + if pipelineRecordLiveBinaryDigest(link) != "unavailable" { + t.Fatal("fixture identity followed a substituted symlink") + } +} + +func TestPipelineRecordDiagnosticsContextClassification(t *testing.T) { + cancelled, cancel := context.WithCancel(context.Background()) + cancel() + deadline, stop := context.WithDeadline(context.Background(), time.Unix(0, 0)) + defer stop() + for _, ctx := range []context.Context{cancelled, deadline} { + diagnostic := newPipelineRecordDiagnostic() + diagnostic.fail("cancelled") + var output bytes.Buffer + diagnostic.emit(ctx, &output) + report, status := safePipelineRecordDiagnostic(output.Bytes()) + if status != "present" || report.Cancelled != (ctx.Err() == context.Canceled) || report.Deadline != (ctx.Err() == context.DeadlineExceeded) { + t.Fatal("context classification lost cancellation/deadline distinction") + } + } +} diff --git a/cmd/wfctl/pipeline_record_host_test.go b/cmd/wfctl/pipeline_record_host_test.go index 2df62981..ade4b4b2 100644 --- a/cmd/wfctl/pipeline_record_host_test.go +++ b/cmd/wfctl/pipeline_record_host_test.go @@ -145,6 +145,10 @@ func runPipelineRecordHost(t *testing.T, binary, config, pluginDir string, extra cmd.Stdout, cmd.Stderr = &stdout, &stderr err := cmd.Run() if ctx.Err() != nil { + if os.Getenv(pipelineRecordDiagnosticEnv) == "1" { + logSafePipelineRecordDiagnostic(t, stderr.Bytes()) + t.Fatalf("built wfctl did not finish within its deadline: host_exit=%d", pipelineRecordHostExit(err)) + } t.Fatalf("built wfctl did not finish within its deadline: %v\nstdout=%s\nstderr=%s", ctx.Err(), stdout.Bytes(), stderr.Bytes()) } return stdout.Bytes(), stderr.Bytes(), err @@ -556,10 +560,16 @@ func assertPipelineRecordHostJournalEmpty(t *testing.T, root string) { t.Helper() entries, err := os.ReadDir(filepath.Join(root, "state", "wfctl", "pipeline-cleanup")) if err != nil { + if os.Getenv(pipelineRecordDiagnosticEnv) == "1" { + t.Fatal("host cleanup state unavailable") + } t.Fatalf("host never established cleanup state: %v", err) } for _, entry := range entries { if strings.HasSuffix(entry.Name(), ".json") { + if os.Getenv(pipelineRecordDiagnosticEnv) == "1" { + t.Fatal("host retained a cleanup journal") + } t.Fatalf("host retained a journal after its empty fake-daemon cleanup: %s", entry.Name()) } } diff --git a/cmd/wfctl/pipeline_record_live_test.go b/cmd/wfctl/pipeline_record_live_test.go index 5df83f58..3ed04489 100644 --- a/cmd/wfctl/pipeline_record_live_test.go +++ b/cmd/wfctl/pipeline_record_live_test.go @@ -34,7 +34,9 @@ func TestPipelineRecordHostLiveDockerSandbox(t *testing.T) { } root := pipelineRecordHostTempDir(t) binary := filepath.Join(root, "wfctl") + t.Cleanup(func() { logPipelineRecordLiveFailure(t, root, binary, strings.ToLower(digest)) }) buildPipelineRecordHostBinary(t, binary, ".") + t.Setenv(pipelineRecordDiagnosticEnv, "1") t.Setenv("XDG_STATE_HOME", filepath.Join(root, "state")) t.Setenv("GITHUB_TOKEN", "host-private-live-canary") t.Setenv("AWS_SESSION_TOKEN", "host-private-live-canary") @@ -77,7 +79,8 @@ cat /work/message` } stdout, stderr, err := runPipelineRecordHost(t, binary, path, "", "--var", "message=live-dynamic-message") if err != nil || string(stdout) != "SDK_RECORD_V1 {\"message\":\"live-dynamic-message\",\"source\":\"live-container\"}\n" { - t.Fatalf("built wfctl failed live sandbox lifecycle: error=%v\nstdout=%s\nstderr=%s", err, stdout, stderr) + logSafePipelineRecordDiagnostic(t, stderr) + t.Fatalf("built wfctl failed live sandbox lifecycle: host_error=%t host_exit=%d", err != nil, pipelineRecordHostExit(err)) } assertPipelineRecordHostJournalEmpty(t, root) } @@ -87,12 +90,14 @@ func pullPipelineRecordLiveImage(t *testing.T) string { ctx, cancel := context.WithTimeout(t.Context(), 90*time.Second) defer cancel() const image = "busybox:1.37.0" - if output, err := exec.CommandContext(ctx, "docker", "pull", image).CombinedOutput(); err != nil { - t.Fatalf("prepare real CI sandbox image: %v\n%s", err, output) + if err := exec.CommandContext(ctx, "docker", "pull", image).Run(); err != nil { + t.Fatalf("prepare real CI sandbox image: phase=pull host_exit=%d cancelled=%t deadline=%t", + pipelineRecordHostExit(err), ctx.Err() == context.Canceled, ctx.Err() == context.DeadlineExceeded) } output, err := exec.CommandContext(ctx, "docker", "image", "inspect", "--format", "{{json .RepoDigests}}", image).Output() if err != nil { - t.Fatalf("resolve real CI sandbox image digest: %v", err) + t.Fatalf("resolve real CI sandbox image digest: phase=image_inspect host_exit=%d cancelled=%t deadline=%t", + pipelineRecordHostExit(err), ctx.Err() == context.Canceled, ctx.Err() == context.DeadlineExceeded) } var digests []string if json.Unmarshal(output, &digests) != nil || len(digests) == 0 { diff --git a/cmd/wfctl/pipeline_record_unix.go b/cmd/wfctl/pipeline_record_unix.go index 79c6b204..dc96beab 100644 --- a/cmd/wfctl/pipeline_record_unix.go +++ b/cmd/wfctl/pipeline_record_unix.go @@ -59,14 +59,23 @@ func runPipelineRecord(request pipelineRecordRequest, options pipelineRecordOpti defer cancel() ctx, deadlineCancel := context.WithTimeout(ctx, 10*time.Minute) defer deadlineCancel() + var diagnostic *pipelineRecordDiagnostic + if os.Getenv(pipelineRecordDiagnosticEnv) == "1" { + diagnostic = newPipelineRecordDiagnostic() + ctx = context.WithValue(ctx, pipelineRecordDiagnosticKey{}, diagnostic) + } output, code := executePipelineRecordParent(ctx, request, options.ConfigPath) if code != "" { + diagnostic.fail("pipeline_result") + diagnostic.emit(ctx, os.Stderr) if err := writePipelineRecord(os.Stdout, options.ErrorPrefix, pipelineRecordErrorPayload(code)); err != nil { return errors.New("pipeline record output failed") } return errors.New("pipeline record execution failed") } if err := writePipelineRecord(os.Stdout, options.SuccessPrefix, output); err != nil { + diagnostic.fail("record_write") + diagnostic.emit(ctx, os.Stderr) return errors.New("pipeline record output failed") } return nil @@ -88,92 +97,114 @@ func pipelineCleanupRoot() (string, error) { } func executePipelineRecordParent(ctx context.Context, request pipelineRecordRequest, configPath string) (map[string]any, string) { - fail := func() (map[string]any, string) { return nil, "pipeline_failed" } + diagnostic := pipelineRecordDiagnosticFromContext(ctx) + fail := func(phase string) (map[string]any, string) { + diagnostic.fail(phase) + return nil, "pipeline_failed" + } root, err := pipelineCleanupRoot() if err != nil { - return fail() + return fail("state_root") } state, err := openPipelineCleanupState(root) if err != nil { - return fail() + return fail("state_open") } defer state.Close() retained, err := state.Entries() if err != nil { - return fail() + return fail("state_entries") } // Kill reuse-safe retained process groups before contacting any daemon or // loading config. A daemon outage must not leave an old child doing work. for i := range retained { entry := retained[i] if reconcilePipelineProcesses(entry) != nil { - return fail() + return fail("retained_processes") } } client, err := resolvePipelineDockerClient(ctx) if err != nil { - return fail() + return fail("docker_resolve") } defer client.Close() for i := range retained { entry := retained[i] - if entry.Docker != client.Identity || finishPipelineCleanup(state, client, entry) != nil { - return fail() + if entry.Docker != client.Identity || finishPipelineCleanupObserved(ctx, state, client, entry, "retained_cleanup") != nil { + diagnostic.cleanup("failed") + return fail("retained_cleanup") } } + diagnostic.cleanup("not_started") request.Config, err = config.LoadFromFile(configPath) if err != nil { - return fail() + return fail("config_load") } closure, err := selectPipelineClosure(request.Config, request.Pipeline, true) if err != nil || closure.names[request.ResultStep] != 1 { + diagnostic.fail("pipeline_result") return nil, "result_invalid" } installed, err := inspectPipelineInstallations(request.PluginDir) if err != nil { - return fail() + return fail("plugin_inspect") } if _, err := installed.resolve(closure.types); err != nil { - return fail() + return fail("plugin_resolve") } request.Config = closure.config parentToken, err := pipelineProcessStartToken(os.Getpid()) if err != nil { - return fail() + return fail("parent_identity") } request.ParentPID, request.ParentStart = os.Getpid(), parentToken data, err := json.Marshal(request) if err != nil || len(data) > maxPipelineRecordBytes { + diagnostic.fail("request_encode") return nil, "result_invalid" } var nonce [16]byte if _, err := rand.Read(nonce[:]); err != nil { - return fail() + return fail("nonce") } deadline, _ := ctx.Deadline() entry := pipelineCleanupEntry{Version: 1, Label: "wfctl-" + hex.EncodeToString(nonce[:]), Parent: pipelineProcessIdentity{PID: os.Getpid(), Start: parentToken}, Docker: client.Identity, Deadline: deadline.UTC().Format(time.RFC3339Nano)} if err := state.Save(entry); err != nil { - return fail() + return fail("state_save") } envelope, childErr := executeIsolatedPipelineChild(ctx, request, data, state, client, &entry) if errors.Is(childErr, errPipelineCleanupIncomplete) { - return fail() + diagnostic.cleanup("incomplete") + return fail("child_cleanup") } - if err := finishPipelineCleanup(state, client, entry); err != nil { - return fail() + if err := finishPipelineCleanupObserved(ctx, state, client, entry, "cleanup"); err != nil { + return fail("cleanup") } if childErr != nil { - return fail() + return fail("child_exit") } output, code, err := selectPipelineRecordEnvelope(envelope) if err != nil { + diagnostic.fail("result_frame") return nil, "result_invalid" } return output, code } +func finishPipelineCleanupObserved(ctx context.Context, state *pipelineCleanupState, client *pipelineDockerClient, entry pipelineCleanupEntry, phase string) error { + err := finishPipelineCleanup(state, client, entry) + diagnostic := pipelineRecordDiagnosticFromContext(ctx) + if err != nil { + diagnostic.cleanup("failed") + diagnostic.fail(phase) + } else { + diagnostic.cleanup("complete") + } + return err +} + func finishPipelineCleanup(state *pipelineCleanupState, client *pipelineDockerClient, entry pipelineCleanupEntry) error { ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() @@ -206,6 +237,7 @@ func finishPipelineCleanup(state *pipelineCleanupState, client *pipelineDockerCl } func executeIsolatedPipelineChild(ctx context.Context, request pipelineRecordRequest, data []byte, state *pipelineCleanupState, client *pipelineDockerClient, entry *pipelineCleanupEntry) (map[string]any, error) { + diagnostic := pipelineRecordDiagnosticFromContext(ctx) var files []*os.File defer func() { for _, file := range files { @@ -221,26 +253,32 @@ func executeIsolatedPipelineChild(ctx context.Context, request pipelineRecordReq } inputRead, inputWrite, err := pipe() if err != nil { + diagnostic.fail("child_setup") return nil, err } resultRead, resultWrite, err := pipe() if err != nil { + diagnostic.fail("child_setup") return nil, err } controlRead, controlWrite, err := pipe() if err != nil { + diagnostic.fail("child_setup") return nil, err } stdoutRead, stdoutWrite, err := pipe() if err != nil { + diagnostic.fail("child_setup") return nil, err } stderrRead, stderrWrite, err := pipe() if err != nil { + diagnostic.fail("child_setup") return nil, err } sockets, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM, 0) if err != nil { + diagnostic.fail("child_setup") return nil, err } for _, fd := range sockets { @@ -251,12 +289,14 @@ func executeIsolatedPipelineChild(ctx context.Context, request pipelineRecordReq files = append(files, brokerParent, brokerChild) brokerConn, err := net.FileConn(brokerParent) if err != nil { + diagnostic.fail("child_setup") return nil, err } _ = brokerParent.Close() defer brokerConn.Close() executable, err := os.Executable() if err != nil { + diagnostic.fail("child_setup") return nil, err } child := exec.Command(executable, pipelineRecordChildCommand) // #nosec G204 -- own executable; no application data in argv. @@ -265,6 +305,7 @@ func executeIsolatedPipelineChild(ctx context.Context, request pipelineRecordReq child.Stdout, child.Stderr = stdoutWrite, stderrWrite child.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} if err := child.Start(); err != nil { + diagnostic.fail("child_start") return nil, err } for _, file := range []*os.File{inputRead, resultWrite, controlRead, brokerChild, stdoutWrite, stderrWrite} { @@ -272,14 +313,18 @@ func executeIsolatedPipelineChild(ctx context.Context, request pipelineRecordReq } childToken, err := pipelineProcessStartToken(child.Process.Pid) if err != nil { + diagnostic.fail("child_identity") _ = child.Process.Kill() _ = child.Wait() + diagnostic.child(child.ProcessState) return nil, err } entry.Child = pipelineProcessIdentity{PID: child.Process.Pid, PGID: child.Process.Pid, Start: childToken} if err := state.Save(*entry); err != nil { + diagnostic.fail("child_journal") _ = child.Process.Kill() _ = child.Wait() + diagnostic.child(child.ProcessState) return nil, err } violation := make(chan struct{}, 1) @@ -316,14 +361,22 @@ func executeIsolatedPipelineChild(ctx context.Context, request pipelineRecordReq brokerStopped := false select { case <-resultDone: + diagnostic.event("result") + diagnostic.result(result.data.Bytes()) case protocolErr = <-brokerDone: + diagnostic.event("broker") + diagnostic.fail("broker_closed") brokerStopped = true if protocolErr == nil { protocolErr = errors.New("sandbox broker closed before result") } case <-ctx.Done(): + diagnostic.event("context") + diagnostic.fail("cancelled") protocolErr = errors.New("pipeline record cancelled") case <-violation: + diagnostic.event("quarantine") + diagnostic.fail("child_quarantine") protocolErr = errors.New("pipeline child output quarantine failed") } _ = controlWrite.Close() @@ -333,10 +386,13 @@ func executeIsolatedPipelineChild(ctx context.Context, request pipelineRecordReq // The leader is deliberately not reaped before escalation, so its PID/PGID // cannot be recycled while TERM/KILL still targets the recorded group. if groupErr != nil { + diagnostic.fail("child_cleanup") _ = child.Process.Kill() } waitErr := child.Wait() + diagnostic.child(child.ProcessState) if waitErr != nil && protocolErr == nil { + diagnostic.fail("child_exit") protocolErr = errors.New("pipeline child failed") } // Every drain has a bounded termination deadline even if a nonconforming @@ -345,6 +401,7 @@ func executeIsolatedPipelineChild(ctx context.Context, request pipelineRecordReq select { case <-done: case <-time.After(time.Second): + diagnostic.fail("child_pipe") _ = stdoutRead.Close() _ = stderrRead.Close() _ = resultRead.Close() @@ -354,12 +411,14 @@ func executeIsolatedPipelineChild(ctx context.Context, request pipelineRecordReq } _ = inputWrite.Close() if inputErr := <-inputDone; inputErr != nil { + diagnostic.fail("child_input") protocolErr = errors.New("pipeline child input failed") } if !brokerStopped { select { case <-brokerDone: case <-time.After(31 * time.Second): + diagnostic.fail("broker_stop") return nil, errPipelineCleanupIncomplete } } @@ -367,9 +426,11 @@ func executeIsolatedPipelineChild(ctx context.Context, request pipelineRecordReq return nil, errPipelineCleanupIncomplete } if err := waitPipelineProcessGroupEmpty(entry.Child.PGID, time.Second); err != nil { + diagnostic.fail("child_cleanup") return nil, errPipelineCleanupIncomplete } if stdout.err != nil || stderr.err != nil || result.err != nil || stdout.data.Len() != 0 || stdout.overflow || stderr.overflow || result.overflow { + diagnostic.fail("child_quarantine") return nil, errors.New("pipeline child quarantine or cleanup failed") } if protocolErr != nil { @@ -396,43 +457,69 @@ func waitPipelineProcessGroupEmpty(pgid int, timeout time.Duration) error { } func executePipelineBrokerSandbox(ctx context.Context, request pipelineBrokerRequest, state *pipelineCleanupState, client *pipelineDockerClient, entry *pipelineCleanupEntry) (*sandbox.ExecResult, error) { + diagnostic := pipelineRecordDiagnosticFromContext(ctx) + diagnostic.beginOperation("create") name := fmt.Sprintf("%s-%d.cid", entry.Label, len(entry.CIDFiles)) args, err := client.CreateArgs(request.Config, request.Command, entry.Label, filepath.Join(state.root, name)) if err != nil { + diagnostic.fail("container_args") return nil, err } entry.CIDFiles = append(entry.CIDFiles, name) if err := state.Save(*entry); err != nil { + diagnostic.fail("state_save") return nil, err } created, err := client.RunInProcessGroup(ctx, entry.Child, append([]string{"create"}, args...)...) + if created != nil { + diagnostic.dockerExit(created.ExitCode) + } if err != nil || created == nil || created.ExitCode != 0 { + diagnostic.fail("container_create") return nil, errors.New("record sandbox create failed") } id := strings.TrimSpace(created.Stdout) if len(id) != 64 || strings.ToLower(id) != id { + diagnostic.fail("container_id") return nil, errors.New("invalid record sandbox ID") } if _, err := hex.DecodeString(id); err != nil { + diagnostic.fail("container_id") return nil, errors.New("invalid record sandbox ID") } entry.IDs = append(entry.IDs, id) if err := state.Save(*entry); err != nil { + diagnostic.fail("state_save") return nil, err } + diagnostic.beginOperation("start") result, err := client.RunInProcessGroup(ctx, entry.Child, "start", "-a", id) + if result != nil { + diagnostic.dockerExit(result.ExitCode) + } if err != nil || result == nil { + diagnostic.fail("container_attach") return nil, errors.New("record sandbox attach failed") } + diagnostic.beginOperation("wait") wait, err := client.RunInProcessGroup(ctx, entry.Child, "wait", id) + if wait != nil { + diagnostic.dockerExit(wait.ExitCode) + } if err != nil || wait == nil || wait.ExitCode != 0 { + diagnostic.fail("container_wait") return nil, errors.New("record sandbox wait failed") } code, err := strconv.Atoi(strings.TrimSpace(wait.Stdout)) if err != nil || code < 0 || code > 255 { + diagnostic.fail("container_status") return nil, errors.New("invalid record sandbox status") } result.ExitCode = code + diagnostic.containerExit(code) + if code != 0 { + diagnostic.fail("container_command") + } return result, nil } diff --git a/docs/WFCTL.md b/docs/WFCTL.md index 03081a7f..4a37dd62 100644 --- a/docs/WFCTL.md +++ b/docs/WFCTL.md @@ -1376,6 +1376,20 @@ values, and invalid UTF-8 fail. Represent identifiers and counters as decimal strings. Failure emits one generic error record and exits nonzero; its payload is defined in `docs/schemas/wfctl-pipeline-record-v1.schema.json`. +For synthetic record-host troubleshooting, `WFCTL_RECORD_DIAGNOSTICS=1` opts +the parent into one failure-only stderr line prefixed +`WFCTL_RECORD_DIAGNOSTIC_V1`. Its JSON contains fixed phase/event/cleanup +names, bounded child/Docker CLI/container exit fields and context flags. +An unknown exit is `-1`; `docker_operation` names the attempted operation, +and a Docker CLI exit does not establish a container exit. These operation +statuses are frozen at the first observed failure, with later child-wait and +cleanup outcomes reported separately. Success and +the default mode emit no diagnostic line. This opt-in is removed from the +isolated child's environment; public stdout records remain unchanged. +Reports contain no raw errors, workload output, environment values or +resource identities. The live synthetic test logs only validated statuses, +fixture digests and bounded cleanup counts on failure. + Before loading application config, the parent locks private cleanup state at `$XDG_STATE_HOME/wfctl/pipeline-cleanup` (default `~/.local/state/wfctl/pipeline-cleanup`), reconciles retained executions, and From df181c857904c388068effe11defe4cb60fe021f Mon Sep 17 00:00:00 2001 From: Jon Langevin Date: Wed, 14 Oct 2026 19:39:11 -0400 Subject: [PATCH 2/2] test(wfctl): isolate record diagnostic image setup --- cmd/wfctl/pipeline_record_diagnostic_unix_test.go | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/cmd/wfctl/pipeline_record_diagnostic_unix_test.go b/cmd/wfctl/pipeline_record_diagnostic_unix_test.go index 5954c532..3ef48a3f 100644 --- a/cmd/wfctl/pipeline_record_diagnostic_unix_test.go +++ b/cmd/wfctl/pipeline_record_diagnostic_unix_test.go @@ -115,7 +115,7 @@ func pipelineRecordHostExit(err error) int { } var exit *exec.ExitError if errors.As(err, &exit) && exit.ProcessState != nil { - return pipelineRecordDiagnosticExit(exit.ProcessState.ExitCode()) + return pipelineRecordDiagnosticExit(exit.ExitCode()) } return -1 } @@ -538,12 +538,16 @@ func TestPipelineRecordDiagnosticsImageSetupRedaction(t *testing.T) { ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) defer cancel() command := exec.CommandContext(ctx, executable, "-test.run=^TestPipelineRecordDiagnosticsImageSetupHelper$") + command.Dir = directory // This setup path invokes Docker directly, whose test helper records - // its environment. Supply only fixture and runtime settings. + // its environment. It needs only the fixed Docker wrapper, whose + // interpreter and test binary are absolute, and no Go toolchain. command.Env = []string{ - "PATH=" + strings.Join([]string{directory, "/usr/bin", "/bin"}, string(os.PathListSeparator)), + "PATH=" + directory, "HOME=" + directory, "TMPDIR=" + directory, + "GOENV=off", "GOWORK=off", "GOTOOLCHAIN=local", + "GOPROXY=off", "GOSUMDB=off", "GOVCS=*:off", "GORACE=" + os.Getenv("GORACE"), "WFCTL_RECORD_DIAGNOSTIC_IMAGE_SETUP=1", pipelineRecordDiagnosticEnv + "=", @@ -572,6 +576,11 @@ func TestPipelineRecordDiagnosticsImageSetupRedaction(t *testing.T) { t.Fatal("image setup copied an ambient credential into its private capture") } } + for _, relative := range []string{"go/pkg/mod", ".cache/go-build", "Library/Caches/go-build"} { + if _, err := os.Lstat(filepath.Join(directory, filepath.FromSlash(relative))); !os.IsNotExist(err) { + t.Fatal("synthetic image setup did not exclude private Go cache creation") + } + } }) } }