Skip to content

Commit f9e5248

Browse files
baywetbaywet
committed
fix: reading streams in an asp.net context would cause async exceptions
1 parent 2436d73 commit f9e5248

File tree

2 files changed

+275
-45
lines changed

2 files changed

+275
-45
lines changed

‎src/Microsoft.OpenApi/Reader/OpenApiModelFactory.cs‎

Lines changed: 73 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
// Licensed under the MIT license.
33

44
using System;
5+
using System.Diagnostics.CodeAnalysis;
56
using System.IO;
67
using System.Linq;
78
using System.Security;
@@ -362,65 +363,98 @@ private static string InspectInputFormat(string input)
362363
return input.StartsWith("{", StringComparison.OrdinalIgnoreCase) || input.StartsWith("[", StringComparison.OrdinalIgnoreCase) ? OpenApiConstants.Json : OpenApiConstants.Yaml;
363364
}
364365

365-
private static string InspectStreamFormat(Stream stream)
366+
/// <summary>
367+
/// Reads the initial bytes of the stream to determine if it is JSON or YAML.
368+
/// </summary>
369+
/// <remarks>
370+
/// It is important NOT TO change the stream type from MemoryStream.
371+
/// In Asp.Net core 3.0+ we could get passed a stream from a request or response body.
372+
/// In such case, we CAN'T use the ReadByte method as it throws NotSupportedException.
373+
/// Therefore, we need to ensure that the stream is a MemoryStream before calling this method.
374+
/// Maintaining this type ensures there won't be any unforeseen wrong usage of the method.
375+
/// </remarks>
376+
/// <param name="stream">The stream to inspect</param>
377+
/// <returns>The format of the stream.</returns>
378+
private static string InspectStreamFormat(MemoryStream stream)
379+
{
380+
return TryInspectStreamFormat(stream, out var format) ? format! : throw new InvalidOperationException("Could not determine the format of the stream.");
381+
}
382+
private static bool TryInspectStreamFormat(Stream stream, out string? format)
366383
{
367384
#if NET6_0_OR_GREATER
368385
ArgumentNullException.ThrowIfNull(stream);
369386
#else
370387
if (stream is null) throw new ArgumentNullException(nameof(stream));
371388
#endif
372389

373-
long initialPosition = stream.Position;
374-
int firstByte = stream.ReadByte();
375-
376-
// Skip whitespace if present and read the next non-whitespace byte
377-
if (char.IsWhiteSpace((char)firstByte))
390+
try
378391
{
379-
firstByte = stream.ReadByte();
380-
}
392+
var initialPosition = stream.Position;
393+
var firstByte = (char)stream.ReadByte();
394+
395+
// Skip whitespace if present and read the next non-whitespace byte
396+
if (char.IsWhiteSpace(firstByte))
397+
{
398+
firstByte = (char)stream.ReadByte();
399+
}
381400

382-
stream.Position = initialPosition; // Reset the stream position to the beginning
401+
stream.Position = initialPosition; // Reset the stream position to the beginning
383402

384-
char firstChar = (char)firstByte;
385-
return firstChar switch
403+
format = firstByte switch
404+
{
405+
'{' or '[' => OpenApiConstants.Json, // If the first character is '{' or '[', assume JSON
406+
_ => OpenApiConstants.Yaml // Otherwise assume YAML
407+
};
408+
return true;
409+
}
410+
catch (NotSupportedException)
411+
{
412+
// https://github.com/dotnet/aspnetcore/blob/c9d0750396e1d319301255ba61842721ab72ab10/src/Servers/Kestrel/Core/src/Internal/Http/HttpResponseStream.cs#L40
413+
}
414+
#if NETSTANDARD2_1_OR_GREATER || NETCOREAPP || NET5_0_OR_GREATER
415+
catch (InvalidOperationException ex) when (ex.Message.Contains("AllowSynchronousIO", StringComparison.Ordinal))
416+
#else
417+
catch (InvalidOperationException ex) when (ex.Message.Contains("AllowSynchronousIO"))
418+
#endif
386419
{
387-
'{' or '[' => OpenApiConstants.Json, // If the first character is '{' or '[', assume JSON
388-
_ => OpenApiConstants.Yaml // Otherwise assume YAML
389-
};
420+
// https://github.com/dotnet/aspnetcore/blob/c9d0750396e1d319301255ba61842721ab72ab10/src/Servers/HttpSys/src/RequestProcessing/RequestStream.cs#L100-L108
421+
// https://github.com/dotnet/aspnetcore/blob/c9d0750396e1d319301255ba61842721ab72ab10/src/Servers/IIS/IIS/src/Core/HttpRequestStream.cs#L24-L30
422+
// https://github.com/dotnet/aspnetcore/blob/c9d0750396e1d319301255ba61842721ab72ab10/src/Servers/Kestrel/Core/src/Internal/Http/HttpRequestStream.cs#L54-L60
423+
}
424+
format = null;
425+
return false;
390426
}
391427

