Create connectionIds using RNGCrypto (#1606)

benchmarks/Microsoft.AspNetCore.SignalR.Microbenchmarks/DefaultHubDispatcherBenchmark.cs

@@ -39,9 +39,8 @@ namespace Microsoft.AspNetCore.SignalR.Microbenchmarks
                 new HubContext<TestHub>(new DefaultHubLifetimeManager<TestHub>()),
                 new Logger<DefaultHubDispatcher<TestHub>>(NullLoggerFactory.Instance));
 
-            var options = new PipeOptions();
+            var pair = DuplexPipe.CreateConnectionPair(PipeOptions.Default, PipeOptions.Default);
-            var pair = DuplexPipe.CreateConnectionPair(options, options);
+            var connection = new Sockets.DefaultConnectionContext(Guid.NewGuid().ToString(), pair.Application, pair.Transport);
-            var connection = new Sockets.DefaultConnectionContext(Guid.NewGuid().ToString(), pair.Transport, pair.Application);
 
             _connectionContext = new NoErrorHubConnectionContext(connection, TimeSpan.Zero, NullLoggerFactory.Instance);
 

build/dependencies.props

@@ -51,6 +51,7 @@
     <MicrosoftExtensionsObjectMethodExecutorSourcesPackageVersion>2.1.0-preview2-30355</MicrosoftExtensionsObjectMethodExecutorSourcesPackageVersion>
     <MicrosoftExtensionsOptionsPackageVersion>2.1.0-preview2-30355</MicrosoftExtensionsOptionsPackageVersion>
     <MicrosoftExtensionsSecurityHelperSourcesPackageVersion>2.1.0-preview2-30355</MicrosoftExtensionsSecurityHelperSourcesPackageVersion>
+    <MicrosoftExtensionsWebEncodersSourcesPackageVersion>2.1.0-preview2-30355</MicrosoftExtensionsWebEncodersSourcesPackageVersion>
     <MicrosoftExtensionsValueStopwatchSourcesPackageVersion>2.1.0-preview2-30355</MicrosoftExtensionsValueStopwatchSourcesPackageVersion>
     <MicrosoftNETCoreApp20PackageVersion>2.0.0</MicrosoftNETCoreApp20PackageVersion>
     <MicrosoftNETCoreApp21PackageVersion>2.1.0-preview2-26314-02</MicrosoftNETCoreApp21PackageVersion>

src/Microsoft.AspNetCore.Sockets.Abstractions/DefaultConnectionContext.cs

