10887591743

Committed 16 Sep 2024 04:01PM UTC coverage: 81.524%. Remained the same

Build # 10887591743

Build Type

push

github

Committed by

web-flow

Commit Message

RNET-1175 Fix potential outgoing message corruption in managed websockets (#3670)

* Fix potential outgoing message corruption in managed websockets

* Update CHANGELOG.md

Run Details

2312 of 2991 branches covered (77.3%)

Branch coverage included in aggregate %.

0 of 1 new or added line in 1 file covered. (0.0%)

19 existing lines in 1 file now uncovered.

6923 of 8337 relevant lines covered (83.04%)

41987.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/Realm/Realm/Native/SyncSocketProvider.WebSocket.cs

////////////////////////////////////////////////////////////////////////////
//
// Copyright 2023 Realm Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
////////////////////////////////////////////////////////////////////////////

using System;
using System.Buffers;
using System.IO;
using System.Net.WebSockets;
using System.Text;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
using Realms.Logging;

namespace Realms.Native;

internal partial class SyncSocketProvider
{
    private class Socket : IDisposable
    {
        private readonly ClientWebSocket _webSocket;
        private readonly IntPtr _observer;
        private readonly ChannelWriter<IWork> _workQueue;
        private readonly CancellationTokenSource _cts = new();
        private readonly CancellationToken _cancellationToken;

        private readonly Uri _uri;
        private readonly Task _readThread;

        private MemoryStream _receiveBuffer = new();

        internal Socket(ClientWebSocket webSocket, IntPtr observer, ChannelWriter<IWork> workQueue, Uri uri)
        {
            RealmLogger.Default.Log(LogLevel.Trace, $"Creating a WebSocket to {uri.GetLeftPart(UriPartial.Path)}");
            _webSocket = webSocket;
            _observer = observer;
            _workQueue = workQueue;
            _uri = uri;
            _cancellationToken = _cts.Token;
            _readThread = Task.Run(ReadThread);
        }

        private async Task ReadThread()
        {
            RealmLogger.Default.Log(LogLevel.Trace, "Entering WebSocket event loop.");

            try
            {
                await _webSocket.ConnectAsync(_uri, _cancellationToken);
                await _workQueue.WriteAsync(new WebSocketConnectedWork(_webSocket.SubProtocol!, _observer, _cancellationToken));
            }
            catch (Exception e)
            {
                var builder = new StringBuilder();
                FormatExceptionForLogging(e, builder);
                RealmLogger.Default.Log(LogLevel.Error, $"Error establishing WebSocket connection {builder}");

                await _workQueue.WriteAsync(new WebSocketClosedWork(false, (WebSocketCloseStatus)RLM_ERR_WEBSOCKET_CONNECTION_FAILED, e.Message, _observer, _cancellationToken));
                return;
            }

            var buffer = new byte[32 * 1024];
            while (_webSocket.State == WebSocketState.Open)
            {
                try
                {
                    var result = await _webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), _cancellationToken);
                    switch (result.MessageType)
                    {
                        case WebSocketMessageType.Binary:
                            await _receiveBuffer.WriteAsync(buffer, 0, result.Count);
                            if (result.EndOfMessage)
                            {
                                var currentBuffer = _receiveBuffer;
                                _receiveBuffer = new MemoryStream();
                                await _workQueue.WriteAsync(new BinaryMessageReceivedWork(currentBuffer, _observer, _cancellationToken));
                            }

                            break;
                        case WebSocketMessageType.Close:
                            RealmLogger.Default.Log(LogLevel.Trace, $"WebSocket closed with status {result.CloseStatus!.Value} and description \"{result.CloseStatusDescription}\"");
                            await _workQueue.WriteAsync(new WebSocketClosedWork(clean: true, result.CloseStatus!.Value, result.CloseStatusDescription!, _observer, _cancellationToken));
                            break;
                        default:
                            RealmLogger.Default.Log(LogLevel.Trace, $"Received unexpected text WebSocket message: {Encoding.UTF8.GetString(buffer, 0, result.Count)}");
                            break;
                    }
                }
                catch (Exception e)
                {
                    var builder = new StringBuilder();
                    FormatExceptionForLogging(e, builder);
                    RealmLogger.Default.Log(LogLevel.Error, $"Error reading from WebSocket {builder}");

                    await _workQueue.WriteAsync(new WebSocketClosedWork(false, (WebSocketCloseStatus)RLM_ERR_WEBSOCKET_READ_ERROR, e.Message, _observer, _cancellationToken));
                    return;
                }
            }
        }

        public async void Write(BinaryValue data, IntPtr native_callback)
        {
            if (_webSocket.State == WebSocketState.Aborted || _cancellationToken.IsCancellationRequested)
            {
                NativeMethods.delete_callback(native_callback);
                return;
            }

            var buffer = data.AsBytes(usePooledArray: true);

            try
            {
                await _webSocket.SendAsync(new(buffer, 0, (int)data.size), WebSocketMessageType.Binary, true, _cancellationToken);
            }
            catch (Exception e)
            {
                var builder = new StringBuilder();
                FormatExceptionForLogging(e, builder);
                RealmLogger.Default.Log(LogLevel.Error, $"Error writing to WebSocket {builder}");

                // in case of errors notify the websocket observer and just dispose the callback
                await _workQueue.WriteAsync(new WebSocketClosedWork(false, (WebSocketCloseStatus)RLM_ERR_WEBSOCKET_WRITE_ERROR, e.Message, _observer, _cancellationToken));
                NativeMethods.delete_callback(native_callback);
                return;
            }
            finally
            {
                ArrayPool<byte>.Shared.Return(buffer);
            }

            await _workQueue.WriteAsync(new EventLoopWork(native_callback, _cancellationToken));
        }

        public async void Dispose()
        {
            _cts.Cancel();

            if (_webSocket.State == WebSocketState.Open)
            {
                try
                {
                    // If the websocket is still open close it, but throw away any close errors.
                    await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, default);
                }
                catch
                {
                }
            }

            _webSocket.Dispose();
            _receiveBuffer.Dispose();
            _cts.Dispose();
            RealmLogger.Default.Log(LogLevel.Trace, "Disposing WebSocket.");

            try
            {
                await _readThread;
            }
            catch(TaskCanceledException)
            {
            }
            catch(ChannelClosedException)
            {
            }
        }

        private static void FormatExceptionForLogging(Exception ex, StringBuilder builder, int nesting = 0)
        {
            var indentation = new string('\t', nesting);
            builder.Append(indentation);

            builder.AppendFormat("{0}: {1}", ex.GetType().FullName, ex.Message);
            builder.AppendLine();
            if (RealmLogger.GetLogLevel(LogCategory.Realm.SDK) >= LogLevel.Trace && !string.IsNullOrEmpty(ex.StackTrace))
            {
                builder.Append(indentation);
                var indentedTrace = ex.StackTrace.Replace(Environment.NewLine, Environment.NewLine + indentation);
                builder.AppendLine(indentedTrace);
            }

            if (ex is AggregateException aggregateException)
            {
                foreach (var inner in aggregateException.InnerExceptions)
                {
                    FormatExceptionForLogging(inner, builder, nesting + 1);
                }
            }
            else if (ex.InnerException is { } inner)
            {
                FormatExceptionForLogging(inner, builder, nesting + 1);
            }
        }
    }

    private abstract class WebSocketWork(IntPtr observer, CancellationToken cancellationToken)
        : IWork
    {
        // Belongs to the Socket and canceled when Native destroys the socket.
        // If it's canceled we shouldn't call any observer methods.
        private readonly CancellationToken _cancellationToken = cancellationToken;

        protected abstract void Execute(IntPtr observer);

        void IWork.Execute()
        {
            if (!_cancellationToken.IsCancellationRequested)
            {
                Execute(observer);
            }
        }
    }

    private sealed class WebSocketConnectedWork(string protocol, IntPtr observer, CancellationToken cancellationToken)
        : WebSocketWork(observer, cancellationToken)
    {
        protected override void Execute(IntPtr observer)
        {
            using var arena = new Arena();
            NativeMethods.observer_connected_handler(observer, StringValue.AllocateFrom(protocol, arena));
        }
    }

    private sealed class BinaryMessageReceivedWork(MemoryStream receiveBuffer, IntPtr observer, CancellationToken cancellationToken)
        : WebSocketWork(observer, cancellationToken)
    {
        protected unsafe override void Execute(IntPtr observer)
        {
            using var buffer = receiveBuffer;
            fixed (byte* data = buffer.GetBuffer())
            {
                NativeMethods.observer_binary_message_received(observer, new() { data = data, size = (IntPtr)buffer.Length });
            }
        }
    }

    private sealed class WebSocketClosedWork(bool clean, WebSocketCloseStatus status, string description, IntPtr observer, CancellationToken cancellationToken)
        : WebSocketWork(observer, cancellationToken)
    {
        protected override void Execute(IntPtr observer)
        {
            if (!clean)
            {
                NativeMethods.observer_error_handler(observer);
            }

            using var arena = new Arena();
            NativeMethods.observer_closed_handler(observer, clean, status, StringValue.AllocateFrom(description, arena));
        }
    }
}

