From 9daeb947eea41a56bf99100dafdc036587d1bc7b Mon Sep 17 00:00:00 2001 From: Charles Wu Date: Tue, 18 Aug 2026 14:58:50 +1000 Subject: [PATCH 1/5] refactor: gather the worker handoff into pkg/utils The collector, scanner and remover hand images off through four named pipes in a shared volume, but only two of the eight operations lived in pkg/utils. The rest were written out inline across three binaries, so the protocol could not be reasoned about, tested, or reimplemented in one place. Move all of them behind CreatePipe, WriteImagesPipe, ReadImagesPipe, WriteCompletionPipe and ReadCompletionPipe. ReadCollectScanPipe and WriteScanErasePipe remain as thin wrappers, since custom scanners may call them directly. No behaviour change: the same syscalls run in the same order, callers keep their own logging and exit codes, and the payload validation stays with the caller so existing handling of unexpected content is preserved. Two log messages differ slightly where an error path was merged. A side effect worth noting: because the FIFO calls now route through the per-GOOS mkfifo shim, pkg/utils, pkg/collector and pkg/scanners/template compile for GOOS=windows for the first time. They do not yet *work* there -- mkfifo returns ErrFifoUnsupported -- so Windows images should not be published until the transport lands. Only the manager still fails to build for Windows, on inotify, and it is Linux-only by design. Signed-off-by: Charles Wu --- pkg/collector/collector.go | 44 +------ pkg/remover/remover.go | 62 +-------- pkg/scanners/template/scanner_template.go | 23 ++-- pkg/utils/handoff.go | 150 ++++++++++++++++++++++ pkg/utils/utils.go | 66 +--------- 5 files changed, 175 insertions(+), 170 deletions(-) create mode 100644 pkg/utils/handoff.go diff --git a/pkg/collector/collector.go b/pkg/collector/collector.go index cd0b012e1b..018bedc065 100644 --- a/pkg/collector/collector.go +++ b/pkg/collector/collector.go @@ -1,10 +1,8 @@ package main import ( - "encoding/json" "flag" "fmt" - "io" "net/http" _ "net/http/pprof" "os" @@ -12,7 +10,6 @@ import ( "github.com/eraser-dev/eraser/pkg/cri" "github.com/eraser-dev/eraser/pkg/logger" - "golang.org/x/sys/unix" logf "sigs.k8s.io/controller-runtime/pkg/log" util "github.com/eraser-dev/eraser/pkg/utils" @@ -73,61 +70,30 @@ func main() { } log.Info("images collected", "finalImages:", finalImages) - data, err := json.Marshal(finalImages) - if err != nil { - log.Error(err, "failed to encode finalImages") - os.Exit(1) - } - path := util.CollectScanPath if *scanDisabled { path = util.ScanErasePath } - if err := unix.Mkfifo(path, util.PipeMode); err != nil { - log.Error(err, "failed to create pipe", "pipeFile", path) + if err := util.WriteImagesPipe(path, finalImages); err != nil { + log.Error(err, "failed to send images", "pipeFile", path) os.Exit(1) } - //nolint:gosec // G304: Opening pipe file is intended functionality - file, err := os.OpenFile(path, os.O_WRONLY, 0) + completion, err := util.CreateCompletionPipe(util.EraseCompleteCollectPath) if err != nil { - log.Error(err, "failed to open pipe", "pipeFile", path) - os.Exit(1) - } - - if _, err := file.Write(data); err != nil { - log.Error(err, "failed to write to pipe", "pipeFile", path) - os.Exit(1) - } - - if err := file.Close(); err != nil { - log.Error(err, "failed to close pipe", "pipeFile", path) - os.Exit(1) - } - if err := unix.Mkfifo(util.EraseCompleteCollectPath, util.PipeMode); err != nil { log.Error(err, "failed to create pipe", "pipeFile", util.EraseCompleteCollectPath) os.Exit(1) } + defer completion.Close() - file, err = os.OpenFile(util.EraseCompleteCollectPath, os.O_RDONLY, 0) - if err != nil { - log.Error(err, "failed to open pipe", "pipeFile", util.EraseCompleteCollectPath) - os.Exit(1) - } - - data, err = io.ReadAll(file) + data, err := completion.Await() if err != nil { log.Error(err, "failed to read pipe", "pipeFile", util.EraseCompleteCollectPath) os.Exit(1) } - if err := file.Close(); err != nil { - log.Error(err, "failed to close pipe", "pipeFile", util.EraseCompleteCollectPath) - os.Exit(1) - } - if string(data) != util.EraseCompleteMessage { log.Info("garbage in pipe", "pipeFile", util.EraseCompleteCollectPath, "in_pipe", string(data)) os.Exit(1) diff --git a/pkg/remover/remover.go b/pkg/remover/remover.go index 4f714ad1f6..5db5bb2dd5 100644 --- a/pkg/remover/remover.go +++ b/pkg/remover/remover.go @@ -2,11 +2,8 @@ package main import ( "context" - "encoding/json" "flag" "fmt" - "io" - "io/fs" "net/http" _ "net/http/pprof" "os" @@ -21,7 +18,6 @@ import ( "github.com/eraser-dev/eraser/pkg/logger" "github.com/eraser-dev/eraser/pkg/metrics" - "github.com/eraser-dev/eraser/api/unversioned" util "github.com/eraser-dev/eraser/pkg/utils" ) @@ -78,38 +74,11 @@ func main() { } if *imageListPtr == "" { - var f *os.File - for { - var err error - - f, err = os.OpenFile(util.ScanErasePath, os.O_RDONLY, 0) - if err == nil { - break - } - if !os.IsNotExist(err) { - log.Error(err, "error opening scanErase pipe") - os.Exit(generalErr) - } - time.Sleep(1 * time.Second) - continue - } - - // json data is list of []unversioned.Image - data, err := io.ReadAll(f) + nonCompliantImages, err := util.ReadImagesPipe(context.Background(), util.ScanErasePath) if err != nil { log.Error(err, "error reading non-compliant images") os.Exit(generalErr) } - if err := f.Close(); err != nil { - log.Error(err, "error closing non-compliant images file") - os.Exit(generalErr) - } - - nonCompliantImages := []unversioned.Image{} - if err = json.Unmarshal(data, &nonCompliantImages); err != nil { - log.Error(err, "error in unmarshal non-compliant images") - os.Exit(generalErr) - } for _, img := range nonCompliantImages { imagelist = append(imagelist, img.ImageID) @@ -158,39 +127,18 @@ func main() { } if *imageListPtr == "" { - file, err := os.OpenFile(util.EraseCompleteCollectPath, os.O_WRONLY, 0) - if err != nil { - log.Error(err, "unable to open pipe", "pipeFile", util.EraseCompleteCollectPath) + if err := util.WriteCompletionPipe(util.EraseCompleteCollectPath); err != nil { + log.Error(err, "unable to signal completion", "pipeFile", util.EraseCompleteCollectPath) os.Exit(generalErr) } - if _, err := file.WriteString(util.EraseCompleteMessage); err != nil { - log.Error(err, "unable to write to pipe", "pipeFile", util.EraseCompleteCollectPath) - os.Exit(generalErr) - } - - if err := file.Close(); err != nil { - log.Error(err, "unable to close pipe", "pipeFile", util.EraseCompleteCollectPath) - os.Exit(generalErr) - } - - file, err = os.OpenFile(util.EraseCompleteScanPath, os.O_WRONLY, fs.ModeNamedPipe) + err := util.WriteCompletionPipe(util.EraseCompleteScanPath) // if the scanner is disabled if os.IsNotExist(err) { return } if err != nil { - log.Error(err, "unable to open pipe", "pipeFile", util.EraseCompleteCollectPath) - os.Exit(generalErr) - } - - if _, err := file.WriteString(util.EraseCompleteMessage); err != nil { - log.Error(err, "unable to write to pipe", "pipeFile", util.EraseCompleteCollectPath) - os.Exit(generalErr) - } - - if err := file.Close(); err != nil { - log.Error(err, "unable to close pipe", "pipeFile", util.EraseCompleteScanPath) + log.Error(err, "unable to signal completion", "pipeFile", util.EraseCompleteScanPath) os.Exit(generalErr) } } diff --git a/pkg/scanners/template/scanner_template.go b/pkg/scanners/template/scanner_template.go index f00da5d05d..2d51442891 100644 --- a/pkg/scanners/template/scanner_template.go +++ b/pkg/scanners/template/scanner_template.go @@ -2,14 +2,12 @@ package template import ( "context" - "io" "os" "os/signal" "syscall" "github.com/eraser-dev/eraser/api/unversioned" "github.com/go-logr/logr" - "golang.org/x/sys/unix" "github.com/eraser-dev/eraser/pkg/metrics" util "github.com/eraser-dev/eraser/pkg/utils" @@ -35,6 +33,10 @@ type config struct { deleteScanFailedImages bool deleteEOLImages bool reportMetrics bool + + // held from ReceiveImages until Finish: the endpoint must stay published so + // the remover can tell a scanner is present + completion *util.CompletionPipe } type ConfigFunc func(*config) @@ -59,7 +61,9 @@ func NewImageProvider(funcs ...ConfigFunc) ImageProvider { func (cfg *config) ReceiveImages() ([]unversioned.Image, error) { var err error - if err := unix.Mkfifo(util.EraseCompleteScanPath, util.PipeMode); err != nil { + // published up front so the remover can tell a scanner is present + cfg.completion, err = util.CreateCompletionPipe(util.EraseCompleteScanPath) + if err != nil { cfg.log.Error(err, "failed to create pipe", "pipeName", util.EraseCompleteScanPath) return nil, err } @@ -107,23 +111,14 @@ func (cfg *config) SendImages(nonCompliantImages, failedImages []unversioned.Ima } func (cfg *config) Finish() error { - file, err := os.OpenFile(util.EraseCompleteScanPath, os.O_RDONLY, 0) - if err != nil { - cfg.log.Error(err, "failed to open pipe", "pipeName", util.EraseCompleteScanPath) - return err - } + defer cfg.completion.Close() - data, err := io.ReadAll(file) + data, err := cfg.completion.Await() if err != nil { cfg.log.Error(err, "failed to read pipe", "pipeName", util.EraseCompleteScanPath) return err } - if err := file.Close(); err != nil { - cfg.log.Error(err, "failed to close pipe", "pipeName", util.EraseCompleteScanPath) - return err - } - if string(data) != util.EraseCompleteMessage { cfg.log.Info("garbage in pipe", "pipeName", util.EraseCompleteScanPath, "in_pipe", string(data)) return err diff --git a/pkg/utils/handoff.go b/pkg/utils/handoff.go new file mode 100644 index 0000000000..c5d936b9f6 --- /dev/null +++ b/pkg/utils/handoff.go @@ -0,0 +1,150 @@ +package utils + +import ( + "context" + "encoding/json" + "io" + "os" + "time" + + "github.com/eraser-dev/eraser/api/unversioned" +) + +// The collector, scanner and remover hand images off to each other through +// endpoints in a shared volume. Each operation below was previously written out +// inline in the three worker binaries; they are gathered here so the transport +// can be swapped per platform without touching the callers. +// +// Three properties are load-bearing and must survive any reimplementation. +// Connecting blocks until the peer is on the other end, which is what makes +// arbitrary container start order safe. The reader sees EOF when the writer +// finishes, which is what frames a message. And the absence of an endpoint is +// itself meaningful: the remover infers that the scanner is disabled from +// ENOENT on the scanner's completion endpoint. + +// CompletionPipe is an endpoint a peer can observe before anything is read from +// it. The scanner creates one early precisely so the remover can tell a scanner +// is present. +type CompletionPipe struct { + path string +} + +// CreateCompletionPipe publishes the endpoint without waiting for a peer. +func CreateCompletionPipe(path string) (*CompletionPipe, error) { + if err := mkfifo(path, PipeMode); err != nil { + return nil, err + } + return &CompletionPipe{path: path}, nil +} + +// Await blocks until a peer signals completion. The payload is returned +// unvalidated so callers keep their existing handling of unexpected content. +func (p *CompletionPipe) Await() ([]byte, error) { + //nolint:gosec // G304: Opening pipe file is intended functionality + file, err := os.OpenFile(p.path, os.O_RDONLY, 0) + if err != nil { + return nil, err + } + + data, err := io.ReadAll(file) + if err != nil { + return nil, err + } + + if err := file.Close(); err != nil { + return nil, err + } + + return data, nil +} + +// Close releases the endpoint. It is a no-op where the endpoint is a plain +// filesystem object. +func (p *CompletionPipe) Close() error { + return nil +} + +// WriteImagesPipe publishes the endpoint and blocks until the reader connects. +func WriteImagesPipe(path string, images []unversioned.Image) error { + data, err := json.Marshal(images) + if err != nil { + return err + } + + if err := mkfifo(path, PipeMode); err != nil { + return err + } + + //nolint:gosec // G304: Opening pipe file is intended functionality + file, err := os.OpenFile(path, os.O_WRONLY, 0) + if err != nil { + return err + } + + if _, err := file.Write(data); err != nil { + return err + } + + return file.Close() +} + +// ReadImagesPipe waits for the endpoint to appear, then reads until the writer +// finishes. It returns ctx.Err() if the context is cancelled while waiting. +func ReadImagesPipe(ctx context.Context, path string) ([]unversioned.Image, error) { + timer := time.NewTimer(time.Second) + if !timer.Stop() { + <-timer.C + } + defer timer.Stop() + + var f *os.File + for { + var err error + + //nolint:gosec // G304: Opening pipe file is intended functionality + f, err = os.OpenFile(path, os.O_RDONLY, 0) + if err == nil { + break + } + if !os.IsNotExist(err) { + return nil, err + } + + timer.Reset(time.Second) + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-timer.C: + continue + } + } + + data, err := io.ReadAll(f) + if err != nil { + return nil, err + } + + images := []unversioned.Image{} + if err := json.Unmarshal(data, &images); err != nil { + return nil, err + } + + return images, nil +} + +// WriteCompletionPipe signals a peer that this stage is done. The returned error +// satisfies os.IsNotExist when the peer never published the endpoint, which is +// how an absent scanner is detected. +func WriteCompletionPipe(path string) error { + //nolint:gosec // G304: Opening pipe file is intended functionality + file, err := os.OpenFile(path, os.O_WRONLY, 0) + if err != nil { + return err + } + + if _, err := file.WriteString(EraseCompleteMessage); err != nil { + return err + } + + return file.Close() +} \ No newline at end of file diff --git a/pkg/utils/utils.go b/pkg/utils/utils.go index be8fb7f158..f3d4521738 100644 --- a/pkg/utils/utils.go +++ b/pkg/utils/utils.go @@ -5,12 +5,10 @@ import ( "encoding/json" "errors" "fmt" - "io" "net" "net/url" "os" "strings" - "time" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" @@ -329,68 +327,16 @@ func readConfigMap(path string) ([]string, error) { return images, nil } +// ReadCollectScanPipe is the scanner-facing spelling of ReadImagesPipe, kept +// because custom scanners may call it directly. func ReadCollectScanPipe(ctx context.Context) ([]unversioned.Image, error) { - timer := time.NewTimer(time.Second) - if !timer.Stop() { - <-timer.C - } - defer timer.Stop() - - var f *os.File - for { - var err error - - f, err = os.OpenFile(CollectScanPath, os.O_RDONLY, 0) - if err == nil { - break - } - if !os.IsNotExist(err) { - return nil, err - } - - timer.Reset(time.Second) - select { - case <-ctx.Done(): - return nil, ctx.Err() - case <-timer.C: - continue - } - } - - // json data is list of []eraserv1.Image - data, err := io.ReadAll(f) - if err != nil { - return nil, err - } - - allImages := []unversioned.Image{} - if err = json.Unmarshal(data, &allImages); err != nil { - return nil, err - } - - return allImages, nil + return ReadImagesPipe(ctx, CollectScanPath) } +// WriteScanErasePipe is the scanner-facing spelling of WriteImagesPipe, kept +// because custom scanners may call it directly. func WriteScanErasePipe(vulnerableImages []unversioned.Image) error { - data, err := json.Marshal(vulnerableImages) - if err != nil { - return err - } - - if err = mkfifo(ScanErasePath, PipeMode); err != nil { - return err - } - - file, err := os.OpenFile(ScanErasePath, os.O_WRONLY, 0) - if err != nil { - return err - } - - if _, err := file.Write(data); err != nil { - return err - } - - return file.Close() + return WriteImagesPipe(ScanErasePath, vulnerableImages) } func ProcessRepoDigests(repoDigests []string) ([]string, []error) { From ed03cad6482d331f7e0f4c2ae41d5864535cbb98 Mon Sep 17 00:00:00 2001 From: Charles Wu Date: Tue, 18 Aug 2026 15:28:22 +1000 Subject: [PATCH 2/5] feat: run the worker handoff over Unix sockets on Windows Windows has no filesystem FIFO, so pkg/collector, pkg/utils and pkg/scanners/template built for GOOS=windows but died on the first mkfifo. Implement the handoff over AF_UNIX instead, which Windows has supported since Server 2019 and which Go's net package exposes there. Linux is untouched: handoff.go moves to handoff_unix.go behind a build tag, byte for byte. Unix sockets were chosen over Windows named pipes because they need no new dependency, the endpoint lives in the pod's own volume rather than a machine-global namespace, and the endpoint stays a file -- so the remover can still infer that the scanner is disabled from its absence. Responsibility inverts between the two implementations: with a FIFO the writer creates the endpoint and the reader polls for it, while with a socket the reader listens and the writer dials. The exported API hides that, so callers are unchanged. Two Windows behaviours worth recording, both found by testing rather than by reading docs: Dialing a socket that does not exist reports WSAECONNREFUSED, not ENOENT, and WSAECONNREFUSED does not match syscall.ECONNREFUSED. Neither os.IsNotExist nor errors.Is(fs.ErrNotExist) matches it. So "did the peer ever publish this endpoint?" is answered with os.Stat, and the writer's retry loop does not classify errors at all. A dial succeeds as soon as a listener exists, even with no Accept pending, so the writer is not blocked as it would be on a FIFO. The handoff still completes in order because the reader gets the buffered payload and EOF on close. The new tests are deliberately build-tag free: they exercise whichever implementation the platform selects, so the two cannot drift apart. Signed-off-by: Charles Wu --- pkg/collector/collector.go | 6 +- pkg/scanners/template/scanner_template.go | 2 +- pkg/utils/handoff_test.go | 108 +++++++++++++ pkg/utils/{handoff.go => handoff_unix.go} | 11 +- pkg/utils/handoff_windows.go | 189 ++++++++++++++++++++++ 5 files changed, 309 insertions(+), 7 deletions(-) create mode 100644 pkg/utils/handoff_test.go rename pkg/utils/{handoff.go => handoff_unix.go} (92%) create mode 100644 pkg/utils/handoff_windows.go diff --git a/pkg/collector/collector.go b/pkg/collector/collector.go index 018bedc065..e2e7dc886e 100644 --- a/pkg/collector/collector.go +++ b/pkg/collector/collector.go @@ -86,7 +86,6 @@ func main() { log.Error(err, "failed to create pipe", "pipeFile", util.EraseCompleteCollectPath) os.Exit(1) } - defer completion.Close() data, err := completion.Await() if err != nil { @@ -94,6 +93,11 @@ func main() { os.Exit(1) } + if err := completion.Close(); err != nil { + log.Error(err, "failed to close pipe", "pipeFile", util.EraseCompleteCollectPath) + os.Exit(1) + } + if string(data) != util.EraseCompleteMessage { log.Info("garbage in pipe", "pipeFile", util.EraseCompleteCollectPath, "in_pipe", string(data)) os.Exit(1) diff --git a/pkg/scanners/template/scanner_template.go b/pkg/scanners/template/scanner_template.go index 2d51442891..9f12dcfa21 100644 --- a/pkg/scanners/template/scanner_template.go +++ b/pkg/scanners/template/scanner_template.go @@ -111,7 +111,7 @@ func (cfg *config) SendImages(nonCompliantImages, failedImages []unversioned.Ima } func (cfg *config) Finish() error { - defer cfg.completion.Close() + defer func() { _ = cfg.completion.Close() }() data, err := cfg.completion.Await() if err != nil { diff --git a/pkg/utils/handoff_test.go b/pkg/utils/handoff_test.go new file mode 100644 index 0000000000..19a74329c9 --- /dev/null +++ b/pkg/utils/handoff_test.go @@ -0,0 +1,108 @@ +package utils + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/eraser-dev/eraser/api/unversioned" +) + +// These run against whichever implementation the platform selects: FIFOs on +// Unix, Unix domain sockets on Windows. Keeping them build-tag free is the point +// -- the two implementations have to stay behaviorally identical. + +// shortTempDir keeps paths well inside the sun_path limit that applies to the +// socket implementation. +func shortTempDir(t *testing.T) string { + t.Helper() + + dir, err := os.MkdirTemp("", "h") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = os.RemoveAll(dir) }) + + return dir +} + +func TestImagesHandoffRoundTrip(t *testing.T) { + path := filepath.Join(shortTempDir(t), "images") + + want := []unversioned.Image{ + {ImageID: "sha256:aaaa", Names: []string{"repo/one:v1"}}, + {ImageID: "sha256:bbbb", Names: []string{"repo/two:v2"}}, + } + + errCh := make(chan error, 1) + go func() { errCh <- WriteImagesPipe(path, want) }() + + got, err := ReadImagesPipe(context.Background(), path) + if err != nil { + t.Fatalf("ReadImagesPipe: %v", err) + } + if err := <-errCh; err != nil { + t.Fatalf("WriteImagesPipe: %v", err) + } + + if len(got) != len(want) { + t.Fatalf("got %d images, want %d", len(got), len(want)) + } + for i := range want { + if got[i].ImageID != want[i].ImageID { + t.Errorf("image %d = %q, want %q", i, got[i].ImageID, want[i].ImageID) + } + } +} + +func TestCompletionHandoffRoundTrip(t *testing.T) { + path := filepath.Join(shortTempDir(t), "complete") + + pipe, err := CreateCompletionPipe(path) + if err != nil { + t.Fatalf("CreateCompletionPipe: %v", err) + } + defer func() { _ = pipe.Close() }() + + errCh := make(chan error, 1) + go func() { errCh <- WriteCompletionPipe(path) }() + + data, err := pipe.Await() + if err != nil { + t.Fatalf("Await: %v", err) + } + if err := <-errCh; err != nil { + t.Fatalf("WriteCompletionPipe: %v", err) + } + + if string(data) != EraseCompleteMessage { + t.Errorf("payload = %q, want %q", string(data), EraseCompleteMessage) + } +} + +// The remover infers "the scanner is disabled" from this error, so it has to +// keep satisfying os.IsNotExist on both platforms. +func TestWriteCompletionPipeAbsentPeerIsNotExist(t *testing.T) { + path := filepath.Join(shortTempDir(t), "no-such-peer") + + err := WriteCompletionPipe(path) + if err == nil { + t.Fatal("expected an error writing to an endpoint nobody published") + } + if !os.IsNotExist(err) { + t.Errorf("os.IsNotExist(%v) = false, want true", err) + } +} + +func TestCompletionPipeCloseIsIdempotentlySafe(t *testing.T) { + path := filepath.Join(shortTempDir(t), "closed") + + pipe, err := CreateCompletionPipe(path) + if err != nil { + t.Fatalf("CreateCompletionPipe: %v", err) + } + if err := pipe.Close(); err != nil { + t.Errorf("Close: %v", err) + } +} diff --git a/pkg/utils/handoff.go b/pkg/utils/handoff_unix.go similarity index 92% rename from pkg/utils/handoff.go rename to pkg/utils/handoff_unix.go index c5d936b9f6..28e7b6c9b4 100644 --- a/pkg/utils/handoff.go +++ b/pkg/utils/handoff_unix.go @@ -1,3 +1,5 @@ +//go:build !windows + package utils import ( @@ -11,9 +13,8 @@ import ( ) // The collector, scanner and remover hand images off to each other through -// endpoints in a shared volume. Each operation below was previously written out -// inline in the three worker binaries; they are gathered here so the transport -// can be swapped per platform without touching the callers. +// endpoints in a shared volume. On Unix those endpoints are FIFOs; see +// handoff_windows.go for the socket-based equivalent. // // Three properties are load-bearing and must survive any reimplementation. // Connecting blocks until the peer is on the other end, which is what makes @@ -89,7 +90,7 @@ func WriteImagesPipe(path string, images []unversioned.Image) error { } // ReadImagesPipe waits for the endpoint to appear, then reads until the writer -// finishes. It returns ctx.Err() if the context is cancelled while waiting. +// finishes. It returns ctx.Err() if the context is canceled while waiting. func ReadImagesPipe(ctx context.Context, path string) ([]unversioned.Image, error) { timer := time.NewTimer(time.Second) if !timer.Stop() { @@ -147,4 +148,4 @@ func WriteCompletionPipe(path string) error { } return file.Close() -} \ No newline at end of file +} diff --git a/pkg/utils/handoff_windows.go b/pkg/utils/handoff_windows.go new file mode 100644 index 0000000000..1fadc4eb75 --- /dev/null +++ b/pkg/utils/handoff_windows.go @@ -0,0 +1,189 @@ +//go:build windows + +package utils + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "io/fs" + "net" + "os" + "time" + + "github.com/eraser-dev/eraser/api/unversioned" +) + +// Windows has no filesystem FIFO, so the handoff runs over Unix domain sockets, +// which Windows has supported since Server 2019. A socket preserves the three +// properties the FIFO implementation relies on: Accept blocks until the peer +// dials, the reader sees EOF when the peer closes, and the endpoint is a file, +// so its absence still tells the remover that no scanner is present. +// +// Responsibility is inverted relative to Unix. With a FIFO the writer creates +// the endpoint and the reader polls for it; with a socket the reader listens and +// the writer dials with retry. + +// maxSocketPath is the sun_path limit. Exceeding it fails deep inside the +// syscall with an opaque error, so it is checked up front. +const maxSocketPath = 108 + +// CompletionPipe is an endpoint a peer can observe before anything is read from +// it. The scanner creates one early precisely so the remover can tell a scanner +// is present, which means the listener has to outlive its creation. +type CompletionPipe struct { + path string + l net.Listener +} + +// CreateCompletionPipe publishes the endpoint without waiting for a peer. +func CreateCompletionPipe(path string) (*CompletionPipe, error) { + l, err := listen(path) + if err != nil { + return nil, err + } + return &CompletionPipe{path: path, l: l}, nil +} + +// Await blocks until a peer signals completion. The payload is returned +// unvalidated so callers keep their existing handling of unexpected content. +func (p *CompletionPipe) Await() ([]byte, error) { + conn, err := p.l.Accept() + if err != nil { + return nil, err + } + defer func() { _ = conn.Close() }() + + return io.ReadAll(conn) +} + +// Close releases the endpoint, which also unpublishes it. +func (p *CompletionPipe) Close() error { + if p.l == nil { + return nil + } + return p.l.Close() +} + +// WriteImagesPipe blocks until the reader is listening, then sends the list. +// The unbounded retry mirrors the Unix implementation, where opening a FIFO for +// writing blocks until a reader arrives. +func WriteImagesPipe(path string, images []unversioned.Image) error { + data, err := json.Marshal(images) + if err != nil { + return err + } + + conn, err := dialForever(path) + if err != nil { + return err + } + + if _, err := conn.Write(data); err != nil { + _ = conn.Close() + return err + } + + // closing is what signals end-of-message to the reader + return conn.Close() +} + +// ReadImagesPipe publishes the endpoint and waits for the writer to connect and +// finish. It returns ctx.Err() if the context is canceled while waiting. +func ReadImagesPipe(ctx context.Context, path string) ([]unversioned.Image, error) { + l, err := listen(path) + if err != nil { + return nil, err + } + defer func() { _ = l.Close() }() + + // Accept has no context form; closing the listener is what unblocks it + done := make(chan struct{}) + defer close(done) + go func() { + select { + case <-ctx.Done(): + _ = l.Close() + case <-done: + } + }() + + conn, err := l.Accept() + if err != nil { + if ctxErr := ctx.Err(); ctxErr != nil { + return nil, ctxErr + } + return nil, err + } + defer func() { _ = conn.Close() }() + + data, err := io.ReadAll(conn) + if err != nil { + return nil, err + } + + images := []unversioned.Image{} + if err := json.Unmarshal(data, &images); err != nil { + return nil, err + } + + return images, nil +} + +// WriteCompletionPipe signals a peer that this stage is done. The returned error +// satisfies os.IsNotExist when the peer never published the endpoint, which is +// how an absent scanner is detected. +func WriteCompletionPipe(path string) error { + // Dialing a socket that is not there reports connection-refused on Windows + // rather than ENOENT, so the filesystem is the only reliable way to tell + // "never published" from "published but gone". + if _, err := os.Stat(path); err != nil { + return err + } + + conn, err := net.Dial("unix", path) + if err != nil { + return err + } + + if _, err := conn.Write([]byte(EraseCompleteMessage)); err != nil { + _ = conn.Close() + return err + } + + return conn.Close() +} + +func listen(path string) (net.Listener, error) { + if len(path) > maxSocketPath { + return nil, fmt.Errorf("socket path %q is %d bytes, over the %d byte limit", path, len(path), maxSocketPath) + } + + // a socket left behind by a previous run would fail the bind + if err := os.Remove(path); err != nil && !errors.Is(err, fs.ErrNotExist) { + return nil, err + } + + return net.Listen("unix", path) +} + +// dialForever waits for the reader to start listening. Errors are not +// classified: Windows reports a missing socket as connection-refused, so there +// is no reliable "not yet" error to match on. Retrying unconditionally mirrors +// the Unix implementation, where opening a FIFO for writing blocks until a +// reader arrives. +func dialForever(path string) (net.Conn, error) { + if len(path) > maxSocketPath { + return nil, fmt.Errorf("socket path %q is %d bytes, over the %d byte limit", path, len(path), maxSocketPath) + } + + for { + conn, err := net.Dial("unix", path) + if err == nil { + return conn, nil + } + time.Sleep(time.Second) + } +} From 8c925e117e9ca0dbe9b29bd4497129f90a35a24f Mon Sep 17 00:00:00 2001 From: Charles Wu Date: Wed, 19 Aug 2026 13:42:09 +1000 Subject: [PATCH 3/5] fix: correct the socket path limit and make Close idempotent Both found in review. sun_path is 108 bytes including the terminating NUL, so a pathname can use at most 107. The guard allowed 108 through to the opaque "bind: invalid argument" it exists to prevent. Confirmed on Windows: a 107-byte path binds, a 108-byte one does not. Close now clears the listener, so a second call reports success rather than net.ErrClosed. That is a real path, not a hypothetical one: the scanner defers Close while the collector also closes explicitly. The existing test only closed once despite its name, so it did not cover what it claimed. It now closes twice, and a new Windows-only test pins the 107/108 boundary so the limit cannot drift back. Signed-off-by: Charles Wu --- pkg/utils/handoff_test.go | 7 ++++++- pkg/utils/handoff_windows.go | 16 +++++++++++----- pkg/utils/platform_windows_test.go | 24 ++++++++++++++++++++++++ 3 files changed, 41 insertions(+), 6 deletions(-) diff --git a/pkg/utils/handoff_test.go b/pkg/utils/handoff_test.go index 19a74329c9..0d1e45a227 100644 --- a/pkg/utils/handoff_test.go +++ b/pkg/utils/handoff_test.go @@ -102,7 +102,12 @@ func TestCompletionPipeCloseIsIdempotentlySafe(t *testing.T) { if err != nil { t.Fatalf("CreateCompletionPipe: %v", err) } + + // the scanner defers Close and the collector also closes explicitly + if err := pipe.Close(); err != nil { + t.Errorf("first Close: %v", err) + } if err := pipe.Close(); err != nil { - t.Errorf("Close: %v", err) + t.Errorf("second Close: %v", err) } } diff --git a/pkg/utils/handoff_windows.go b/pkg/utils/handoff_windows.go index 1fadc4eb75..0323da79dc 100644 --- a/pkg/utils/handoff_windows.go +++ b/pkg/utils/handoff_windows.go @@ -26,9 +26,11 @@ import ( // the endpoint and the reader polls for it; with a socket the reader listens and // the writer dials with retry. -// maxSocketPath is the sun_path limit. Exceeding it fails deep inside the -// syscall with an opaque error, so it is checked up front. -const maxSocketPath = 108 +// maxSocketPath is the longest usable pathname. The sun_path field is 108 +// bytes and has to hold a terminating NUL, so 107 is the real ceiling. +// Exceeding it fails deep inside the syscall with an opaque error, so it is +// checked up front. +const maxSocketPath = 107 // CompletionPipe is an endpoint a peer can observe before anything is read from // it. The scanner creates one early precisely so the remover can tell a scanner @@ -59,12 +61,16 @@ func (p *CompletionPipe) Await() ([]byte, error) { return io.ReadAll(conn) } -// Close releases the endpoint, which also unpublishes it. +// Close releases the endpoint, which also unpublishes it. Callers both defer +// this and close explicitly, so repeat calls report success rather than +// net.ErrClosed. func (p *CompletionPipe) Close() error { if p.l == nil { return nil } - return p.l.Close() + l := p.l + p.l = nil + return l.Close() } // WriteImagesPipe blocks until the reader is listening, then sends the list. diff --git a/pkg/utils/platform_windows_test.go b/pkg/utils/platform_windows_test.go index 19e5693df5..4b4a911adb 100644 --- a/pkg/utils/platform_windows_test.go +++ b/pkg/utils/platform_windows_test.go @@ -7,6 +7,8 @@ import ( "errors" "fmt" "os" + "path/filepath" + "strings" "testing" "github.com/Microsoft/go-winio" @@ -55,6 +57,28 @@ func TestGetAddressAndDialer(t *testing.T) { } } +// The guard exists to replace an opaque syscall failure with a clear message, +// so the boundary it enforces has to be the real one. +func TestSocketPathLimitBoundary(t *testing.T) { + dir := shortTempDir(t) + + pathOfLen := func(n int) string { + return filepath.Join(dir, strings.Repeat("a", n-len(dir)-1)) + } + + atLimit := pathOfLen(maxSocketPath) + l, err := listen(atLimit) + if err != nil { + t.Fatalf("listen(%d-byte path) = %v, want success", len(atLimit), err) + } + _ = l.Close() + + overLimit := pathOfLen(maxSocketPath + 1) + if _, err := listen(overLimit); err == nil { + t.Errorf("listen(%d-byte path) = nil, want the guard to reject it", len(overLimit)) + } +} + func TestMkfifoUnsupported(t *testing.T) { if err := mkfifo("ignored", PipeMode); !errors.Is(err, ErrFifoUnsupported) { t.Errorf("mkfifo on windows = %v, want ErrFifoUnsupported", err) From 118719d443efa18f24dffb1f85db0a860784e477 Mon Sep 17 00:00:00 2001 From: Charles Wu Date: Wed, 19 Aug 2026 15:20:06 +1000 Subject: [PATCH 4/5] fix: publish the collector completion endpoint before the payload Found in review. The collector handed over the image list and only then created the endpoint it waits on. The peer can read that list and signal completion before the endpoint exists, in which case the remover fails its completion write and exits, while the collector blocks in Await forever. This is pre-existing on main and affects Linux identically: a FIFO that has not been created yet reports ENOENT from open() exactly as a missing socket reports it from stat(). The window is one syscall wide on either platform, which is why it has not been hit in practice. The scanner already publishes its completion endpoint before reading, so this makes the collector consistent with the rest of the codebase. It does change Linux ordering, which the PR description now calls out rather than continuing to claim Linux is untouched. Signed-off-by: Charles Wu --- pkg/collector/collector.go | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/pkg/collector/collector.go b/pkg/collector/collector.go index e2e7dc886e..16646f84f6 100644 --- a/pkg/collector/collector.go +++ b/pkg/collector/collector.go @@ -76,17 +76,20 @@ func main() { path = util.ScanErasePath } - if err := util.WriteImagesPipe(path, finalImages); err != nil { - log.Error(err, "failed to send images", "pipeFile", path) - os.Exit(1) - } - + // Published before the payload, not after: the peer can finish and signal + // back the moment it has read the list, so an endpoint created afterwards + // can be missed entirely. The scanner already publishes in this order. completion, err := util.CreateCompletionPipe(util.EraseCompleteCollectPath) if err != nil { log.Error(err, "failed to create pipe", "pipeFile", util.EraseCompleteCollectPath) os.Exit(1) } + if err := util.WriteImagesPipe(path, finalImages); err != nil { + log.Error(err, "failed to send images", "pipeFile", path) + os.Exit(1) + } + data, err := completion.Await() if err != nil { log.Error(err, "failed to read pipe", "pipeFile", util.EraseCompleteCollectPath) From e7455e10c7faca054b63690b6f3b300bed3fd6b3 Mon Sep 17 00:00:00 2001 From: Charles Wu Date: Wed, 19 Aug 2026 15:35:01 +1000 Subject: [PATCH 5/5] fix: close the FIFO descriptor on every path Found in review. ReadImagesPipe never closed the file it opened -- not on success, not on a read error, not on a JSON error. The remover closed that descriptor before this refactor, so it is a regression rather than an inherited bug. Three neighbours had the same class of leak on their error paths: Await, WriteImagesPipe and WriteCompletionPipe each returned early on a failed read or write without closing. All four now close on every path and propagate the close error only when the operation itself succeeded, which preserves the previous behaviour of surfacing a failed close. The socket implementation was already clean here; only the ported FIFO code leaked. Signed-off-by: Charles Wu --- pkg/utils/handoff_unix.go | 32 ++++++++++++++++++++------------ 1 file changed, 20 insertions(+), 12 deletions(-) diff --git a/pkg/utils/handoff_unix.go b/pkg/utils/handoff_unix.go index 28e7b6c9b4..49ae8dd35e 100644 --- a/pkg/utils/handoff_unix.go +++ b/pkg/utils/handoff_unix.go @@ -48,19 +48,22 @@ func (p *CompletionPipe) Await() ([]byte, error) { } data, err := io.ReadAll(file) - if err != nil { - return nil, err + if closeErr := file.Close(); closeErr != nil && err == nil { + err = closeErr } - - if err := file.Close(); err != nil { + if err != nil { return nil, err } return data, nil } -// Close releases the endpoint. It is a no-op where the endpoint is a plain -// filesystem object. +// Close releases this process's hold on the endpoint. The FIFO itself is left +// in place: the remover decides whether a scanner exists by whether the +// scanner's endpoint is on disk, so unlinking here would make a live scanner +// look absent. The socket implementation cannot keep the endpoint after Close +// because the listener owns it, which is the one lifecycle difference between +// the two. func (p *CompletionPipe) Close() error { return nil } @@ -82,11 +85,12 @@ func WriteImagesPipe(path string, images []unversioned.Image) error { return err } - if _, err := file.Write(data); err != nil { - return err + _, err = file.Write(data) + if closeErr := file.Close(); closeErr != nil && err == nil { + err = closeErr } - return file.Close() + return err } // ReadImagesPipe waits for the endpoint to appear, then reads until the writer @@ -121,6 +125,9 @@ func ReadImagesPipe(ctx context.Context, path string) ([]unversioned.Image, erro } data, err := io.ReadAll(f) + if closeErr := f.Close(); closeErr != nil && err == nil { + err = closeErr + } if err != nil { return nil, err } @@ -143,9 +150,10 @@ func WriteCompletionPipe(path string) error { return err } - if _, err := file.WriteString(EraseCompleteMessage); err != nil { - return err + _, err = file.WriteString(EraseCompleteMessage) + if closeErr := file.Close(); closeErr != nil && err == nil { + err = closeErr } - return file.Close() + return err }