From 7595fabf87eb8dd4df01d80d9d3f9a84e73015b5 Mon Sep 17 00:00:00 2001 From: Andrew Nesbitt Date: Wed, 21 Oct 2026 10:40:56 +0100 Subject: [PATCH 1/2] Allow explicit recovery after ZIP member errors --- README.md | 6 +++ stream.go | 27 +++++++++++ stream_recover_test.go | 100 +++++++++++++++++++++++++++++++++++++++++ stream_zip.go | 11 +++++ 4 files changed, 144 insertions(+) create mode 100644 stream_recover_test.go diff --git a/README.md b/README.md index 986e9db..51fe8b7 100644 --- a/README.md +++ b/README.md @@ -168,6 +168,12 @@ are terminal. The caller owns the input, and `Close` releases decoder resources without closing or draining it. A TAR end marker ends iteration, so trailing data and compression trailers beyond that marker may remain unchecked. +After a ZIP member error, callers can explicitly call `stream.Recover()` to +discard that member and continue with `Next`. The central directory provides +the next member's offset. Other formats return `errors.ErrUnsupported` from +`Recover`; cancellation and limit errors remain terminal. Recovery does not +validate unread data in the discarded member. + 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 63f8baf..5e1eb24 100644 --- a/stream.go +++ b/stream.go @@ -240,6 +240,33 @@ func (s *Stream) Read(p []byte) (int, error) { return n, err } +// Recover abandons a failed ZIP entry so the next call to Next can continue. +// Other formats return errors.ErrUnsupported. Cancellation and resource-limit +// errors remain terminal. Recovery does not validate the discarded body. +func (s *Stream) Recover() error { + if s.closed { + return fs.ErrClosed + } + if err := s.ctx.Err(); err != nil { + return err + } + if s.err == nil || errors.Is(s.err, io.EOF) { + return errors.ErrUnsupported + } + if errors.Is(s.err, ErrInputLimit) || errors.Is(s.err, ErrEntryLimit) || errors.Is(s.err, ErrEntrySizeLimit) || errors.Is(s.err, ErrDecompressLimit) { + return s.err + } + zip, ok := s.source.(*zipStream) + if !ok { + return errors.ErrUnsupported + } + if err := zip.recover(); err != nil { + return err + } + s.err = nil + return nil +} + // Close releases resources without draining or closing the input. Unread // content and compressed trailers after a TAR end marker are not validated. func (s *Stream) Close() error { diff --git a/stream_recover_test.go b/stream_recover_test.go new file mode 100644 index 0000000..5df34c3 --- /dev/null +++ b/stream_recover_test.go @@ -0,0 +1,100 @@ +package archives + +import ( + "archive/zip" + "bytes" + "context" + "errors" + "io" + "testing" +) + +func corruptZipMember(t *testing.T) []byte { + t.Helper() + var buf bytes.Buffer + w := zip.NewWriter(&buf) + for _, name := range []string{"broken", "valid"} { + member, err := w.CreateHeader(&zip.FileHeader{Name: name, Method: zip.Store}) + if err != nil { + t.Fatal(err) + } + if _, err := io.WriteString(member, name); err != nil { + t.Fatal(err) + } + } + if err := w.Close(); err != nil { + t.Fatal(err) + } + data := buf.Bytes() + reader, err := zip.NewReader(bytes.NewReader(data), int64(len(data))) + if err != nil { + t.Fatal(err) + } + offset, err := reader.File[0].DataOffset() + if err != nil { + t.Fatal(err) + } + data[offset] ^= 1 + return data +} + +func TestStreamRecoverZIP(t *testing.T) { + for _, skip := range []bool{false, true} { + s, err := OpenStreamBytes("test.zip", corruptZipMember(t), StreamOptions{}) + if err != nil { + t.Fatal(err) + } + if _, err := s.Next(); err != nil { + t.Fatal(err) + } + if skip { + _, err = s.Next() + } else { + _, err = io.ReadAll(s) + } + if !errors.Is(err, zip.ErrChecksum) { + t.Fatalf("member error: %v", err) + } + if err := s.Recover(); err != nil { + t.Fatal(err) + } + next, err := s.Next() + if err != nil || next.Path != "valid" { + t.Fatalf("next: %+v %v", next, err) + } + body, err := io.ReadAll(s) + if err != nil || string(body) != "valid" { + t.Fatalf("body: %q %v", body, err) + } + if _, err := s.Next(); err != io.EOF { + t.Fatal(err) + } + if err := s.Close(); err != nil { + t.Fatal(err) + } + } +} + +func TestStreamRecoverPreservesLimitsAndCancellation(t *testing.T) { + for _, options := range []StreamOptions{{MaxEntryBytes: 1}, {MaxExpandedBytes: 1}} { + s, err := OpenStreamBytes("test.zip", corruptZipMember(t), options) + if err != nil { + t.Fatal(err) + } + _, original := s.Next() + if original == nil || s.Recover() != original { + t.Fatalf("limit bypassed: %v", original) + } + _ = s.Close() + } + ctx, cancel := context.WithCancel(context.Background()) + s, err := OpenStreamBytesContext(ctx, "test.zip", corruptZipMember(t), StreamOptions{}) + if err != nil { + t.Fatal(err) + } + cancel() + if err := s.Recover(); !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + _ = s.Close() +} diff --git a/stream_zip.go b/stream_zip.go index 09fb4dd..b2f1d59 100644 --- a/stream_zip.go +++ b/stream_zip.go @@ -85,6 +85,17 @@ func (z *zipStream) Close() error { return nil } +func (z *zipStream) recover() error { + if z.current != nil { + if err := z.current.Close(); err != nil { + return err + } + } + z.current = nil + z.file = nil + return nil +} + type condaStream struct { outer *Stream inner *tarStream From 132a52f8efa20b8901dea48e53e1690504a3d4f9 Mon Sep 17 00:00:00 2001 From: Andrew Nesbitt Date: Wed, 21 Oct 2026 10:51:21 +0100 Subject: [PATCH 2/2] Discard failed deflate readers during ZIP recovery --- README.md | 7 ++- stream.go | 9 ++- stream_recover_test.go | 124 +++++++++++++++++++++++++++++------------ stream_zip.go | 8 +-- 4 files changed, 98 insertions(+), 50 deletions(-) diff --git a/README.md b/README.md index 51fe8b7..3e2f886 100644 --- a/README.md +++ b/README.md @@ -170,9 +170,10 @@ data and compression trailers beyond that marker may remain unchecked. After a ZIP member error, callers can explicitly call `stream.Recover()` to discard that member and continue with `Next`. The central directory provides -the next member's offset. Other formats return `errors.ErrUnsupported` from -`Recover`; cancellation and limit errors remain terminal. Recovery does not -validate unread data in the discarded member. +the next member's offset. TAR-based formats, including conda, return +`errors.ErrUnsupported` from `Recover`, as does a stream with no pending error. +Cancellation and limit errors remain terminal. Recovery does not validate unread +data in the discarded member. ZIP and conda require random access to their compressed input. `OpenStream` buffers that input within `MaxInputBytes`; `OpenStreamBytes(name, data, options)` diff --git a/stream.go b/stream.go index 5e1eb24..8373fc5 100644 --- a/stream.go +++ b/stream.go @@ -241,8 +241,9 @@ func (s *Stream) Read(p []byte) (int, error) { } // Recover abandons a failed ZIP entry so the next call to Next can continue. -// Other formats return errors.ErrUnsupported. Cancellation and resource-limit -// errors remain terminal. Recovery does not validate the discarded body. +// TAR-based formats, including conda, return errors.ErrUnsupported, as does a +// stream with no pending error. Cancellation and resource-limit errors remain +// terminal. Recovery does not validate the discarded body. func (s *Stream) Recover() error { if s.closed { return fs.ErrClosed @@ -260,9 +261,7 @@ func (s *Stream) Recover() error { if !ok { return errors.ErrUnsupported } - if err := zip.recover(); err != nil { - return err - } + zip.recover() s.err = nil return nil } diff --git a/stream_recover_test.go b/stream_recover_test.go index 5df34c3..2d542b7 100644 --- a/stream_recover_test.go +++ b/stream_recover_test.go @@ -3,18 +3,19 @@ package archives import ( "archive/zip" "bytes" + "compress/flate" "context" "errors" "io" "testing" ) -func corruptZipMember(t *testing.T) []byte { +func corruptZipMember(t *testing.T, method uint16) []byte { t.Helper() var buf bytes.Buffer w := zip.NewWriter(&buf) for _, name := range []string{"broken", "valid"} { - member, err := w.CreateHeader(&zip.FileHeader{Name: name, Method: zip.Store}) + member, err := w.CreateHeader(&zip.FileHeader{Name: name, Method: method}) if err != nil { t.Fatal(err) } @@ -34,50 +35,62 @@ func corruptZipMember(t *testing.T) []byte { if err != nil { t.Fatal(err) } - data[offset] ^= 1 + if method == zip.Store { + data[offset] ^= 1 + } else { + data[offset] = 0x07 // Reserved DEFLATE block type. + } return data } func TestStreamRecoverZIP(t *testing.T) { - for _, skip := range []bool{false, true} { - s, err := OpenStreamBytes("test.zip", corruptZipMember(t), StreamOptions{}) - if err != nil { - t.Fatal(err) - } - if _, err := s.Next(); err != nil { - t.Fatal(err) - } - if skip { - _, err = s.Next() - } else { - _, err = io.ReadAll(s) - } - if !errors.Is(err, zip.ErrChecksum) { - t.Fatalf("member error: %v", err) - } - if err := s.Recover(); err != nil { - t.Fatal(err) - } - next, err := s.Next() - if err != nil || next.Path != "valid" { - t.Fatalf("next: %+v %v", next, err) - } - body, err := io.ReadAll(s) - if err != nil || string(body) != "valid" { - t.Fatalf("body: %q %v", body, err) - } - if _, err := s.Next(); err != io.EOF { - t.Fatal(err) - } - if err := s.Close(); err != nil { - t.Fatal(err) + for _, method := range []uint16{zip.Store, zip.Deflate} { + for _, skip := range []bool{false, true} { + assertRecoverZIP(t, method, skip) } } } +func assertRecoverZIP(t *testing.T, method uint16, skip bool) { + t.Helper() + s, err := OpenStreamBytes("test.zip", corruptZipMember(t, method), StreamOptions{}) + if err != nil { + t.Fatal(err) + } + if _, err := s.Next(); err != nil { + t.Fatal(err) + } + if skip { + _, err = s.Next() + } else { + _, err = io.ReadAll(s) + } + var corrupt flate.CorruptInputError + if (method == zip.Store && !errors.Is(err, zip.ErrChecksum)) || (method == zip.Deflate && !errors.As(err, &corrupt)) { + t.Fatalf("member error: %v", err) + } + if err := s.Recover(); err != nil { + t.Fatal(err) + } + next, err := s.Next() + if err != nil || next.Path != "valid" { + t.Fatalf("next: %+v %v", next, err) + } + body, err := io.ReadAll(s) + if err != nil || string(body) != "valid" { + t.Fatalf("body: %q %v", body, err) + } + if _, err := s.Next(); err != io.EOF { + t.Fatal(err) + } + if err := s.Close(); err != nil { + t.Fatal(err) + } +} + func TestStreamRecoverPreservesLimitsAndCancellation(t *testing.T) { for _, options := range []StreamOptions{{MaxEntryBytes: 1}, {MaxExpandedBytes: 1}} { - s, err := OpenStreamBytes("test.zip", corruptZipMember(t), options) + s, err := OpenStreamBytes("test.zip", corruptZipMember(t, zip.Store), options) if err != nil { t.Fatal(err) } @@ -88,7 +101,7 @@ func TestStreamRecoverPreservesLimitsAndCancellation(t *testing.T) { _ = s.Close() } ctx, cancel := context.WithCancel(context.Background()) - s, err := OpenStreamBytesContext(ctx, "test.zip", corruptZipMember(t), StreamOptions{}) + s, err := OpenStreamBytesContext(ctx, "test.zip", corruptZipMember(t, zip.Store), StreamOptions{}) if err != nil { t.Fatal(err) } @@ -98,3 +111,40 @@ func TestStreamRecoverPreservesLimitsAndCancellation(t *testing.T) { } _ = s.Close() } + +func TestStreamRecoverUnsupported(t *testing.T) { + raw := tarWithPayloads(t, make([]byte, 1024))[:1023] + for _, name := range []string{"test.tar", "test.tar.gz", "test.conda"} { + s, err := OpenStreamBytes(name, streamTestArchive(t, name, raw), StreamOptions{}) + if err != nil { + t.Fatal(err) + } + if _, err := s.Next(); err != nil { + t.Fatal(err) + } + if _, err := io.ReadAll(s); !errors.Is(err, io.ErrUnexpectedEOF) { + t.Fatalf("%s: %v", name, err) + } + if err := s.Recover(); !errors.Is(err, errors.ErrUnsupported) { + t.Fatalf("%s: %v", name, err) + } + _ = s.Close() + } +} + +func TestStreamRecoverWithoutError(t *testing.T) { + s, err := OpenStreamBytes("test.zip", createTestZip(), StreamOptions{}) + if err != nil { + t.Fatal(err) + } + defer func() { _ = s.Close() }() + if err := s.Recover(); !errors.Is(err, errors.ErrUnsupported) { + t.Fatal(err) + } + if err := consumeStream(s, true); err != nil { + t.Fatal(err) + } + if err := s.Recover(); !errors.Is(err, errors.ErrUnsupported) { + t.Fatal(err) + } +} diff --git a/stream_zip.go b/stream_zip.go index b2f1d59..126f6f2 100644 --- a/stream_zip.go +++ b/stream_zip.go @@ -85,15 +85,13 @@ func (z *zipStream) Close() error { return nil } -func (z *zipStream) recover() error { +func (z *zipStream) recover() { if z.current != nil { - if err := z.current.Close(); err != nil { - return err - } + // A decompressor can repeat the already-reported member error on Close. + _ = z.current.Close() } z.current = nil z.file = nil - return nil } type condaStream struct {