1	////////////////////////////////////////////////////////////////////////////
2	//
3	// Copyright 2023 Realm Inc.
4	//
5	// Licensed under the Apache License, Version 2.0 (the "License");
6	// you may not use this file except in compliance with the License.
7	// You may obtain a copy of the License at
8	//
9	// http://www.apache.org/licenses/LICENSE-2.0
10	//
11	// Unless required by applicable law or agreed to in writing, software
12	// distributed under the License is distributed on an "AS IS" BASIS,
13	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	// See the License for the specific language governing permissions and
15	// limitations under the License.
16	//
17	////////////////////////////////////////////////////////////////////////////
18
19	using System;
20	using System.Buffers;
21	using System.IO;
22	using System.Net.WebSockets;
23	using System.Text;
24	using System.Threading;
25	using System.Threading.Channels;
26	using System.Threading.Tasks;
27	using Realms.Logging;
28
29	namespace Realms.Native;
30
31	internal partial class SyncSocketProvider
32	{
33	private class Socket : IDisposable
34	{
35	private readonly ClientWebSocket _webSocket;
36	private readonly IntPtr _observer;
37	private readonly ChannelWriter<IWork> _workQueue;
38	private readonly CancellationTokenSource _cts = new();	×
39	private readonly CancellationToken _cancellationToken;
40
41	private readonly Uri _uri;
42	private readonly Task _readThread;
43
44	private MemoryStream _receiveBuffer = new();	×
45
46	internal Socket(ClientWebSocket webSocket, IntPtr observer, ChannelWriter<IWork> workQueue, Uri uri)	×
47	{
48	RealmLogger.Default.Log(LogLevel.Trace, $"Creating a WebSocket to {uri.GetLeftPart(UriPartial.Path)}");	×
49	_webSocket = webSocket;	×
50	_observer = observer;	×
51	_workQueue = workQueue;	×
52	_uri = uri;	×
53	_cancellationToken = _cts.Token;	×
54	_readThread = Task.Run(ReadThread);	×
55	}	×
56
57	private async Task ReadThread()
58	{
59	RealmLogger.Default.Log(LogLevel.Trace, "Entering WebSocket event loop.");	×
60
61	try
62	{
63	await _webSocket.ConnectAsync(_uri, _cancellationToken);	×
64	await _workQueue.WriteAsync(new WebSocketConnectedWork(_webSocket.SubProtocol!, _observer, _cancellationToken));	×
65	}	×
66	catch (Exception e)	×
67	{
68	var builder = new StringBuilder();	×
69	FormatExceptionForLogging(e, builder);	×
70	RealmLogger.Default.Log(LogLevel.Error, $"Error establishing WebSocket connection {builder}");	×
71
72	await _workQueue.WriteAsync(new WebSocketClosedWork(false, (WebSocketCloseStatus)RLM_ERR_WEBSOCKET_CONNECTION_FAILED, e.Message, _observer, _cancellationToken));	×
73	return;	×
74	}
75
76	var buffer = new byte[32 * 1024];	×
77	while (_webSocket.State == WebSocketState.Open)	×
78	{
79	try
80	{
81	var result = await _webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), _cancellationToken);	×
82	switch (result.MessageType)	×
83	{
84	case WebSocketMessageType.Binary:
85	await _receiveBuffer.WriteAsync(buffer, 0, result.Count);	×
86	if (result.EndOfMessage)	×
87	{
88	var currentBuffer = _receiveBuffer;	×
89	_receiveBuffer = new MemoryStream();	×
90	await _workQueue.WriteAsync(new BinaryMessageReceivedWork(currentBuffer, _observer, _cancellationToken));	×
91	}
92
93	break;	×
94	case WebSocketMessageType.Close:
95	RealmLogger.Default.Log(LogLevel.Trace, $"WebSocket closed with status {result.CloseStatus!.Value} and description \"{result.CloseStatusDescription}\"");	×
96	await _workQueue.WriteAsync(new WebSocketClosedWork(clean: true, result.CloseStatus!.Value, result.CloseStatusDescription!, _observer, _cancellationToken));	×
97	break;	×
98	default:
99	RealmLogger.Default.Log(LogLevel.Trace, $"Received unexpected text WebSocket message: {Encoding.UTF8.GetString(buffer, 0, result.Count)}");	×
100	break;
101	}
102	}	×
103	catch (Exception e)	×
104	{
105	var builder = new StringBuilder();	×
106	FormatExceptionForLogging(e, builder);	×
107	RealmLogger.Default.Log(LogLevel.Error, $"Error reading from WebSocket {builder}");	×
108
109	await _workQueue.WriteAsync(new WebSocketClosedWork(false, (WebSocketCloseStatus)RLM_ERR_WEBSOCKET_READ_ERROR, e.Message, _observer, _cancellationToken));	×
110	return;	×
111	}
112	}
113	}	×
114
115	public async void Write(BinaryValue data, IntPtr native_callback)
116	{
117	if (_webSocket.State == WebSocketState.Aborted \|\| _cancellationToken.IsCancellationRequested)	×
118	{
119	NativeMethods.delete_callback(native_callback);	×
120	return;	×
121	}
122
123	var buffer = data.AsBytes(usePooledArray: true);	×
124
125	try
126	{
NEW 127	await _webSocket.SendAsync(new(buffer, 0, (int)data.size), WebSocketMessageType.Binary, true, _cancellationToken);	×
128	}	×
129	catch (Exception e)	×
130	{
131	var builder = new StringBuilder();	×
132	FormatExceptionForLogging(e, builder);	×
133	RealmLogger.Default.Log(LogLevel.Error, $"Error writing to WebSocket {builder}");	×
134
135	// in case of errors notify the websocket observer and just dispose the callback
136	await _workQueue.WriteAsync(new WebSocketClosedWork(false, (WebSocketCloseStatus)RLM_ERR_WEBSOCKET_WRITE_ERROR, e.Message, _observer, _cancellationToken));	×
137	NativeMethods.delete_callback(native_callback);	×
138	return;	×
139	}
140	finally
141	{
142	ArrayPool<byte>.Shared.Return(buffer);	×
143	}
144
145	await _workQueue.WriteAsync(new EventLoopWork(native_callback, _cancellationToken));	×
146	}	×
147
148	public async void Dispose()
149	{
150	_cts.Cancel();	×
151
152	if (_webSocket.State == WebSocketState.Open)	×
153	{
154	try
155	{
156	// If the websocket is still open close it, but throw away any close errors.
157	await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, default);	×
158	}	×
159	catch	×
160	{
161	}	×
162	}
163
164	_webSocket.Dispose();	×
165	_receiveBuffer.Dispose();	×
166	_cts.Dispose();	×
167	RealmLogger.Default.Log(LogLevel.Trace, "Disposing WebSocket.");	×
168
169	try
170	{
171	await _readThread;	×
172	}	×
173	catch(TaskCanceledException)	×
174	{
175	}	×
176	catch(ChannelClosedException)	×
177	{
178	}	×
179	}	×
180
181	private static void FormatExceptionForLogging(Exception ex, StringBuilder builder, int nesting = 0)
182	{
183	var indentation = new string('\t', nesting);	×
184	builder.Append(indentation);	×
185
186	builder.AppendFormat("{0}: {1}", ex.GetType().FullName, ex.Message);	×
187	builder.AppendLine();	×
188	if (RealmLogger.GetLogLevel(LogCategory.Realm.SDK) >= LogLevel.Trace && !string.IsNullOrEmpty(ex.StackTrace))	×
189	{
190	builder.Append(indentation);	×
191	var indentedTrace = ex.StackTrace.Replace(Environment.NewLine, Environment.NewLine + indentation);	×
192	builder.AppendLine(indentedTrace);	×
193	}
194
195	if (ex is AggregateException aggregateException)	×
196	{
197	foreach (var inner in aggregateException.InnerExceptions)	×
198	{
199	FormatExceptionForLogging(inner, builder, nesting + 1);	×
200	}
201	}
202	else if (ex.InnerException is { } inner)	×
203	{
204	FormatExceptionForLogging(inner, builder, nesting + 1);	×
205	}
206	}	×
207	}
208
209	private abstract class WebSocketWork(IntPtr observer, CancellationToken cancellationToken)	×
210	: IWork
211	{
212	// Belongs to the Socket and canceled when Native destroys the socket.
213	// If it's canceled we shouldn't call any observer methods.
214	private readonly CancellationToken _cancellationToken = cancellationToken;	×
215
216	protected abstract void Execute(IntPtr observer);
217
218	void IWork.Execute()
219	{
220	if (!_cancellationToken.IsCancellationRequested)	×
221	{
222	Execute(observer);	×
223	}
224	}	×
225	}
226
227	private sealed class WebSocketConnectedWork(string protocol, IntPtr observer, CancellationToken cancellationToken)
228	: WebSocketWork(observer, cancellationToken)	×
229	{
230	protected override void Execute(IntPtr observer)
231	{
232	using var arena = new Arena();	×
233	NativeMethods.observer_connected_handler(observer, StringValue.AllocateFrom(protocol, arena));	×
234	}	×
235	}
236
237	private sealed class BinaryMessageReceivedWork(MemoryStream receiveBuffer, IntPtr observer, CancellationToken cancellationToken)
238	: WebSocketWork(observer, cancellationToken)	×
239	{
240	protected unsafe override void Execute(IntPtr observer)
241	{
242	using var buffer = receiveBuffer;	×
243	fixed (byte* data = buffer.GetBuffer())	×
244	{
245	NativeMethods.observer_binary_message_received(observer, new() { data = data, size = (IntPtr)buffer.Length });	×
246	}
247	}	×
248	}
249
250	private sealed class WebSocketClosedWork(bool clean, WebSocketCloseStatus status, string description, IntPtr observer, CancellationToken cancellationToken)
251	: WebSocketWork(observer, cancellationToken)	×
252	{
253	protected override void Execute(IntPtr observer)
254	{
255	if (!clean)	×
256	{
257	NativeMethods.observer_error_handler(observer);	×
258	}
259
260	using var arena = new Arena();	×
261	NativeMethods.observer_closed_handler(observer, clean, status, StringValue.AllocateFrom(description, arena));	×
262	}	×
263	}
264	}

realm / realm-dotnet / 10887591743

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous