fix(auth): close cancellation and workspace races

This commit is contained in:
2026-08-17 19:32:03 +02:00
parent d12b629420
commit 0a2c667231
10 changed files with 822 additions and 221 deletions
@@ -1,26 +1,50 @@
package compose
import "time"
import (
"errors"
"os"
"time"
)
// terminateWindowsProcess reports a reap failure only when the process has not exited within the
// final bound. taskkill and Process.Kill can both race with a child that has already been reaped.
func terminateWindowsProcess(
// processLifetime owns a started Docker CLI process until it has been reaped and every process
// it may have started has been released. Platform implementations must make close idempotent.
type processLifetime interface {
terminate(done <-chan error) error
close() error
}
var startProcessForRunner = startProcess
var terminateProcessForRunner = func(process processLifetime, done <-chan error) error {
return process.terminate(done)
}
// terminateProcessTree terminates the owned tree before consulting the direct child's reaping
// result. A direct child can be reaped while descendants survive, so a ready done channel is never
// evidence that tree termination can be skipped. The direct kill is only a fallback when tree
// ownership could not terminate the tree; an already-reaped direct child is benign in that case.
func terminateProcessTree(
done <-chan error,
terminateTree func() error,
terminateDirect func() error,
bound time.Duration,
) error {
if processReaped(done) {
return nil
treeErr := terminateTree()
var directErr error
if treeErr != nil {
directErr = terminateDirect()
if errors.Is(directErr, os.ErrProcessDone) {
directErr = nil
}
}
_ = terminateTree()
if processReaped(done) {
return nil
}
_ = terminateDirect()
if processReaped(done) {
reapErr := waitForProcessReap(done, bound)
if treeErr == nil && directErr == nil && reapErr == nil {
return nil
}
return errors.Join(ErrProcessReap, treeErr, directErr, reapErr)
}
func waitForProcessReap(done <-chan error, bound time.Duration) error {
timer := time.NewTimer(bound)
defer timer.Stop()
select {
@@ -30,12 +54,3 @@ func terminateWindowsProcess(
return ErrProcessReap
}
}
func processReaped(done <-chan error) bool {
select {
case <-done:
return true
default:
return false
}
}
@@ -2,88 +2,80 @@ package compose
import (
"errors"
"os"
"testing"
"time"
)
func TestWindowsCancellationTreatsProcessReapedBeforeTreeTerminationAsSuccess(t *testing.T) {
func TestTerminationKillsTheTreeEvenWhenTheDirectChildIsAlreadyReaped(t *testing.T) {
done := make(chan error, 1)
done <- nil
treeFailure := errors.New("tree termination failed")
treeCalls := 0
directCalls := 0
err := terminateWindowsProcess(done, func() error {
err := terminateProcessTree(done, func() error {
treeCalls++
return errors.New("tree termination should not run")
return treeFailure
}, func() error {
directCalls++
return errors.New("direct termination should not run")
}, 0)
return os.ErrProcessDone
}, time.Second)
if err != nil {
t.Fatalf("terminateWindowsProcess() error = %v, want nil", err)
if treeCalls != 1 {
t.Fatalf("tree termination calls = %d, want 1", treeCalls)
}
if treeCalls != 0 || directCalls != 0 {
t.Fatalf("termination calls = tree:%d direct:%d, want none", treeCalls, directCalls)
if directCalls != 1 {
t.Fatalf("direct termination calls = %d, want 1 fallback after tree failure", directCalls)
}
if !errors.Is(err, treeFailure) || !errors.Is(err, ErrProcessReap) {
t.Fatalf("terminateProcessTree() error = %v, want tree failure and ErrProcessReap", err)
}
if errors.Is(err, os.ErrProcessDone) {
t.Fatalf("terminateProcessTree() error = %v, must treat an already-reaped direct child as benign", err)
}
}
func TestWindowsCancellationTreatsProcessReapedAfterTreeTerminationAsSuccess(t *testing.T) {
func TestTerminationWaitsForTheOriginalProcessAfterTreeTermination(t *testing.T) {
done := make(chan error, 1)
done <- nil
directCalls := 0
err := terminateWindowsProcess(done, func() error {
done <- nil
return ErrProcessReap
}, func() error {
err := terminateProcessTree(done, func() error { return nil }, func() error {
directCalls++
return errors.New("direct termination should not run")
}, 0)
return errors.New("direct fallback should not run after a successful tree termination")
}, time.Second)
if err != nil {
t.Fatalf("terminateWindowsProcess() error = %v, want nil", err)
t.Fatalf("terminateProcessTree() error = %v, want nil", err)
}
if directCalls != 0 {
t.Fatalf("direct termination calls = %d, want 0", directCalls)
}
}
func TestWindowsCancellationTreatsProcessReapedAfterDirectTerminationAsSuccess(t *testing.T) {
done := make(chan error, 1)
directCalls := 0
func TestTerminationJoinsTreeDirectAndReapFailures(t *testing.T) {
done := make(chan error)
treeFailure := errors.New("tree termination failed")
directFailure := errors.New("direct termination failed")
err := terminateWindowsProcess(done, func() error {
return ErrProcessReap
}, func() error {
directCalls++
done <- nil
return ErrProcessReap
}, 0)
err := terminateProcessTree(done, func() error { return treeFailure }, func() error { return directFailure }, 0)
if err != nil {
t.Fatalf("terminateWindowsProcess() error = %v, want nil", err)
}
if directCalls != 1 {
t.Fatalf("direct termination calls = %d, want 1", directCalls)
for _, want := range []error{treeFailure, directFailure, ErrProcessReap} {
if !errors.Is(err, want) {
t.Fatalf("terminateProcessTree() error = %v, want %v", err, want)
}
}
}
func TestWindowsCancellationReportsReapFailureWhenProcessSurvives(t *testing.T) {
func TestTerminationReportsReapFailureWhenTheTreeTerminatesButTheProcessNeverReaps(t *testing.T) {
done := make(chan error)
treeCalls := 0
directCalls := 0
err := terminateWindowsProcess(done, func() error {
treeCalls++
return ErrProcessReap
}, func() error {
directCalls++
return ErrProcessReap
err := terminateProcessTree(done, func() error { return nil }, func() error {
return errors.New("direct fallback should not run")
}, 0)
if !errors.Is(err, ErrProcessReap) {
t.Fatalf("terminateWindowsProcess() error = %v, want ErrProcessReap", err)
}
if treeCalls != 1 || directCalls != 1 {
t.Fatalf("termination calls = tree:%d direct:%d, want one each", treeCalls, directCalls)
t.Fatalf("terminateProcessTree() error = %v, want ErrProcessReap", err)
}
}
@@ -15,6 +15,25 @@ func configureProcess(command *exec.Cmd) {
command.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
}
type unixProcessLifetime struct {
command *exec.Cmd
}
func startProcess(command *exec.Cmd) (processLifetime, error) {
if err := command.Start(); err != nil {
return nil, err
}
return unixProcessLifetime{command: command}, nil
}
func (process unixProcessLifetime) terminate(done <-chan error) error {
return terminateProcess(process.command, done)
}
func (unixProcessLifetime) close() error {
return nil
}
func terminateProcess(command *exec.Cmd, done <-chan error) error {
if command.Process == nil {
return nil
+150 -44
View File
@@ -3,75 +3,181 @@
package compose
import (
"context"
"io"
"errors"
"os"
"os/exec"
"path/filepath"
"strconv"
"syscall"
"time"
"unsafe"
"golang.org/x/sys/windows"
)
const finalTerminationBound = 2 * time.Second
const processTreeTerminationBound = 2 * time.Second
const createNewProcessGroup = 0x00000200
var windowsSystemDirectory = windows.GetSystemDirectory
const createSuspendedProcess = 0x00000004
const jobTerminationExitCode = 1
func configureProcess(command *exec.Cmd) {
command.SysProcAttr = &syscall.SysProcAttr{CreationFlags: createNewProcessGroup}
command.SysProcAttr = &syscall.SysProcAttr{
CreationFlags: createNewProcessGroup | createSuspendedProcess,
}
}
func terminateProcess(command *exec.Cmd, done <-chan error) error {
if command.Process == nil {
type windowsProcessJob struct {
handle windows.Handle
}
func newWindowsProcessJob() (*windowsProcessJob, error) {
handle, err := windows.CreateJobObject(nil, nil)
if err != nil {
return nil, err
}
limits := windows.JOBOBJECT_EXTENDED_LIMIT_INFORMATION{}
limits.BasicLimitInformation.LimitFlags = windows.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE
if _, err := windows.SetInformationJobObject(
handle,
windows.JobObjectExtendedLimitInformation,
uintptr(unsafe.Pointer(&limits)),
uint32(unsafe.Sizeof(limits)),
); err != nil {
return nil, errors.Join(err, windows.CloseHandle(handle))
}
return &windowsProcessJob{handle: handle}, nil
}
func (job *windowsProcessJob) close() error {
if job == nil || job.handle == 0 {
return nil
}
return terminateWindowsProcess(
handle := job.handle
job.handle = 0
return windows.CloseHandle(handle)
}
type windowsProcessLifetime struct {
command *exec.Cmd
job *windowsProcessJob
}
// startProcess creates the Docker process suspended, assigns it to a kill-on-close Job Object,
// then resumes its one primary thread. The suspension closes the startup window in which a direct
// child could otherwise create descendants before the Job Object begins to own them.
func startProcess(command *exec.Cmd) (processLifetime, error) {
job, err := newWindowsProcessJob()
if err != nil {
return nil, err
}
if err := command.Start(); err != nil {
return nil, errors.Join(err, job.close())
}
if command.Process == nil {
return nil, errors.Join(ErrProcessReap, abortWindowsStart(command, job))
}
if err := assignWindowsProcessToJob(job, command.Process.Pid); err != nil {
return nil, errors.Join(err, abortWindowsStart(command, job))
}
if err := resumeWindowsProcess(command.Process.Pid); err != nil {
return nil, errors.Join(err, abortWindowsStart(command, job))
}
return &windowsProcessLifetime{command: command, job: job}, nil
}
func (process *windowsProcessLifetime) terminate(done <-chan error) error {
if process == nil || process.command == nil || process.command.Process == nil || process.job == nil || process.job.handle == 0 {
return ErrProcessReap
}
return terminateProcessTree(
done,
func() error { return terminateWindowsProcessTree(command.Process.Pid) },
command.Process.Kill,
func() error { return windows.TerminateJobObject(process.job.handle, jobTerminationExitCode) },
process.command.Process.Kill,
finalTerminationBound,
)
}
func terminateWindowsProcessTree(pid int) error {
ctx, cancel := context.WithTimeout(context.Background(), processTreeTerminationBound)
defer cancel()
taskkillPath, err := systemTaskkillPath()
if err != nil {
return ErrProcessReap
func (process *windowsProcessLifetime) close() error {
if process == nil {
return nil
}
command := windowsTreeKillCommand(ctx, taskkillPath, pid)
command.Stdout = io.Discard
command.Stderr = io.Discard
if err := command.Run(); err != nil {
return ErrProcessReap
}
if ctx.Err() != nil {
return ErrProcessReap
}
return nil
return process.job.close()
}
// systemTaskkillPath resolves taskkill through Windows' protected system-directory API, never PATH.
func systemTaskkillPath() (string, error) {
systemDirectory, err := windowsSystemDirectory()
func assignWindowsProcessToJob(job *windowsProcessJob, pid int) error {
process, err := windows.OpenProcess(windows.PROCESS_SET_QUOTA|windows.PROCESS_TERMINATE, false, uint32(pid))
if err != nil {
return "", ErrProcessReap
return err
}
systemDirectory = filepath.Clean(systemDirectory)
if systemDirectory == "." || !filepath.IsAbs(systemDirectory) {
return "", ErrProcessReap
}
taskkillPath := filepath.Join(systemDirectory, "taskkill.exe")
if !filepath.IsAbs(taskkillPath) || filepath.Dir(taskkillPath) != systemDirectory {
return "", ErrProcessReap
}
return taskkillPath, nil
return errors.Join(
windows.AssignProcessToJobObject(job.handle, process),
windows.CloseHandle(process),
)
}
func windowsTreeKillCommand(ctx context.Context, taskkillPath string, pid int) *exec.Cmd {
return exec.CommandContext(ctx, taskkillPath, "/PID", strconv.Itoa(pid), "/T", "/F")
func resumeWindowsProcess(pid int) error {
snapshot, err := windows.CreateToolhelp32Snapshot(windows.TH32CS_SNAPTHREAD, 0)
if err != nil {
return err
}
threadID, threadErr := suspendedPrimaryThreadID(snapshot, uint32(pid))
closeSnapshotErr := windows.CloseHandle(snapshot)
if threadErr != nil || closeSnapshotErr != nil {
return errors.Join(threadErr, closeSnapshotErr)
}
thread, err := windows.OpenThread(windows.THREAD_SUSPEND_RESUME, false, threadID)
if err != nil {
return err
}
_, resumeErr := windows.ResumeThread(thread)
closeThreadErr := windows.CloseHandle(thread)
return errors.Join(resumeErr, closeThreadErr)
}
func suspendedPrimaryThreadID(snapshot windows.Handle, pid uint32) (uint32, error) {
entry := windows.ThreadEntry32{Size: uint32(unsafe.Sizeof(windows.ThreadEntry32{}))}
if err := windows.Thread32First(snapshot, &entry); err != nil {
return 0, err
}
threadIDs := make([]uint32, 0, 1)
for {
if entry.OwnerProcessID == pid {
threadIDs = append(threadIDs, entry.ThreadID)
}
entry.Size = uint32(unsafe.Sizeof(windows.ThreadEntry32{}))
err := windows.Thread32Next(snapshot, &entry)
if errors.Is(err, windows.ERROR_NO_MORE_FILES) {
break
}
if err != nil {
return 0, err
}
}
if len(threadIDs) != 1 {
return 0, ErrProcessReap
}
return threadIDs[0], nil
}
// abortWindowsStart runs only while the direct process is still suspended or immediately after
// it was resumed unsuccessfully. It waits exactly once, so the normal runner path never races a
// second Wait or leaves a waiter goroutine behind.
func abortWindowsStart(command *exec.Cmd, job *windowsProcessJob) error {
var cleanupErr error
if job != nil && job.handle != 0 {
cleanupErr = errors.Join(cleanupErr, windows.TerminateJobObject(job.handle, jobTerminationExitCode))
}
if command.Process != nil {
if err := command.Process.Kill(); err != nil && !errors.Is(err, os.ErrProcessDone) {
cleanupErr = errors.Join(cleanupErr, err)
}
if err := command.Wait(); err != nil {
var exitError *exec.ExitError
if !errors.As(err, &exitError) {
cleanupErr = errors.Join(cleanupErr, err)
}
}
}
if job != nil {
cleanupErr = errors.Join(cleanupErr, job.close())
}
return cleanupErr
}
@@ -6,28 +6,29 @@ import (
"testing"
)
func TestWindowsTerminationUsesBoundedExactPIDTreeKillWithoutAShell(t *testing.T) {
func TestWindowsTerminationOwnsTheProcessTreeWithAKillOnCloseJobObjectWithoutAShell(t *testing.T) {
source, err := os.ReadFile("process_windows.go")
if err != nil {
t.Fatal(err)
}
text := string(source)
for _, required := range []string{
"context.WithTimeout", "windows.GetSystemDirectory", "filepath.Clean", "filepath.IsAbs",
`filepath.Join(systemDirectory, "taskkill.exe")`, "systemTaskkillPath",
`exec.CommandContext(ctx, taskkillPath, "/PID", strconv.Itoa(pid), "/T", "/F")`,
"io.Discard", "ErrProcessReap",
"windows.CreateJobObject", "windows.JOBOBJECT_EXTENDED_LIMIT_INFORMATION",
"windows.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE", "windows.SetInformationJobObject",
"windows.AssignProcessToJobObject", "windows.TerminateJobObject", "windows.OpenProcess",
"windows.CreateToolhelp32Snapshot", "windows.ResumeThread", "windows.CloseHandle",
"createSuspendedProcess",
} {
if !strings.Contains(text, required) {
t.Errorf("process_windows.go is missing %q", required)
}
}
for _, forbidden := range []string{
"cmd.exe", "powershell", `exec.Command("taskkill.exe"`,
`exec.CommandContext(ctx, "taskkill.exe"`, `exec.LookPath("taskkill.exe")`,
"taskkill", "cmd.exe", "powershell", "exec.Command(", "exec.CommandContext(",
"exec.LookPath(", "windows.GetSystemDirectory",
} {
if strings.Contains(strings.ToLower(text), strings.ToLower(forbidden)) {
t.Errorf("process_windows.go contains unsafe command form %q", forbidden)
t.Errorf("process_windows.go contains unsafe or unowned process-tree mechanism %q", forbidden)
}
}
}
@@ -3,38 +3,44 @@
package compose
import (
"context"
"errors"
"path/filepath"
"reflect"
"os/exec"
"testing"
"unsafe"
"golang.org/x/sys/windows"
)
func TestWindowsTreeKillCommandUsesTrustedAbsoluteSystemPathAndExactPIDArgumentArray(t *testing.T) {
original := windowsSystemDirectory
windowsSystemDirectory = func() (string, error) { return `C:\\Windows\\System32`, nil }
t.Cleanup(func() { windowsSystemDirectory = original })
taskkillPath, err := systemTaskkillPath()
func TestWindowsProcessJobUsesKillOnClose(t *testing.T) {
job, err := newWindowsProcessJob()
if err != nil {
t.Fatal(err)
}
if !filepath.IsAbs(taskkillPath) {
t.Fatalf("taskkill path = %q, want absolute trusted system path", taskkillPath)
t.Cleanup(func() {
if closeErr := job.close(); closeErr != nil {
t.Error(closeErr)
}
})
var limits windows.JOBOBJECT_EXTENDED_LIMIT_INFORMATION
if err := windows.QueryInformationJobObject(
job.handle,
windows.JobObjectExtendedLimitInformation,
uintptr(unsafe.Pointer(&limits)),
uint32(unsafe.Sizeof(limits)),
nil,
); err != nil {
t.Fatal(err)
}
command := windowsTreeKillCommand(context.Background(), taskkillPath, 4242)
want := []string{taskkillPath, "/PID", "4242", "/T", "/F"}
if !reflect.DeepEqual(command.Args, want) {
t.Fatalf("taskkill args = %#v, want %#v", command.Args, want)
if limits.BasicLimitInformation.LimitFlags&windows.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE == 0 {
t.Fatalf("job limits = %#x, want JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE", limits.BasicLimitInformation.LimitFlags)
}
}
func TestWindowsSystemTaskkillPathRejectsRelativeSystemDirectory(t *testing.T) {
original := windowsSystemDirectory
windowsSystemDirectory = func() (string, error) { return `System32`, nil }
t.Cleanup(func() { windowsSystemDirectory = original })
func TestWindowsProcessConfigurationStartsSuspendedBeforeJobAssignment(t *testing.T) {
command := exec.Command("docker")
configureProcess(command)
if _, err := systemTaskkillPath(); !errors.Is(err, ErrProcessReap) {
t.Fatalf("systemTaskkillPath() error = %v, want ErrProcessReap", err)
if command.SysProcAttr == nil || command.SysProcAttr.CreationFlags&createSuspendedProcess == 0 {
t.Fatalf("configureProcess() flags = %#x, want CREATE_SUSPENDED", command.SysProcAttr.CreationFlags)
}
}
+9 -7
View File
@@ -31,8 +31,6 @@ var ErrProcessReap = errors.New("Docker command process could not be reaped")
// ErrContainerCleanup reports that Docker could not confirm removal of a named one-shot container.
var ErrContainerCleanup = errors.New("Docker one-shot container cleanup failed")
var terminateProcessForRunner = terminateProcess
// CaptureLimits bounds each captured stream while the child is running.
type CaptureLimits struct {
StdoutBytes int
@@ -115,7 +113,8 @@ func (r execRunner) runBounded(ctx context.Context, args []string, stdin io.Read
stderr := newCappedBuffer(limits.StderrBytes, overflow, stderrObserver)
command.Stdout = stdout
command.Stderr = stderr
if err := command.Start(); err != nil {
process, err := startProcessForRunner(command)
if err != nil {
return startFailure(err)
}
done := make(chan error, 1)
@@ -128,16 +127,19 @@ func (r execRunner) runBounded(ctx context.Context, args []string, stdin io.Read
case <-ctx.Done():
interrupted = true
processErr = ctx.Err()
if err := terminateProcessForRunner(command, done); err != nil {
lifecycleErr = errors.Join(lifecycleErr, ErrProcessReap)
if err := terminateProcessForRunner(process, done); err != nil {
lifecycleErr = errors.Join(lifecycleErr, err)
}
case <-overflow:
interrupted = true
processErr = ErrOutputLimit
if err := terminateProcessForRunner(command, done); err != nil {
lifecycleErr = errors.Join(lifecycleErr, ErrProcessReap)
if err := terminateProcessForRunner(process, done); err != nil {
lifecycleErr = errors.Join(lifecycleErr, err)
}
}
if err := process.close(); err != nil {
lifecycleErr = errors.Join(lifecycleErr, ErrProcessReap, err)
}
if containerName != "" {
if err := r.cleanupOneShotContainer(containerName); err != nil {
lifecycleErr = errors.Join(lifecycleErr, ErrContainerCleanup)
+2 -2
View File
@@ -245,8 +245,8 @@ func TestRunnerSurfacesCleanupFailureWithoutContainerName(t *testing.T) {
func TestRunnerPropagatesAReapingFailure(t *testing.T) {
original := terminateProcessForRunner
terminateProcessForRunner = func(command *exec.Cmd, done <-chan error) error {
_ = terminateProcess(command, done)
terminateProcessForRunner = func(process processLifetime, done <-chan error) error {
_ = process.terminate(done)
return ErrProcessReap
}
t.Cleanup(func() { terminateProcessForRunner = original })