diff --git a/src/SharpCompress/Archives/Zip/ZipArchive.cs b/src/SharpCompress/Archives/Zip/ZipArchive.cs index 1a87df5e..7ce8a116 100644 --- a/src/SharpCompress/Archives/Zip/ZipArchive.cs +++ b/src/SharpCompress/Archives/Zip/ZipArchive.cs @@ -299,9 +299,10 @@ public class ZipArchive : AbstractWritableArchive stream = new SharpCompressStream(stream, bufferSize: bufferSize); } - var header = headerFactory - .ReadStreamHeader(stream) - .FirstOrDefault(x => x.ZipHeaderType != ZipHeaderType.Split); + var header = await headerFactory + .ReadStreamHeaderAsync(stream) + .Where(x => x.ZipHeaderType != ZipHeaderType.Split) + .FirstOrDefaultAsync(); if (header is null) { return false; diff --git a/src/SharpCompress/Common/Zip/StreamingZipHeaderFactory.cs b/src/SharpCompress/Common/Zip/StreamingZipHeaderFactory.cs index 031287ed..eafab136 100644 --- a/src/SharpCompress/Common/Zip/StreamingZipHeaderFactory.cs +++ b/src/SharpCompress/Common/Zip/StreamingZipHeaderFactory.cs @@ -2,6 +2,9 @@ using System; using System.Collections.Generic; using System.IO; using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using SharpCompress.Common; using SharpCompress.Common.Zip.Headers; using SharpCompress.IO; @@ -200,4 +203,331 @@ internal class StreamingZipHeaderFactory : ZipHeaderFactory yield return header; } } + + /// + /// Reads ZIP headers asynchronously for streams that do not support synchronous reads. + /// + internal IAsyncEnumerable ReadStreamHeaderAsync(Stream stream) => + new StreamHeaderAsyncEnumerable(this, stream); + + /// + /// Invokes the shared async header parsing logic on the base factory. + /// + private ValueTask ReadHeaderAsyncInternal( + uint headerBytes, + AsyncBinaryReader reader + ) => ReadHeader(headerBytes, reader); + + /// + /// Exposes the last parsed local entry header to the async enumerator so it can handle streaming data descriptors. + /// + private LocalEntryHeader? LastEntryHeader + { + get => _lastEntryHeader; + set => _lastEntryHeader = value; + } + + /// + /// Produces an async enumerator for streaming ZIP headers. + /// + private sealed class StreamHeaderAsyncEnumerable : IAsyncEnumerable + { + private readonly StreamingZipHeaderFactory _headerFactory; + private readonly Stream _stream; + + public StreamHeaderAsyncEnumerable(StreamingZipHeaderFactory headerFactory, Stream stream) + { + _headerFactory = headerFactory; + _stream = stream; + } + + public IAsyncEnumerator GetAsyncEnumerator( + CancellationToken cancellationToken = default + ) => new StreamHeaderAsyncEnumerator(_headerFactory, _stream, cancellationToken); + } + + /// + /// Async implementation of using to avoid sync reads. + /// + private sealed class StreamHeaderAsyncEnumerator : IAsyncEnumerator, IDisposable + { + private readonly StreamingZipHeaderFactory _headerFactory; + private readonly SharpCompressStream _rewindableStream; + private readonly AsyncBinaryReader _reader; + private readonly CancellationToken _cancellationToken; + private bool _completed; + + public StreamHeaderAsyncEnumerator( + StreamingZipHeaderFactory headerFactory, + Stream stream, + CancellationToken cancellationToken + ) + { + _headerFactory = headerFactory; + _rewindableStream = EnsureSharpCompressStream(stream); + _reader = new AsyncBinaryReader(_rewindableStream, leaveOpen: true); + _cancellationToken = cancellationToken; + } + + private ZipHeader? _current; + + public ZipHeader Current => + _current ?? throw new InvalidOperationException("No current header is available."); + + /// + /// Advances to the next ZIP header in the stream, honoring streaming data descriptors where applicable. + /// + public async ValueTask MoveNextAsync() + { + if (_completed) + { + return false; + } + + while (true) + { + _cancellationToken.ThrowIfCancellationRequested(); + + uint headerBytes; + var lastEntryHeader = _headerFactory.LastEntryHeader; + if ( + lastEntryHeader != null + && FlagUtility.HasFlag(lastEntryHeader.Flags, HeaderFlags.UsePostDataDescriptor) + ) + { + if (lastEntryHeader.Part is null) + { + continue; + } + + var pos = _rewindableStream.CanSeek ? (long?)_rewindableStream.Position : null; + + var crc = await _reader + .ReadUInt32Async(_cancellationToken) + .ConfigureAwait(false); + if (crc == POST_DATA_DESCRIPTOR) + { + crc = await _reader + .ReadUInt32Async(_cancellationToken) + .ConfigureAwait(false); + } + lastEntryHeader.Crc = crc; + + //attempt 32bit read + ulong compressedSize = await _reader + .ReadUInt32Async(_cancellationToken) + .ConfigureAwait(false); + ulong uncompressedSize = await _reader + .ReadUInt32Async(_cancellationToken) + .ConfigureAwait(false); + headerBytes = await _reader + .ReadUInt32Async(_cancellationToken) + .ConfigureAwait(false); + + //check for zip64 sentinel or unexpected header + bool isSentinel = + compressedSize == 0xFFFFFFFF || uncompressedSize == 0xFFFFFFFF; + bool isHeader = headerBytes == 0x04034b50 || headerBytes == 0x02014b50; + + if (!isHeader && !isSentinel) + { + //reshuffle into 64-bit values + compressedSize = (uncompressedSize << 32) | compressedSize; + uncompressedSize = + ((ulong)headerBytes << 32) + | await _reader + .ReadUInt32Async(_cancellationToken) + .ConfigureAwait(false); + headerBytes = await _reader + .ReadUInt32Async(_cancellationToken) + .ConfigureAwait(false); + } + else if (isSentinel) + { + //standards-compliant zip64 descriptor + compressedSize = await _reader + .ReadUInt64Async(_cancellationToken) + .ConfigureAwait(false); + uncompressedSize = await _reader + .ReadUInt64Async(_cancellationToken) + .ConfigureAwait(false); + } + + lastEntryHeader.CompressedSize = (long)compressedSize; + lastEntryHeader.UncompressedSize = (long)uncompressedSize; + + if (pos.HasValue) + { + lastEntryHeader.DataStartPosition = pos - lastEntryHeader.CompressedSize; + } + } + else if (lastEntryHeader != null && lastEntryHeader.IsZip64) + { + if (lastEntryHeader.Part is null) + { + continue; + } + + var pos = _rewindableStream.CanSeek ? (long?)_rewindableStream.Position : null; + + headerBytes = await _reader + .ReadUInt32Async(_cancellationToken) + .ConfigureAwait(false); + + _ = await _reader.ReadUInt16Async(_cancellationToken).ConfigureAwait(false); // version + _ = await _reader.ReadUInt16Async(_cancellationToken).ConfigureAwait(false); // flags + _ = await _reader.ReadUInt16Async(_cancellationToken).ConfigureAwait(false); // compressionMethod + _ = await _reader.ReadUInt16Async(_cancellationToken).ConfigureAwait(false); // lastModifiedDate + _ = await _reader.ReadUInt16Async(_cancellationToken).ConfigureAwait(false); // lastModifiedTime + + var crc = await _reader + .ReadUInt32Async(_cancellationToken) + .ConfigureAwait(false); + + if (crc == POST_DATA_DESCRIPTOR) + { + crc = await _reader + .ReadUInt32Async(_cancellationToken) + .ConfigureAwait(false); + } + lastEntryHeader.Crc = crc; + + // The DataDescriptor can be either 64bit or 32bit + var compressedSize = await _reader + .ReadUInt32Async(_cancellationToken) + .ConfigureAwait(false); + var uncompressedSize = await _reader + .ReadUInt32Async(_cancellationToken) + .ConfigureAwait(false); + + // Check if we have header or 64bit DataDescriptor + var testHeader = !(headerBytes == 0x04034b50 || headerBytes == 0x02014b50); + + var test64Bit = ((long)uncompressedSize << 32) | compressedSize; + if (test64Bit == lastEntryHeader.CompressedSize && testHeader) + { + lastEntryHeader.UncompressedSize = + ( + (long) + await _reader + .ReadUInt32Async(_cancellationToken) + .ConfigureAwait(false) << 32 + ) | headerBytes; + headerBytes = await _reader + .ReadUInt32Async(_cancellationToken) + .ConfigureAwait(false); + } + else + { + lastEntryHeader.UncompressedSize = uncompressedSize; + } + + if (pos.HasValue) + { + lastEntryHeader.DataStartPosition = pos - lastEntryHeader.CompressedSize; + + // 4 = First 4 bytes of the entry header (i.e. 50 4B 03 04) + _rewindableStream.Position = pos.Value + 4; + } + } + else + { + headerBytes = await _reader + .ReadUInt32Async(_cancellationToken) + .ConfigureAwait(false); + } + + _headerFactory.LastEntryHeader = null; + var header = await _headerFactory + .ReadHeaderAsyncInternal(headerBytes, _reader) + .ConfigureAwait(false); + if (header is null) + { + _completed = true; + return false; + } + + //entry could be zero bytes so we need to know that. + if (header.ZipHeaderType == ZipHeaderType.LocalEntry) + { + var localHeader = (LocalEntryHeader)header; + var directoryHeader = _headerFactory._entries?.FirstOrDefault(entry => + entry.Key == localHeader.Name + && localHeader.CompressedSize == 0 + && localHeader.UncompressedSize == 0 + && localHeader.Crc == 0 + && localHeader.IsDirectory == false + ); + + if (directoryHeader != null) + { + localHeader.UncompressedSize = directoryHeader.Size; + localHeader.CompressedSize = directoryHeader.CompressedSize; + localHeader.Crc = (uint)directoryHeader.Crc; + } + + // If we have CompressedSize, there is data to be read + if (localHeader.CompressedSize > 0) + { + header.HasData = true; + } // Check if zip is streaming ( Length is 0 and is declared in PostDataDescriptor ) + else if (localHeader.Flags.HasFlag(HeaderFlags.UsePostDataDescriptor)) + { + var nextHeaderBytes = await _reader + .ReadUInt32Async(_cancellationToken) + .ConfigureAwait(false); + ((IStreamStack)_rewindableStream).Rewind(sizeof(uint)); + + // Check if next data is PostDataDescriptor, streamed file with 0 length + header.HasData = !IsHeader(nextHeaderBytes); + } + else // We are not streaming and compressed size is 0, we have no data + { + header.HasData = false; + } + } + + _current = header; + return true; + } + } + + public ValueTask DisposeAsync() + { + Dispose(); + return default; + } + + /// + /// Disposes the underlying reader (without closing the archive stream). + /// + public void Dispose() + { + _reader.Dispose(); + } + + /// + /// Ensures the stream is a so header parsing can use rewind/buffer helpers. + /// + private static SharpCompressStream EnsureSharpCompressStream(Stream stream) + { + if (stream is SharpCompressStream sharpCompressStream) + { + return sharpCompressStream; + } + + // Ensure the stream is already a SharpCompressStream so the buffer/size is set. + // The original code wrapped this with RewindableStream; use SharpCompressStream so we can get the buffer size. + if (stream is SourceStream src) + { + return new SharpCompressStream( + stream, + src.ReaderOptions.LeaveStreamOpen, + bufferSize: src.ReaderOptions.BufferSize + ); + } + + throw new ArgumentException("Stream must be a SharpCompressStream", nameof(stream)); + } + } } diff --git a/src/SharpCompress/Common/Zip/ZipHeaderFactory.cs b/src/SharpCompress/Common/Zip/ZipHeaderFactory.cs index e085a0de..d25f1025 100644 --- a/src/SharpCompress/Common/Zip/ZipHeaderFactory.cs +++ b/src/SharpCompress/Common/Zip/ZipHeaderFactory.cs @@ -2,6 +2,7 @@ using System; using System.IO; using System.Linq; using System.Threading.Tasks; +using SharpCompress; using SharpCompress.Common.Zip.Headers; using SharpCompress.IO; @@ -47,7 +48,7 @@ internal class ZipHeaderFactory { var entryHeader = new LocalEntryHeader(_archiveEncoding); await entryHeader.Read(reader); - LoadHeader(entryHeader, reader.BaseStream); + await LoadHeaderAsync(entryHeader, reader.BaseStream).ConfigureAwait(false); _lastEntryHeader = entryHeader; return entryHeader; @@ -282,4 +283,82 @@ internal class ZipHeaderFactory //} } + + /// + /// Loads encryption metadata and stream positioning for a header using async reads where needed. + /// + private async ValueTask LoadHeaderAsync(ZipFileEntry entryHeader, Stream stream) + { + if (FlagUtility.HasFlag(entryHeader.Flags, HeaderFlags.Encrypted)) + { + if ( + !entryHeader.IsDirectory + && entryHeader.CompressedSize == 0 + && FlagUtility.HasFlag(entryHeader.Flags, HeaderFlags.UsePostDataDescriptor) + ) + { + throw new NotSupportedException( + "SharpCompress cannot currently read non-seekable Zip Streams with encrypted data that has been written in a non-seekable manner." + ); + } + + if (_password is null) + { + throw new CryptographicException("No password supplied for encrypted zip."); + } + + entryHeader.Password = _password; + + if (entryHeader.CompressionMethod == ZipCompressionMethod.WinzipAes) + { + var data = entryHeader.Extra.SingleOrDefault(x => + x.Type == ExtraDataType.WinZipAes + ); + if (data != null) + { + var keySize = (WinzipAesKeySize)data.DataBytes[4]; + + var salt = new byte[WinzipAesEncryptionData.KeyLengthInBytes(keySize) / 2]; + var passwordVerifyValue = new byte[2]; + await stream.ReadExactAsync(salt, 0, salt.Length).ConfigureAwait(false); + await stream.ReadExactAsync(passwordVerifyValue, 0, 2).ConfigureAwait(false); + + entryHeader.WinzipAesEncryptionData = new WinzipAesEncryptionData( + keySize, + salt, + passwordVerifyValue, + _password + ); + + entryHeader.CompressedSize -= (uint)(salt.Length + 2); + } + } + } + + if (entryHeader.IsDirectory) + { + return; + } + + switch (_mode) + { + case StreamingMode.Seekable: + { + entryHeader.DataStartPosition = stream.Position; + stream.Position += entryHeader.CompressedSize; + break; + } + + case StreamingMode.Streaming: + { + entryHeader.PackedStream = stream; + break; + } + + default: + { + throw new InvalidFormatException("Invalid StreamingMode"); + } + } + } } diff --git a/src/SharpCompress/Polyfills/AsyncEnumerableExtensions.cs b/src/SharpCompress/Polyfills/AsyncEnumerableExtensions.cs new file mode 100644 index 00000000..f1379cbb --- /dev/null +++ b/src/SharpCompress/Polyfills/AsyncEnumerableExtensions.cs @@ -0,0 +1,32 @@ +using System; +using System.Collections.Generic; +using System.Threading.Tasks; + +namespace SharpCompress; + +public static class AsyncEnumerableExtensions +{ + extension(IAsyncEnumerable source) + { + public async IAsyncEnumerable Where(Func predicate) + { + await foreach (var item in source) + { + if (predicate(item)) + { + yield return item; + } + } + } + + public async ValueTask FirstOrDefaultAsync() + { + await foreach (var item in source) + { + return item; // Returns the very first item found + } + + return default; // Returns null/default if the stream is empty + } + } +} diff --git a/src/SharpCompress/Readers/AbstractReader.cs b/src/SharpCompress/Readers/AbstractReader.cs index bd35a6e1..01340f24 100644 --- a/src/SharpCompress/Readers/AbstractReader.cs +++ b/src/SharpCompress/Readers/AbstractReader.cs @@ -18,6 +18,11 @@ public abstract class AbstractReader : IReader { private bool _completed; private IEnumerator? _entriesForCurrentReadStream; + + /// + /// Holds the async entry enumerator when the reader is operating in an async-only mode. + /// + private IAsyncEnumerator? _asyncEntriesForCurrentReadStream; private bool _wroteCurrentEntry; internal AbstractReader(ReaderOptions options, ArchiveType archiveType) @@ -36,15 +41,22 @@ public abstract class AbstractReader : IReader public abstract TVolume? Volume { get; } /// - /// Current file entry + /// Current file entry (from either sync or async enumeration). /// - public TEntry Entry => _entriesForCurrentReadStream.NotNull().Current; + public TEntry Entry => + _entriesForCurrentReadStream?.Current + ?? _asyncEntriesForCurrentReadStream?.Current + ?? throw new InvalidOperationException("No current entry is available."); #region IDisposable Members public virtual void Dispose() { _entriesForCurrentReadStream?.Dispose(); + if (_asyncEntriesForCurrentReadStream is IDisposable disposable) + { + disposable.Dispose(); + } Volume?.Dispose(); } @@ -67,6 +79,12 @@ public abstract class AbstractReader : IReader public bool MoveToNextEntry() { + if (_asyncEntriesForCurrentReadStream is not null) + { + throw new InvalidOperationException( + $"{nameof(MoveToNextEntry)} cannot be used after {nameof(MoveToNextEntryAsync)} has been used." + ); + } if (_completed) { return false; @@ -102,16 +120,17 @@ public abstract class AbstractReader : IReader { throw new ReaderCancelledException("Reader has been cancelled."); } - if (_entriesForCurrentReadStream is null) + if (_entriesForCurrentReadStream is null && _asyncEntriesForCurrentReadStream is null) { - return LoadStreamForReading(RequestInitialStream()); + return await LoadStreamForReadingAsync(RequestInitialStream(), cancellationToken) + .ConfigureAwait(false); } if (!_wroteCurrentEntry) { await SkipEntryAsync(cancellationToken).ConfigureAwait(false); } _wroteCurrentEntry = false; - if (NextEntryForCurrentStream()) + if (await NextEntryForCurrentStreamAsync(cancellationToken).ConfigureAwait(false)) { return true; } @@ -121,6 +140,12 @@ public abstract class AbstractReader : IReader protected bool LoadStreamForReading(Stream stream) { + if (_asyncEntriesForCurrentReadStream is not null) + { + throw new InvalidOperationException( + $"{nameof(LoadStreamForReading)} cannot be used after {nameof(LoadStreamForReadingAsync)} has been used." + ); + } _entriesForCurrentReadStream?.Dispose(); if (stream is null || !stream.CanRead) { @@ -134,14 +159,69 @@ public abstract class AbstractReader : IReader return _entriesForCurrentReadStream.MoveNext(); } + /// + /// Loads the stream for reading entries asynchronously, using an async entry enumerator when available. + /// + protected async Task LoadStreamForReadingAsync( + Stream stream, + CancellationToken cancellationToken = default + ) + { + // Always reset the previous async enumerator so that a new stream can be loaded cleanly. + if (_asyncEntriesForCurrentReadStream is IDisposable disposable) + { + disposable.Dispose(); + } + _asyncEntriesForCurrentReadStream = null; + + if (stream is null || !stream.CanRead) + { + throw new MultipartStreamRequiredException( + "File is split into multiple archives: '" + + Entry.Key + + "'. A new readable stream is required. Use Cancel if it was intended." + ); + } + + var entriesAsync = GetEntriesAsync(stream); + if (entriesAsync is null) + { + _entriesForCurrentReadStream = GetEntries(stream).GetEnumerator(); + return _entriesForCurrentReadStream.MoveNext(); + } + + _asyncEntriesForCurrentReadStream = entriesAsync.GetAsyncEnumerator(cancellationToken); + return await _asyncEntriesForCurrentReadStream.MoveNextAsync().ConfigureAwait(false); + } + protected virtual Stream RequestInitialStream() => Volume.NotNull("Volume isn't loaded.").Stream; internal virtual bool NextEntryForCurrentStream() => _entriesForCurrentReadStream.NotNull().MoveNext(); + /// + /// Moves the current async enumerator to the next entry. + /// + internal virtual ValueTask NextEntryForCurrentStreamAsync( + CancellationToken cancellationToken = default + ) + { + if (_asyncEntriesForCurrentReadStream is not null) + { + return _asyncEntriesForCurrentReadStream.MoveNextAsync(); + } + + return new ValueTask(NextEntryForCurrentStream()); + } + protected abstract IEnumerable GetEntries(Stream stream); + /// + /// Optionally returns an async entry sequence for formats that support true async header parsing. + /// + protected virtual IAsyncEnumerable? GetEntriesAsync(Stream stream) => null; + #region Entry Skip/Write private void SkipEntry() diff --git a/src/SharpCompress/Readers/Zip/ZipReader.cs b/src/SharpCompress/Readers/Zip/ZipReader.cs index 3a257845..673d6ec7 100644 --- a/src/SharpCompress/Readers/Zip/ZipReader.cs +++ b/src/SharpCompress/Readers/Zip/ZipReader.cs @@ -1,5 +1,8 @@ +using System; using System.Collections.Generic; using System.IO; +using System.Threading; +using System.Threading.Tasks; using SharpCompress.Common; using SharpCompress.Common.Zip; using SharpCompress.Common.Zip.Headers; @@ -91,4 +94,102 @@ public class ZipReader : AbstractReader } } } + + /// + /// Returns entries asynchronously for streams that only support async reads. + /// + protected override IAsyncEnumerable? GetEntriesAsync(Stream stream) => + new ZipEntryAsyncEnumerable(_headerFactory, stream); + + /// + /// Adapts an async header sequence into an async entry sequence. + /// + private sealed class ZipEntryAsyncEnumerable : IAsyncEnumerable + { + private readonly StreamingZipHeaderFactory _headerFactory; + private readonly Stream _stream; + + public ZipEntryAsyncEnumerable(StreamingZipHeaderFactory headerFactory, Stream stream) + { + _headerFactory = headerFactory; + _stream = stream; + } + + public IAsyncEnumerator GetAsyncEnumerator( + CancellationToken cancellationToken = default + ) => new ZipEntryAsyncEnumerator(_headerFactory, _stream, cancellationToken); + } + + /// + /// Yields entries from streaming ZIP headers without requiring synchronous stream reads. + /// + private sealed class ZipEntryAsyncEnumerator : IAsyncEnumerator, IDisposable + { + private readonly Stream _stream; + private readonly IAsyncEnumerator _headerEnumerator; + private ZipEntry? _current; + + public ZipEntryAsyncEnumerator( + StreamingZipHeaderFactory headerFactory, + Stream stream, + CancellationToken cancellationToken + ) + { + _stream = stream; + _headerEnumerator = headerFactory + .ReadStreamHeaderAsync(stream) + .GetAsyncEnumerator(cancellationToken); + } + + public ZipEntry Current => + _current ?? throw new InvalidOperationException("No current entry is available."); + + /// + /// Advances to the next non-directory entry-relevant header and materializes a . + /// + public async ValueTask MoveNextAsync() + { + while (await _headerEnumerator.MoveNextAsync().ConfigureAwait(false)) + { + var header = _headerEnumerator.Current; + switch (header.ZipHeaderType) + { + case ZipHeaderType.LocalEntry: + _current = new ZipEntry( + new StreamingZipFilePart((LocalEntryHeader)header, _stream) + ); + return true; + case ZipHeaderType.DirectoryEntry: + // DirectoryEntry headers are intentionally skipped in streaming mode. + break; + case ZipHeaderType.DirectoryEnd: + _current = null; + return false; + } + } + + _current = null; + return false; + } + + /// + /// Disposes the underlying header enumerator. + /// + public ValueTask DisposeAsync() + { + Dispose(); + return default; + } + + /// + /// Disposes the underlying header enumerator. + /// + public void Dispose() + { + if (_headerEnumerator is IDisposable disposable) + { + disposable.Dispose(); + } + } + } }