428+
private static async Task<MemoryStream> CopyToMemoryStreamAsync(Stream input, CancellationToken token)
429+
{
430+
var bufferStream = new MemoryStream();
431+
#if NETSTANDARD2_1_OR_GREATER || NETCOREAPP || NET5_0_OR_GREATER
432+
await input.CopyToAsync(bufferStream, token).ConfigureAwait(false);
433+
#else
434+
await input.CopyToAsync(bufferStream, 81920, token).ConfigureAwait(false);
435+
#endif
436+
bufferStream.Position = 0;
437+
return bufferStream;
438+
}
439+
392440
private static async Task<(Stream, string)> PrepareStreamForReadingAsync(Stream input, string? format, CancellationToken token = default)
393441
{
394442
Stream preparedStream = input;
395443

396-
if (!input.CanSeek)
444+
if (input is MemoryStream ms)
397445
{
398-
// Use a temporary buffer to read a small portion for format detection
399-
using var bufferStream = new MemoryStream();
400-
await input.CopyToAsync(bufferStream, 1024, token).ConfigureAwait(false);
401-
bufferStream.Position = 0;
402-
403-
// Inspect the format from the buffered portion
404-
format ??= InspectStreamFormat(bufferStream);
405-
406-
// we need to copy the stream to memory string we've already started reading it and can't reposition it
407-
preparedStream = new MemoryStream();
408-
bufferStream.Position = 0;
409-
await bufferStream.CopyToAsync(preparedStream, 81920, token).ConfigureAwait(false); // Copy buffered portion
410-
await input.CopyToAsync(preparedStream, 81920, token).ConfigureAwait(false); // Copy remaining data
411-
preparedStream.Position = 0;
446+
format ??= InspectStreamFormat(ms);
412447
}
413-
else
448+
else if (!input.CanSeek)
414449
{
415-
format ??= InspectStreamFormat(input);
416-
417-
if (!format.Equals(OpenApiConstants.Json, StringComparison.OrdinalIgnoreCase))
418-
{
419-
// Buffer stream for non-JSON formats (e.g., YAML) since they require synchronous reading
420-
preparedStream = new MemoryStream();
421-
await input.CopyToAsync(preparedStream, 81920, token).ConfigureAwait(false);
422-
preparedStream.Position = 0;
423-
}
450+
// Copy to a MemoryStream to enable seeking and perform format inspection
451+
var bufferStream = await CopyToMemoryStreamAsync(input, token).ConfigureAwait(false);
452+
return await PrepareStreamForReadingAsync(bufferStream, format, token).ConfigureAwait(false);
453+
}
454+
else if (!TryInspectStreamFormat(input, out format!))
455+
{
456+
var bufferStream = await CopyToMemoryStreamAsync(input, token).ConfigureAwait(false);
457+
return await PrepareStreamForReadingAsync(bufferStream, format, token).ConfigureAwait(false);
424458
}
425459

426460
return (preparedStream, format);

‎test/Microsoft.OpenApi.Tests/Reader/OpenApiModelFactoryTests.cs‎

Lines changed: 202 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -120,11 +120,7 @@ await File.WriteAllTextAsync(tempFilePathReferrer,
120120
Assert.NotNull(readResult.Document.Components);
121121
Assert.Equal(baseUri, readResult.Document.BaseUri);
122122
}
123-
[Fact]
124-
public async Task CanLoadANonSeekableStream()
125-
{
126-
// Given
127-
var documentJson =
123+
private readonly string documentJson =
128124
"""
129125
{
130126
"openapi": "3.1.0",
@@ -135,6 +131,18 @@ public async Task CanLoadANonSeekableStream()
135131
"paths": {}
136132
}
137133
""";
134+
private readonly string documentYaml =
135+
"""
136+
openapi: 3.1.0
137+
info:
138+
title: Sample API
139+
version: 1.0.0
140+
paths: {}
141+
""";
142+
[Fact]
143+
public async Task CanLoadANonSeekableStreamInJsonAndDetectFormat()
144+
{
145+
// Given
138146
using var memoryStream = new MemoryStream(System.Text.Encoding.UTF8.GetBytes(documentJson));
139147
using var nonSeekableStream = new NonSeekableStream(memoryStream);
140148

@@ -146,6 +154,194 @@ public async Task CanLoadANonSeekableStream()
146154
Assert.Equal("Sample API", document.Info.Title);
147155
}
148156

157+
[Fact]
158+
public async Task CanLoadANonSeekableStreamInYamlAndDetectFormat()
159+
{
160+
// Given
161+
using var memoryStream = new MemoryStream(System.Text.Encoding.UTF8.GetBytes(documentYaml));
162+
using var nonSeekableStream = new NonSeekableStream(memoryStream);
163+
var settings = new OpenApiReaderSettings();
164+
settings.AddYamlReader();
165+
166+
// When
167+
var (document, _) = await OpenApiDocument.LoadAsync(nonSeekableStream, settings: settings);
168+
169+
// Then
170+
Assert.NotNull(document);
171+
Assert.Equal("Sample API", document.Info.Title);
172+
}
173+
174+
[Fact]
175+
public async Task CanLoadAnAsyncOnlyStreamInJsonAndDetectFormat()
176+
{
177+
// Given
178+
await using var memoryStream = new MemoryStream(System.Text.Encoding.UTF8.GetBytes(documentJson));
179+
await using var nonSeekableStream = new AsyncOnlyStream(memoryStream);
180+
181+
// When
182+
var (document, _) = await OpenApiDocument.LoadAsync(nonSeekableStream);
183+
184+
// Then
185+
Assert.NotNull(document);
186+
Assert.Equal("Sample API", document.Info.Title);
187+
}
188+
189+
[Fact]
190+
public async Task CanLoadAnAsyncOnlyStreamInYamlAndDetectFormat()
191+
{
192+
// Given
193+
await using var memoryStream = new MemoryStream(System.Text.Encoding.UTF8.GetBytes(documentYaml));
194+
await using var nonSeekableStream = new AsyncOnlyStream(memoryStream);
195+
var settings = new OpenApiReaderSettings();
196+
settings.AddYamlReader();
197+
198+
// When
199+
var (document, _) = await OpenApiDocument.LoadAsync(nonSeekableStream, settings: settings);
200+
201+
// Then
202+
Assert.NotNull(document);
203+
Assert.Equal("Sample API", document.Info.Title);
204+
}
205+
206+
public sealed class AsyncOnlyStream : Stream
207+
{
208+
private readonly Stream _innerStream;
209+
public AsyncOnlyStream(Stream stream) : base()
210+
{
211+
_innerStream = stream;
212+
}
213+
public override bool CanSeek => _innerStream.CanSeek;
214+
215+
public override long Position { get => _innerStream.Position; set => throw new NotSupportedException("Blocking operations are not supported"); }
216+
217+
public override bool CanRead => _innerStream.CanRead;
218+
219+
public override bool CanWrite => _innerStream.CanWrite;
220+
221+
public override long Length => _innerStream.Length;
222+
public override IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback callback, object state)
223+
{
224+
return _innerStream.BeginRead(buffer, offset, count, callback, state);
225+
}
226+
227+
public override void Flush()
228+
{
229+
throw new NotSupportedException("Blocking operations are not supported.");
230+
}
231+
232+
public override int Read(byte[] buffer, int offset, int count)
233+
{
234+
throw new NotSupportedException("Blocking operations are not supported.");
235+
}
236+
237+
public override long Seek(long offset, SeekOrigin origin)
238+
{
239+
throw new NotSupportedException("Blocking operations are not supported.");
240+
}
241+
242+
public override void SetLength(long value)
243+
{
244+
_innerStream.SetLength(value);
245+
}
246+
247+
public override void Write(byte[] buffer, int offset, int count)
248+
{
249+
throw new NotSupportedException("Blocking operations are not supported.");
250+
}
251+
protected override void Dispose(bool disposing)
252+
{
253+
throw new NotSupportedException("Blocking operations are not supported.");
254+
}
255+
256+
public override async ValueTask DisposeAsync()
257+
{
258+
await _innerStream.DisposeAsync();
259+
await base.DisposeAsync();
260+
}
261+
262+
public override Task CopyToAsync(Stream destination, int bufferSize, CancellationToken cancellationToken)
263+
{
264+
return _innerStream.CopyToAsync(destination, bufferSize, cancellationToken);
265+
}
266+
267+
public override bool CanTimeout => _innerStream.CanTimeout;
268+
269+
public override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, AsyncCallback callback, object state)
270+
{
271+
return _innerStream.BeginWrite(buffer, offset, count, callback, state);
272+
}
273+
274+
public override void CopyTo(Stream destination, int bufferSize)
275+
{
276+
throw new NotSupportedException("Blocking operations are not supported.");
277+
}
278+
279+
public override void Close()
280+
{
281+
_innerStream.Close();
282+
}
283+
284+
public override int EndRead(IAsyncResult asyncResult)
285+
{
286+
return _innerStream.EndRead(asyncResult);
287+
}
288+
289+
public override void EndWrite(IAsyncResult asyncResult)
290+
{
291+
_innerStream.EndWrite(asyncResult);
292+
}
293+
294+
public override int ReadByte()
295+
{
296+
throw new NotSupportedException("Blocking operations are not supported.");
297+
}
298+
299+
public override void WriteByte(byte value)
300+
{
301+
throw new NotSupportedException("Blocking operations are not supported.");
302+
}
303+
304+
public override Task FlushAsync(CancellationToken cancellationToken)
305+
{
306+
return _innerStream.FlushAsync(cancellationToken);
307+
}
308+
309+
public override int Read(Span<byte> buffer)
310+
{
311+
throw new NotSupportedException("Blocking operations are not supported.");
312+
}
313+
314+
public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
315+
{
316+
return _innerStream.ReadAsync(buffer, offset, count, cancellationToken);
317+
}
318+
319+
public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
320+
{
321+
return _innerStream.ReadAsync(buffer, cancellationToken);
322+
}
323+
324+
public override int ReadTimeout { get => _innerStream.ReadTimeout; set => _innerStream.ReadTimeout = value; }
325+
326+
public override void Write(ReadOnlySpan<byte> buffer)
327+
{
328+
throw new NotSupportedException("Blocking operations are not supported.");
329+
}
330+
331+
public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
332+
{
333+
return _innerStream.WriteAsync(buffer, offset, count, cancellationToken);
334+
}
335+
336+
public override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
337+
{
338+
return _innerStream.WriteAsync(buffer, cancellationToken);
339+
}
340+
341+
public override int WriteTimeout { get => _innerStream.WriteTimeout; set => _innerStream.WriteTimeout = value; }
342+
343+
}
344+
149345
public sealed class NonSeekableStream : Stream
150346
{
151347
private readonly Stream _innerStream;
@@ -155,7 +351,7 @@ public NonSeekableStream(Stream stream) : base()
155351
}
156352
public override bool CanSeek => false;
157353

158-
public override long Position { get => _innerStream.Position; set => throw new InvalidOperationException("Seeking is not supported."); }
354+
public override long Position { get => _innerStream.Position; set => throw new NotSupportedException("Seeking is not supported."); }
159355

160356
public override bool CanRead => _innerStream.CanRead;
161357

0 commit comments

Comments
 (0)