From cc0337c4d5620f560e17bd1a8cf28a1be34f0141 Mon Sep 17 00:00:00 2001 From: Andrew Nesbitt Date: Wed, 21 Oct 2026 11:41:38 +0100 Subject: [PATCH 1/2] Allow skipping unread ZIP entries after size rejection --- README.md | 8 +++ stream.go | 53 ++++++++++++-- stream_skip_test.go | 171 ++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 228 insertions(+), 4 deletions(-) create mode 100644 stream_skip_test.go diff --git a/README.md b/README.md index b9617b1..a8b89ad 100644 --- a/README.md +++ b/README.md @@ -175,6 +175,14 @@ the next member's offset. TAR-based formats, including conda, return Cancellation and limit errors remain terminal. Recovery does not validate unread data in the discarded member. +`stream.Skip()` abandons an unread ZIP entry without decompressing or validating +its body. This also clears a size-limit rejection from `Next`, which returns the +rejected entry alongside its error. The entry still counts toward `MaxEntries`, +but its unread body does not consume the expanded-byte budget. Later entries can +use the remaining budget. `Skip` rejects entries already read and does not support +TAR-based formats or conda. Input limits, entry-count limits and cancellation +remain terminal. + ZIP and conda require random access to their compressed input. `OpenStream` buffers that input within `MaxInputBytes`; `OpenStreamBytes(name, data, options)` reuses an existing byte slice without copying it. Keep that slice unchanged diff --git a/stream.go b/stream.go index 30892d3..67b537b 100644 --- a/stream.go +++ b/stream.go @@ -45,6 +45,9 @@ type Stream struct { unknownSize bool entryBytes int64 entryPath string + pending *StreamEntry + entryRead bool + reserved int64 } type streamSource interface { @@ -189,7 +192,8 @@ func streamCompression(format string) (string, bool) { // the body is exposed. Unknown sizes count as bytes are read or skipped. // MaxEntries counts visible entries; nested containers // have a separate MaxEntries budget. Hidden TAR extension headers are excluded. -// Any iteration error is terminal. +// Entry-limit errors return the rejected entry alongside the error. +// Errors remain terminal unless explicitly cleared by Recover or Skip. func (s *Stream) Next() (*StreamEntry, error) { if s.closed { return nil, fs.ErrClosed @@ -206,20 +210,27 @@ func (s *Stream) Next() (*StreamEntry, error) { return nil, err } } + s.pending = nil + s.reserved = 0 + s.entryRead = false entry, err := s.source.Next() if err == nil { + s.pending = entry err = s.checkEntry(entry) + if !errors.Is(err, ErrEntryLimit) { + s.entries++ + } } if err != nil { s.err = err - return nil, err + return entry, err } - s.entries++ s.unknownSize = entry.Size == -1 s.entryBytes = 0 s.entryPath = entry.Path if !s.unknownSize { s.expanded += entry.Size + s.reserved = entry.Size } return entry, nil } @@ -233,7 +244,7 @@ func (s *Stream) checkEntry(entry *StreamEntry) error { return fmt.Errorf("%w: %s exceeds %d bytes", ErrEntrySizeLimit, entry.Path, s.options.MaxEntryBytes) } if entry.Size > s.options.MaxExpandedBytes-s.expanded { - return fmt.Errorf("%w: exceeds %d bytes", ErrDecompressLimit, s.options.MaxExpandedBytes) + return fmt.Errorf("%w: %s exceeds remaining budget of %d bytes", ErrDecompressLimit, entry.Path, s.options.MaxExpandedBytes-s.expanded) } return nil } @@ -253,6 +264,9 @@ func (s *Stream) Read(p []byte) (int, error) { } var n int var err error + if len(p) > 0 { + s.entryRead = true + } if s.unknownSize { n, err = s.readUnknownSize(p) } else { @@ -312,6 +326,37 @@ func (s *Stream) Recover() error { } zip.recover() s.err = nil + s.pending = nil + return nil +} + +// Skip abandons an unread ZIP entry without decompressing or validating it. +// It also clears size-limit errors returned with an entry by Next. The entry +// counts toward MaxEntries, but its unread body does not consume expanded bytes. +// Other formats, missing entries and entries already read return ErrUnsupported. +// Cancellation, input limits and entry-count limits remain terminal. +func (s *Stream) Skip() error { + if s.closed { + return fs.ErrClosed + } + if err := s.ctx.Err(); err != nil { + return err + } + if errors.Is(s.err, io.EOF) { + return errors.ErrUnsupported + } + if s.err != nil && !errors.Is(s.err, ErrEntrySizeLimit) && !errors.Is(s.err, ErrDecompressLimit) { + return s.err + } + zip, ok := s.source.(*zipStream) + if !ok || s.pending == nil || s.entryRead { + return errors.ErrUnsupported + } + zip.recover() + s.expanded -= s.reserved + s.pending = nil + s.reserved = 0 + s.err = nil return nil } diff --git a/stream_skip_test.go b/stream_skip_test.go new file mode 100644 index 0000000..6ed03b9 --- /dev/null +++ b/stream_skip_test.go @@ -0,0 +1,171 @@ +package archives + +import ( + "archive/zip" + "bytes" + "context" + "errors" + "io" + "io/fs" + "math" + "testing" +) + +func skipZIPFixture(t *testing.T, method uint16) []byte { + t.Helper() + var buf bytes.Buffer + w := zip.NewWriter(&buf) + for _, entry := range []struct{ name, body string }{{"big", string(bytes.Repeat([]byte{'x'}, 2700))}, {"small", "valid"}, {"tail", "end"}} { + body, err := w.CreateHeader(&zip.FileHeader{Name: entry.name, Method: method}) + if err != nil { + t.Fatal(err) + } + if _, err := io.WriteString(body, entry.body); err != nil { + t.Fatal(err) + } + } + if err := w.Close(); err != nil { + t.Fatal(err) + } + return buf.Bytes() +} + +func TestStreamSkipRejectedZIPEntry(t *testing.T) { + for _, method := range []uint16{zip.Store, zip.Deflate} { + for _, options := range []StreamOptions{{MaxEntryBytes: 5}, {MaxExpandedBytes: 5}} { + assertSkipRejectedZIPEntry(t, method, options) + } + } +} + +func assertSkipRejectedZIPEntry(t *testing.T, method uint16, options StreamOptions) { + t.Helper() + s, err := OpenStreamBytes("renamed", skipZIPFixture(t, method), options) + if err != nil { + t.Fatal(err) + } + entry, err := s.Next() + if entry == nil || entry.Path != "big" || (!errors.Is(err, ErrEntrySizeLimit) && !errors.Is(err, ErrDecompressLimit)) { + t.Fatalf("rejected entry: %+v %v", entry, err) + } + if _, again := s.Next(); again != err { + t.Fatal("size rejection should remain terminal until Skip") + } + if err := s.Skip(); err != nil { + t.Fatal(err) + } + entry, err = s.Next() + if err != nil || entry.Path != "small" { + t.Fatalf("small entry: %+v %v", entry, err) + } + body, err := io.ReadAll(s) + if err != nil || string(body) != "valid" { + t.Fatalf("small body: %q %v", body, err) + } + _, err = s.Next() + if options.MaxExpandedBytes > 0 && !errors.Is(err, ErrDecompressLimit) { + t.Fatalf("read bytes lost from budget: %v", err) + } + if err := s.Close(); err != nil { + t.Fatal(err) + } +} + +func TestStreamSkipUnreadZIPEntry(t *testing.T) { + s, err := OpenStreamBytes("renamed", skipZIPFixture(t, zip.Deflate), StreamOptions{MaxExpandedBytes: 2700}) + if err != nil { + t.Fatal(err) + } + defer func() { _ = s.Close() }() + if err := s.Skip(); !errors.Is(err, errors.ErrUnsupported) { + t.Fatal(err) + } + entry, err := s.Next() + if err != nil { + t.Fatal(err) + } + entry.Size = math.MaxInt64 + if err := s.Skip(); err != nil { + t.Fatal(err) + } + if err := s.Skip(); !errors.Is(err, errors.ErrUnsupported) { + t.Fatal(err) + } + entry, err = s.Next() + if err != nil || entry.Path != "small" { + t.Fatalf("reserved bytes not released: %+v %v", entry, err) + } + if _, err := s.Read(make([]byte, 1)); err != nil { + t.Fatal(err) + } + if err := s.Skip(); !errors.Is(err, errors.ErrUnsupported) { + t.Fatalf("partially read entry skipped: %v", err) + } + if err := consumeStream(s, true); err != nil { + t.Fatal(err) + } + if err := s.Skip(); !errors.Is(err, errors.ErrUnsupported) { + t.Fatal(err) + } +} + +func TestStreamSkipDoesNotReadRejectedBody(t *testing.T) { + for _, method := range []uint16{zip.Store, zip.Deflate} { + s, err := OpenStreamBytes("renamed", corruptZipMember(t, method), StreamOptions{MaxExpandedBytes: 5}) + if err != nil { + t.Fatal(err) + } + if _, err := s.Next(); !errors.Is(err, ErrDecompressLimit) { + t.Fatal(err) + } + if err := s.Skip(); err != nil { + t.Fatal(err) + } + entry, err := s.Next() + if err != nil || entry.Path != "valid" { + t.Fatalf("next after corrupt skipped body: %+v %v", entry, err) + } + body, err := io.ReadAll(s) + if err != nil || string(body) != "valid" { + t.Fatalf("body: %q %v", body, err) + } + if err := s.Close(); err != nil { + t.Fatal(err) + } + } +} + +func TestStreamSkipUnsupportedFormatsAndCancellation(t *testing.T) { + for _, name := range []string{"test.tar", "test.tar.gz", "test.conda"} { + s, err := OpenStreamBytes(name, streamTestArchive(t, name, tarWithPayloads(t, make([]byte, 2700))), StreamOptions{MaxEntryBytes: 10}) + if err != nil { + t.Fatal(err) + } + if _, err := s.Next(); !errors.Is(err, ErrEntrySizeLimit) { + t.Fatal(err) + } + if err := s.Skip(); !errors.Is(err, errors.ErrUnsupported) { + t.Fatalf("%s: %v", name, err) + } + _ = s.Close() + } + ctx, cancel := context.WithCancel(context.Background()) + s, err := OpenStreamBytesContext(ctx, "renamed", skipZIPFixture(t, zip.Store), StreamOptions{}) + if err != nil { + t.Fatal(err) + } + if _, err := s.Next(); err != nil { + t.Fatal(err) + } + cancel() + if err := s.Skip(); !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + _ = s.Close() + if err := s.Skip(); !errors.Is(err, fs.ErrClosed) { + t.Fatal(err) + } + if _, err := OpenStreamBytes("renamed", skipZIPFixture(t, zip.Store), StreamOptions{MaxEntries: 2}); !errors.Is(err, ErrEntryLimit) { + t.Fatalf("entry count limit bypassed: %v", err) + } +} From f6f78caa1a39490d2f20c7c9bb18cf50bacace36 Mon Sep 17 00:00:00 2001 From: Andrew Nesbitt Date: Wed, 21 Oct 2026 11:47:49 +0100 Subject: [PATCH 2/2] Clarify which Next errors return the rejected entry --- stream.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/stream.go b/stream.go index 67b537b..12eafc4 100644 --- a/stream.go +++ b/stream.go @@ -192,7 +192,7 @@ func streamCompression(format string) (string, bool) { // the body is exposed. Unknown sizes count as bytes are read or skipped. // MaxEntries counts visible entries; nested containers // have a separate MaxEntries budget. Hidden TAR extension headers are excluded. -// Entry-limit errors return the rejected entry alongside the error. +// Size and entry-count limit errors return the rejected entry alongside the error. // Errors remain terminal unless explicitly cleared by Recover or Skip. func (s *Stream) Next() (*StreamEntry, error) { if s.closed {