@@ -22,16 +22,20 @@ namespace Microsoft.AspNetCore.Sockets
                                             IConnectionHeartbeatFeature,
                                             ITransferFormatFeature
     {
-        private List<(Action<object> handler, object state)> _heartbeatHandlers = new List<(Action<object> handler, object state)>();
+        private object _heartbeatLock = new object();
+        private List<(Action<object> handler, object state)> _heartbeatHandlers;
 
         // This tcs exists so that multiple calls to DisposeAsync all wait asynchronously
         // on the same task
         private TaskCompletionSource<object> _disposeTcs = new TaskCompletionSource<object>();
 
-        public DefaultConnectionContext(string id, IDuplexPipe transport, IDuplexPipe application)
+        /// <summary>
+        /// Creates the DefaultConnectionContext without Pipes to avoid upfront allocations.
+        /// The caller is expected to set the <see cref="Transport"/> and <see cref="Application"/> pipes manually.
+        /// </summary>
+        /// <param name="id"></param>
+        public DefaultConnectionContext(string id)
         {
-            Transport = transport;
-            Application = application;
             ConnectionId = id;
             LastSeenUtc = DateTime.UtcNow;
 
@@ -50,6 +54,13 @@ namespace Microsoft.AspNetCore.Sockets
             Features.Set<ITransferFormatFeature>(this);
         }
 
+        public DefaultConnectionContext(string id, IDuplexPipe transport, IDuplexPipe application)
+            : this(id)
+        {
+            Transport = transport;
+            Application = application;
+        }
+
         public CancellationTokenSource Cancellation { get; set; }
 
         public SemaphoreSlim Lock { get; } = new SemaphoreSlim(1, 1);
@@ -80,16 +91,25 @@ namespace Microsoft.AspNetCore.Sockets
 
         public void OnHeartbeat(Action<object> action, object state)
         {
-            lock (_heartbeatHandlers)
+            lock (_heartbeatLock)
             {
+                if (_heartbeatHandlers == null)
+                {
+                    _heartbeatHandlers = new List<(Action<object> handler, object state)>();
+                }
                 _heartbeatHandlers.Add((action, state));
             }
         }
 
         public void TickHeartbeat()
         {
-            lock (_heartbeatHandlers)
+            lock (_heartbeatLock)
             {
+                if (_heartbeatHandlers == null)
+                {
+                    return;
+                }
+
                 foreach (var (handler, state) in _heartbeatHandlers)
                 {
                     handler(state);

src/Microsoft.AspNetCore.Sockets.Http/ConnectionManager.cs

@@ -2,12 +2,14 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System;
+using System.Buffers.Text;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.IO;
 using System.IO.Pipelines;
 using System.Net.WebSockets;
+using System.Security.Cryptography;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Hosting;
@@ -22,6 +24,8 @@ namespace Microsoft.AspNetCore.Sockets
         // TODO: Consider making this configurable? At least for testing?
         private static readonly TimeSpan _heartbeatTickRate = TimeSpan.FromSeconds(1);
 
+        private static readonly RNGCryptoServiceProvider _keyGenerator = new RNGCryptoServiceProvider();
+
         private readonly ConcurrentDictionary<string, (DefaultConnectionContext Connection, ValueStopwatch Timer)> _connections = new ConcurrentDictionary<string, (DefaultConnectionContext Connection, ValueStopwatch Timer)>();
         private Timer _timer;
         private readonly ILogger<ConnectionManager> _logger;
@@ -63,23 +67,31 @@ namespace Microsoft.AspNetCore.Sockets
             return false;
         }
 
-        public DefaultConnectionContext CreateConnection(PipeOptions transportPipeOptions, PipeOptions appPipeOptions)
+        /// <summary>
+        /// Creates a connection without Pipes setup to allow saving allocations until Pipes are needed.
+        /// </summary>
+        /// <returns></returns>
+        public DefaultConnectionContext CreateConnection()
         {
             var id = MakeNewConnectionId();
 
             _logger.CreatedNewConnection(id);
             var connectionTimer = SocketEventSource.Log.ConnectionStart(id);
-            var pair = DuplexPipe.CreateConnectionPair(transportPipeOptions, appPipeOptions);
 
-            var connection = new DefaultConnectionContext(id, pair.Application, pair.Transport);
+            var connection = new DefaultConnectionContext(id);
 
             _connections.TryAdd(id, (connection, connectionTimer));
             return connection;
         }
 
-        public DefaultConnectionContext CreateConnection()
+        public DefaultConnectionContext CreateConnection(PipeOptions transportPipeOptions, PipeOptions appPipeOptions)
         {
-            return CreateConnection(PipeOptions.Default, PipeOptions.Default);
+            var connection = CreateConnection();
+            var pair = DuplexPipe.CreateConnectionPair(transportPipeOptions, appPipeOptions);
+            connection.Application = pair.Transport;
+            connection.Transport = pair.Application;
+
+            return connection;
         }
 
         public void RemoveConnection(string id)
@@ -94,8 +106,12 @@ namespace Microsoft.AspNetCore.Sockets
 
         private static string MakeNewConnectionId()
         {
-            // TODO: We need to sign and encyrpt this
+            // TODO: Use Span when WebEncoders implements Span methods https://github.com/aspnet/Home/issues/2966
-            return Guid.NewGuid().ToString();
+            // 128 bit buffer / 8 bits per byte = 16 bytes
+            var buffer = new byte[16];
+            _keyGenerator.GetBytes(buffer);
+            // Generate the id with RNGCrypto because we want a cryptographically random id, which GUID is not
+            return WebEncoders.Base64UrlEncode(buffer);
         }
 
         private static void Scan(object state)

src/Microsoft.AspNetCore.Sockets.Http/HttpConnectionDispatcher.cs

@@ -49,7 +49,7 @@ namespace Microsoft.AspNetCore.Sockets
                 if (HttpMethods.IsPost(context.Request.Method))
                 {
                     // POST /{path}
-                    await ProcessSend(context);
+                    await ProcessSend(context, options);
                 }
                 else if (HttpMethods.IsGet(context.Request.Method))
                 {
@@ -99,7 +99,7 @@ namespace Microsoft.AspNetCore.Sockets
             if (headers.Accept?.Contains(new Net.Http.Headers.MediaTypeHeaderValue("text/event-stream")) == true)
             {
                 // Connection must already exist
-                var connection = await GetConnectionAsync(context);
+                var connection = await GetConnectionAsync(context, options);
                 if (connection == null)
                 {
                     // No such connection, GetConnection already set the response status code
@@ -149,7 +149,7 @@ namespace Microsoft.AspNetCore.Sockets
                 // GET /{path} maps to long polling
 
                 // Connection must already exist
-                var connection = await GetConnectionAsync(context);
+                var connection = await GetConnectionAsync(context, options);
                 if (connection == null)
                 {
                     // No such connection, GetConnection already set the response status code
@@ -361,7 +361,7 @@ namespace Microsoft.AspNetCore.Sockets
             context.Response.ContentType = "application/json";
 
             // Establish the connection
-            var connection = CreateConnectionInternal(options);
+            var connection = _manager.CreateConnection();
 
             // Set the Connection ID on the logging scope so that logs from now on will have the
             // Connection ID metadata set.
@@ -429,9 +429,9 @@ namespace Microsoft.AspNetCore.Sockets
 
         private static string GetConnectionId(HttpContext context) => context.Request.Query["id"];
 
-        private async Task ProcessSend(HttpContext context)
+        private async Task ProcessSend(HttpContext context, HttpSocketOptions options)
         {
-            var connection = await GetConnectionAsync(context);
+            var connection = await GetConnectionAsync(context, options);
             if (connection == null)
             {
                 // No such connection, GetConnection already set the response status code
@@ -505,7 +505,7 @@ namespace Microsoft.AspNetCore.Sockets
             return true;
         }
 
-        private async Task<DefaultConnectionContext> GetConnectionAsync(HttpContext context)
+        private async Task<DefaultConnectionContext> GetConnectionAsync(HttpContext context, HttpSocketOptions options)
         {
             var connectionId = GetConnectionId(context);
 
@@ -527,16 +527,25 @@ namespace Microsoft.AspNetCore.Sockets
                 return null;
             }
 
+            EnsureConnectionStateInternal(connection, options);
+
             return connection;
         }
 
-        private DefaultConnectionContext CreateConnectionInternal(HttpSocketOptions options)
+        private void EnsureConnectionStateInternal(DefaultConnectionContext connection, HttpSocketOptions options)
         {
-            var transportPipeOptions = new PipeOptions(pauseWriterThreshold: options.TransportMaxBufferSize, resumeWriterThreshold: options.TransportMaxBufferSize / 2, readerScheduler: PipeScheduler.ThreadPool, useSynchronizationContext: false);
+            // If the connection doesn't have a pipe yet then create one, we lazily create the pipe to save on allocations until the client actually connects
-            var appPipeOptions = new PipeOptions(pauseWriterThreshold: options.ApplicationMaxBufferSize, resumeWriterThreshold: options.ApplicationMaxBufferSize / 2, readerScheduler: PipeScheduler.ThreadPool, useSynchronizationContext: false);
+            if (connection.Transport == null)
-            return _manager.CreateConnection(transportPipeOptions, appPipeOptions);
+            {
+                var transportPipeOptions = new PipeOptions(pauseWriterThreshold: options.TransportMaxBufferSize, resumeWriterThreshold: options.TransportMaxBufferSize / 2, readerScheduler: PipeScheduler.ThreadPool, useSynchronizationContext: false);
+                var appPipeOptions = new PipeOptions(pauseWriterThreshold: options.ApplicationMaxBufferSize, resumeWriterThreshold: options.ApplicationMaxBufferSize / 2, readerScheduler: PipeScheduler.ThreadPool, useSynchronizationContext: false);
+                var pair = DuplexPipe.CreateConnectionPair(transportPipeOptions, appPipeOptions);
+                connection.Transport = pair.Application;
+                connection.Application = pair.Transport;
+            }
         }
 
+        // This is only used for WebSockets connections, which can connect directly without negotiating
         private async Task<DefaultConnectionContext> GetOrCreateConnectionAsync(HttpContext context, HttpSocketOptions options)
         {
             var connectionId = GetConnectionId(context);
@@ -545,7 +554,7 @@ namespace Microsoft.AspNetCore.Sockets
             // There's no connection id so this is a brand new connection
             if (StringValues.IsNullOrEmpty(connectionId))
             {
-                connection = CreateConnectionInternal(options);
+                connection = _manager.CreateConnection();
             }
             else if (!_manager.TryGetConnection(connectionId, out connection))
             {
@@ -555,6 +564,8 @@ namespace Microsoft.AspNetCore.Sockets
                 return null;
             }
 
+            EnsureConnectionStateInternal(connection, options);
+
             return connection;
         }
     }

src/Microsoft.AspNetCore.Sockets.Http/Microsoft.AspNetCore.Sockets.Http.csproj

@@ -10,7 +10,7 @@
     <Compile Include="..\Common\WebSocketExtensions.cs" Link="WebSocketExtensions.cs" />
     <Compile Include="..\Common\StreamExtensions.cs" Link="StreamExtensions.cs" />
   </ItemGroup>
-  
+
   <ItemGroup>
     <ProjectReference Include="..\Microsoft.AspNetCore.Sockets.Common.Http\Microsoft.AspNetCore.Sockets.Common.Http.csproj" />
     <ProjectReference Include="..\Microsoft.AspNetCore.Sockets.Abstractions\Microsoft.AspNetCore.Sockets.Abstractions.csproj" />
@@ -22,6 +22,7 @@
     <PackageReference Include="Microsoft.AspNetCore.Routing" Version="$(MicrosoftAspNetCoreRoutingPackageVersion)" />
     <PackageReference Include="Microsoft.AspNetCore.WebSockets" Version="$(MicrosoftAspNetCoreWebSocketsPackageVersion)" />
     <PackageReference Include="Microsoft.Extensions.SecurityHelper.Sources" PrivateAssets="All" Version="$(MicrosoftExtensionsSecurityHelperSourcesPackageVersion)" />
+    <PackageReference Include="Microsoft.Extensions.WebEncoders.Sources" Version="$(MicrosoftExtensionsWebEncodersSourcesPackageVersion)" PrivateAssets="All" />
     <PackageReference Include="Microsoft.Extensions.ValueStopwatch.Sources" Version="$(MicrosoftExtensionsValueStopwatchSourcesPackageVersion)" PrivateAssets="All" />
     <PackageReference Include="Newtonsoft.Json" Version="$(NewtonsoftJsonPackageVersion)" />
   </ItemGroup>

test/Microsoft.AspNetCore.Sockets.Tests/ConnectionManagerTests.cs

@@ -2,6 +2,7 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System;
+using System.IO.Pipelines;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Hosting;
 using Microsoft.Extensions.Logging;
@@ -22,8 +23,8 @@ namespace Microsoft.AspNetCore.Sockets.Tests
             Assert.Null(connection.ApplicationTask);
             Assert.Null(connection.TransportTask);
             Assert.Null(connection.Cancellation);
-            Assert.NotEqual(default(DateTime), connection.LastSeenUtc);
+            Assert.NotEqual(default, connection.LastSeenUtc);
-            Assert.NotNull(connection.Transport);
+            Assert.Null(connection.Transport);
         }
 
         [Fact]
@@ -42,7 +43,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
         public void AddNewConnection()
         {
             var connectionManager = CreateConnectionManager();
-            var connection = connectionManager.CreateConnection();
+            var connection = connectionManager.CreateConnection(PipeOptions.Default, PipeOptions.Default);
 
             var transport = connection.Transport;
 
@@ -58,7 +59,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
         public void RemoveConnection()
         {
             var connectionManager = CreateConnectionManager();
-            var connection = connectionManager.CreateConnection();
+            var connection = connectionManager.CreateConnection(PipeOptions.Default, PipeOptions.Default);
 
             var transport = connection.Transport;
 
@@ -77,7 +78,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
         public async Task CloseConnectionsEndsAllPendingConnections()
         {
             var connectionManager = CreateConnectionManager();
-            var connection = connectionManager.CreateConnection();
+            var connection = connectionManager.CreateConnection(PipeOptions.Default, PipeOptions.Default);
 
             connection.ApplicationTask = Task.Run(async () =>
             {
@@ -89,7 +90,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
                 }
                 finally
                 {
-                    connection.Transport.Input.AdvanceTo(result.Buffer.End);    
+                    connection.Transport.Input.AdvanceTo(result.Buffer.End);
                 }
             });
 
@@ -115,7 +116,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
         public async Task DisposingConnectionMultipleTimesWaitsOnConnectionClose()
         {
             var connectionManager = CreateConnectionManager();
-            var connection = connectionManager.CreateConnection();
+            var connection = connectionManager.CreateConnection(PipeOptions.Default, PipeOptions.Default);
             var tcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
 
             connection.ApplicationTask = tcs.Task;
@@ -135,7 +136,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
         public async Task DisposingConnectionMultipleGetsExceptionFromTransportOrApp()
         {
             var connectionManager = CreateConnectionManager();
-            var connection = connectionManager.CreateConnection();
+            var connection = connectionManager.CreateConnection(PipeOptions.Default, PipeOptions.Default);
             var tcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
 
             connection.ApplicationTask = tcs.Task;
@@ -159,7 +160,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
         public async Task DisposingConnectionMultipleGetsCancellation()
         {
             var connectionManager = CreateConnectionManager();
-            var connection = connectionManager.CreateConnection();
+            var connection = connectionManager.CreateConnection(PipeOptions.Default, PipeOptions.Default);
             var tcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
 
             connection.ApplicationTask = tcs.Task;
@@ -180,7 +181,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
         public async Task DisposeInactiveConnection()
         {
             var connectionManager = CreateConnectionManager();
-            var connection = connectionManager.CreateConnection();;
+            var connection = connectionManager.CreateConnection(PipeOptions.Default, PipeOptions.Default);
 
             Assert.NotNull(connection.ConnectionId);
             Assert.NotNull(connection.Transport);
@@ -209,7 +210,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
 
             appLifetime.Start();
 
-            var connection = connectionManager.CreateConnection();
+            var connection = connectionManager.CreateConnection(PipeOptions.Default, PipeOptions.Default);
 
             connection.Application.Output.OnReaderCompleted((error, state) =>
             {

test/Microsoft.AspNetCore.Sockets.Tests/HttpConnectionDispatcherTests.cs

@@ -77,7 +77,10 @@ namespace Microsoft.AspNetCore.Sockets.Tests
                 await dispatcher.ExecuteNegotiateAsync(context, httpSocketOptions);
                 var negotiateResponse = JsonConvert.DeserializeObject<JObject>(Encoding.UTF8.GetString(ms.ToArray()));
                 var connectionId = negotiateResponse.Value<string>("connectionId");
+                context.Request.QueryString = context.Request.QueryString.Add("id", connectionId);
                 Assert.True(manager.TryGetConnection(connectionId, out var connection));
+                // Fake actual connection after negotiate to populate the pipes on the connection
+                await dispatcher.ExecuteAsync(context, httpSocketOptions, c => Task.CompletedTask);
 
                 // This write should complete immediately but it exceeds the writer threshold
                 var writeTask = connection.Application.Output.WriteAsync(new byte[] { (byte)'b', (byte)'y', (byte)'t', (byte)'e', (byte)'s' });

test/Microsoft.AspNetCore.Sockets.Tests/LongPollingTests.cs

@@ -21,6 +21,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
         {
             var pair = DuplexPipe.CreateConnectionPair(PipeOptions.Default, PipeOptions.Default);
             var connection = new DefaultConnectionContext("foo", pair.Transport, pair.Application);
+
             var context = new DefaultHttpContext();
 
             var poll = new LongPollingTransport(CancellationToken.None, connection.Application.Input, connectionId: string.Empty, loggerFactory: new LoggerFactory());

test/Microsoft.AspNetCore.Sockets.Tests/ServerSentEventsTests.cs

@@ -4,7 +4,6 @@
 using System.IO;
 using System.IO.Pipelines;
 using System.Text;
-using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Http;
 using Microsoft.AspNetCore.Http.Features;
@@ -58,7 +57,6 @@ namespace Microsoft.AspNetCore.Sockets.Tests
             var connection = new DefaultConnectionContext("foo", pair.Transport, pair.Application);
             var context = new DefaultHttpContext();
 
-
             var ms = new MemoryStream();
             context.Response.Body = ms;
             var sse = new ServerSentEventsTransport(connection.Application.Input, connectionId: string.Empty, loggerFactory: new LoggerFactory());

test/Microsoft.AspNetCore.Sockets.Tests/WebSocketsTests.cs

@@ -320,6 +320,7 @@ namespace Microsoft.AspNetCore.Sockets.Tests
                         // We want to verify behavior without timeout affecting it
                         CloseTimeout = TimeSpan.FromSeconds(20)
                     };
+
                     var connectionContext = new DefaultConnectionContext(string.Empty, null, null);
                     var ws = new WebSocketsTransport(options, connection.Application, connectionContext, loggerFactory);