Add interface for websockets

Author: Trolldemorted

libsignal-service-dotnet/Base64.cs

@@ -1,48 +1,48 @@
-using System;
-using System.Text;
-
-namespace libsignalservice.util
+using System;
+using System.Text;
+
+namespace libsignalservice.util
 {
 #pragma warning disable CS1591 // Missing XML comment for publicly visible type or member
     public class Base64
-    {
-        public static string EncodeBytes(byte[] input)
-        {
-            return Convert.ToBase64String(input);
-        }
-
-        public static string EncodeBytesWithoutPadding(byte[] input)
-        {
-            String encoded = null;
-
-            encoded = EncodeBytes(input);
-
-            if (encoded[encoded.Length - 2] == '=') return encoded.Substring(0, encoded.Length - 2);
-            else if (encoded[encoded.Length - 1] == '=') return encoded.Substring(0, encoded.Length - 1);
-            else return encoded;
-        }
-
-        public static byte[] Encode(byte[] input)
-        {
-            char[] chars = System.Text.Encoding.UTF8.GetString(input, 0, input.Length).ToCharArray();
-            return Encoding.UTF8.GetBytes(Convert.ToBase64String(input));
-        }
-
-        public static byte[] Decode(string input)
-        {
-            return Convert.FromBase64String(input);
-        }
-
-        public static byte[] DecodeWithoutPadding(string input)
-        {
-            int padding = input.Length % 4;
-
-            if (padding == 1) input = input + "=";
-            else if (padding == 2) input = input + "==";
-            else if (padding == 3) input = input + "=";
-
-            return Decode(input);
-        }
+    {
+        public static string EncodeBytes(byte[] input)
+        {
+            return Convert.ToBase64String(input);
+        }
+
+        public static string EncodeBytesWithoutPadding(byte[] input)
+        {
+            String encoded = null;
+
+            encoded = EncodeBytes(input);
+
+            if (encoded[encoded.Length - 2] == '=') return encoded.Substring(0, encoded.Length - 2);
+            else if (encoded[encoded.Length - 1] == '=') return encoded.Substring(0, encoded.Length - 1);
+            else return encoded;
+        }
+
+        public static byte[] Encode(byte[] input)
+        {
+            char[] chars = System.Text.Encoding.UTF8.GetString(input, 0, input.Length).ToCharArray();
+            return Encoding.UTF8.GetBytes(Convert.ToBase64String(input));
+        }
+
+        public static byte[] Decode(string input)
+        {
+            return Convert.FromBase64String(input);
+        }
+
+        public static byte[] DecodeWithoutPadding(string input)
+        {
+            int padding = input.Length % 4;
+
+            if (padding == 1) input = input + "=";
+            else if (padding == 2) input = input + "==";
+            else if (padding == 3) input = input + "=";
+
+            return Decode(input);
+        }
     }
-#pragma warning restore CS1591 // Missing XML comment for publicly visible type or member
-}
+#pragma warning restore CS1591 // Missing XML comment for publicly visible type or member
+}

libsignal-service-dotnet/SignalServiceAccountManager.cs

@@ -11,6 +11,7 @@
 using libsignalservice.push;
 using libsignalservice.push.http;
 using libsignalservice.util;
+using libsignalservice.websocket;
 using Strilanc.Value;
 using System;
 using System.Collections.Generic;
@@ -53,11 +54,11 @@ public SignalServiceAccountManager(SignalServiceConfiguration configuration,
         /// </summary>
         /// <param name="configuration">The URL configuration for the Signal Service</param>
         /// <param name="userAgent">A string which identifies the client software</param>
-        public SignalServiceAccountManager(SignalServiceConfiguration configuration, string userAgent)
+        /// <param name="webSocketFactory">A factory which creates websocket connection objects</param>
+        public SignalServiceAccountManager(SignalServiceConfiguration configuration, string userAgent, ISignalWebSocketFactory webSocketFactory)
         {
             Configuration = configuration;
             UserAgent = userAgent;
-            ProvisioningSocket = new ProvisioningSocket(configuration.SignalServiceUrls[0].Url);
             PushServiceSocket = new PushServiceSocket(configuration, new StaticCredentialsProvider(null, null, null, (int)SignalServiceAddress.DEFAULT_DEVICE_ID), userAgent);
         }
 
@@ -238,11 +239,12 @@ public async Task<List<ContactTokenDetails>> GetContacts(CancellationToken token
         /// Request a UUID from the server for linking as a new device.
         /// Called by the new device.
         /// </summary>
-        /// <param name="token">The UUID, Base64 encoded</param>
+        /// <param name="token">A CancellationToken for the PrivisioningSocket's websocket connection</param>
+        /// /// <param name="webSocketFactory">A factory which creates websocket connection objects</param>
         /// <returns></returns>
-        public async Task<string> GetNewDeviceUuid(CancellationToken token)
+        public async Task<string> GetNewDeviceUuid(CancellationToken token, ISignalWebSocketFactory webSocketFactory)
         {
-            ProvisioningSocket = new ProvisioningSocket(Configuration.SignalServiceUrls[0].Url);
+            ProvisioningSocket = new ProvisioningSocket(Configuration.SignalServiceUrls[0].Url, webSocketFactory, token);
             return (await ProvisioningSocket.GetProvisioningUuid(token)).Uuid;
         }
 

libsignal-service-dotnet/SignalServiceMessagePipe.cs

@@ -23,16 +23,19 @@ namespace libsignalservice
     public class SignalServiceMessagePipe
     {
         private readonly ILogger Logger = LibsignalLogging.CreateLogger<SignalServiceMessagePipe>();
+        private readonly ISignalWebSocketFactory SignalWebSocketFactory;
         private readonly SignalWebSocketConnection Websocket;
-        private readonly CredentialsProvider CredentialsProvider;
+        private readonly ICredentialsProvider CredentialsProvider;
         private CancellationToken Token;
 
-        internal SignalServiceMessagePipe(CancellationToken token, SignalWebSocketConnection websocket, CredentialsProvider credentialsProvider)
+        internal SignalServiceMessagePipe(CancellationToken token, SignalWebSocketConnection websocket,
+            ICredentialsProvider credentialsProvider, ISignalWebSocketFactory webSocketFactory)
         {
             Logger.LogTrace("SignalServiceMessagePipe()");
-            this.Token = token;
-            this.Websocket = websocket;
-            this.CredentialsProvider = credentialsProvider;
+            Token = token;
+            Websocket = websocket;
+            CredentialsProvider = credentialsProvider;
+            SignalWebSocketFactory = webSocketFactory;
         }
 
         /// <summary>
@@ -42,7 +45,7 @@ internal SignalServiceMessagePipe(CancellationToken token, SignalWebSocketConnec
         public async Task Connect()
         {
             Logger.LogTrace("Connecting to message pipe");
-            await Websocket.Connect(Token);
+            await Websocket.Connect();
         }
 
         /// <summary>
@@ -99,18 +102,19 @@ public async Task<SendMessageResponse> Send(OutgoingPushMessageList list)
                 Body = ByteString.CopyFrom(Encoding.UTF8.GetBytes(JsonUtil.ToJson(list)))
             };
             requestmessage.Headers.Add("content-type:application/json");
-            Logger.LogDebug("Sending message {0}", requestmessage.Id);
-            var t = Websocket.SendRequest(requestmessage);
-            await t;
-            if (t.IsCompleted)
+            var sendTask = (await Websocket.SendRequest(requestmessage)).Task;
+            var timerCancelSource = new CancellationTokenSource();
+            if (await Task.WhenAny(sendTask, Task.Delay(10*1000, timerCancelSource.Token)) == sendTask)
             {
-                var response = t.Result;
-                if (response.Item1 < 200 || response.Item1 >= 300)
+                timerCancelSource.Cancel();
+                var (Status, Body) = sendTask.Result;
+                if (Status < 200 || Status >= 300)
                 {
-                    Logger.LogError("Sending message {0} failed: {1}", requestmessage.Id, response.Item2);
-                    throw new IOException("non-successfull response: " + response.Item1 + " " + response.Item2);
+                    throw new IOException("non-successfull response: " + Status);
                 }
-                return JsonUtil.FromJson<SendMessageResponse>(response.Item2);
+                if (Util.IsEmpty(Body))
+                    return new SendMessageResponse(false);
+                return JsonUtil.FromJson<SendMessageResponse>(Body);
             }
             else
             {
@@ -134,16 +138,17 @@ public async Task<SignalServiceProfile> GetProfile(SignalServiceAddress address)
                 Path = $"/v1/profile/{address.E164number}"
             };
 
-            var t = Websocket.SendRequest(requestMessage);
-            await t;
-            if (t.IsCompleted)
+            var sendTask = (await Websocket.SendRequest(requestMessage)).Task;
+            var timerCancelSource = new CancellationTokenSource();
+            if (await Task.WhenAny(sendTask, Task.Delay(10 * 1000, timerCancelSource.Token)) == sendTask)
             {
-                var response = t.Result;
-                if (response.Item1 < 200 || response.Item1 >= 300)
+                timerCancelSource.Cancel();
+                var (Status, Body) = sendTask.Result;
+                if (Status < 200 || Status >= 300)
                 {
-                    throw new IOException("non-successfull response: " + response.Item1 + " " + response.Item2);
+                    throw new IOException("non-successfull response: " + Status);
                 }
-                return JsonUtil.FromJson<SignalServiceProfile>(response.Item2);
+                return JsonUtil.FromJson<SignalServiceProfile>(Body);
             }
             else
             {

libsignal-service-dotnet/SignalServiceMessageReceiver.cs

@@ -32,24 +32,21 @@ public class SignalServiceMessageReceiver
         private const int MAC_KEY_SIZE = 32;
         private readonly PushServiceSocket Socket;
         private readonly SignalServiceConfiguration Urls;
-        private readonly CredentialsProvider CredentialsProvider;
+        private readonly ICredentialsProvider CredentialsProvider;
         private readonly string UserAgent;
-        private readonly ConnectivityListener ConnectivityListener;
 
         /// <summary>
         /// Construct a SignalServiceMessageReceiver.
         /// </summary>
         /// <param name="urls">The URL of the Signal Service.</param>
         /// <param name="credentials">The Signal Service user's credentials</param>
         /// <param name="userAgent"></param>
-        /// <param name="connectivityListener"></param>
-        public SignalServiceMessageReceiver(SignalServiceConfiguration urls, CredentialsProvider credentials, string userAgent, ConnectivityListener connectivityListener)
+        public SignalServiceMessageReceiver(SignalServiceConfiguration urls, ICredentialsProvider credentials, string userAgent)
         {
             Urls = urls;
             CredentialsProvider = credentials;
             Socket = new PushServiceSocket(urls, credentials, userAgent);
             UserAgent = userAgent;
-            ConnectivityListener = connectivityListener;
         }
 
         /// <summary>
@@ -111,10 +108,11 @@ public async Task<string> RetrieveAttachmentDownloadUrl(CancellationToken token,
         /// Callers must call <see cref="SignalServiceMessagePipe.Shutdown()"/> when finished with the pipe.
         /// </summary>
         /// <returns>A SignalServiceMessagePipe for receiving Signal Service messages.</returns>
-        public async Task<SignalServiceMessagePipe> CreateMessagePipe(CancellationToken token)
+        public async Task<SignalServiceMessagePipe> CreateMessagePipe(CancellationToken token, ISignalWebSocketFactory webSocketFactory)
         {
-            SignalWebSocketConnection webSocket = new SignalWebSocketConnection(token, Urls.SignalServiceUrls[0].Url, CredentialsProvider, UserAgent, ConnectivityListener);
-            var messagePipe = new SignalServiceMessagePipe(token, webSocket, CredentialsProvider);
+            SignalWebSocketConnection webSocket = new SignalWebSocketConnection(token, Urls.SignalServiceUrls[0].Url,
+                CredentialsProvider, UserAgent, webSocketFactory);
+            var messagePipe = new SignalServiceMessagePipe(token, webSocket, CredentialsProvider, webSocketFactory);
             await messagePipe.Connect();
             return messagePipe;
         }

libsignal-service-dotnet/libsignal-service-dotnet.csproj

@@ -33,7 +33,6 @@
     <PackageReference Include="Microsoft.Extensions.Logging" Version="1.1.2" />
     <PackageReference Include="Newtonsoft.Json" Version="11.0.1" />
     <PackageReference Include="Portable.BouncyCastle" Version="1.8.2" />
-    <PackageReference Include="System.Net.WebSockets.Client" Version="4.3.2" />
     <PackageReference Include="System.ValueTuple" Version="4.4.0" />
   </ItemGroup>
 

libsignal-service-dotnet/messages/SignalServiceAttachmentPointer.cs

@@ -1,11 +1,12 @@
 using System.IO;
+using System.Threading;
 
 namespace libsignalservice.messages
 {
     /// <summary>
     /// Represents a received SignalServiceAttachment "handle."  This
     /// is a pointer to the actual attachment content, which needs to be
-    /// retrieved using <see cref="SignalServiceMessageReceiver.RetrieveAttachment(SignalServiceAttachmentPointer, Stream, int, IProgressListener)"/>
+    /// retrieved using <see cref="SignalServiceMessageReceiver.RetrieveAttachment(CancellationToken, SignalServiceAttachmentPointer, Stream, int, IProgressListener)"/>
     /// </summary>
 #pragma warning disable CS1591 // Missing XML comment for publicly visible type or member
     public class SignalServiceAttachmentPointer : SignalServiceAttachment

libsignal-service-dotnet/push/ProvisioningSocket.cs

@@ -1,52 +1,54 @@
-using Coe.WebSocketWrapper;
-using libsignalservice.crypto;
-using libsignalservice.push;
-using libsignalservice.websocket;
-using System.Collections.Concurrent;
-using System.Threading;
+using Coe.WebSocketWrapper;
+using libsignalservice.crypto;
+using libsignalservice.push;
+using libsignalservice.websocket;
+using System;
+using System.Collections.Concurrent;
+using System.IO;
+using System.Threading;
 using System.Threading.Tasks;
 
-namespace libsignal.push
-{
-    internal class ProvisioningSocket
-    {
-        private WebSocketWrapper WebSocket;
-        private string WsUri;
-        private readonly BlockingCollection<byte[]> IncomingRequests = new BlockingCollection<byte[]>(new ConcurrentQueue<byte[]>());
-
-        public ProvisioningSocket(string httpUri)
-        {
-            WsUri = httpUri.Replace("https://", "wss://")
-                .Replace("http://", "ws://") + "/v1/websocket/provisioning/";
-            WebSocket = new WebSocketWrapper(WsUri);
-            WebSocket.OnMessage(Connection_OnMessage);
+namespace libsignal.push
+{
+    internal class ProvisioningSocket
+    {
+        private ISignalWebSocket SignalWebSocket;
+        private readonly string WsUri;
+        private readonly BlockingCollection<Stream> IncomingRequests = new BlockingCollection<Stream>(new ConcurrentQueue<Stream>());
+
+        public ProvisioningSocket(string httpUri, ISignalWebSocketFactory webSocketFactory, CancellationToken token)
+        {
+            WsUri = httpUri.Replace("https://", "wss://")
+                .Replace("http://", "ws://") + "/v1/websocket/provisioning/";
+            SignalWebSocket = webSocketFactory.CreateSignalWebSocket(token, new Uri(WsUri));
+            SignalWebSocket.MessageReceived += SignalWebSocket_MessageReceived;
+        }
+
+        private void SignalWebSocket_MessageReceived(object sender, SignalWebSocketMessageReceivedEventArgs e)
+        {
+            IncomingRequests.Add(e.Message);
         }
 
-        private async Task<byte[]> TakeAsync(CancellationToken token)
+        private async Task<Stream> TakeAsync(CancellationToken token)
         {
             return await Task.Run(() =>
             {
                 return IncomingRequests.Take(token); //TODO don't block
             });
-        }
-
-        public async Task<ProvisioningUuid> GetProvisioningUuid(CancellationToken token)
-        {
-            await WebSocket.Connect(token);
-            byte[] raw = await TakeAsync(token);
-            return ProvisioningUuid.Parser.ParseFrom(WebSocketMessage.Parser.ParseFrom(raw).Request.Body);
-        }
-
-        public async Task<ProvisionMessage> GetProvisioningMessage(CancellationToken token, IdentityKeyPair tmpIdentity)
-        {
-            byte[] raw = await TakeAsync(token);
-            WebSocketMessage msg = WebSocketMessage.Parser.ParseFrom(raw);
-            return new ProvisioningCipher(null).Decrypt(tmpIdentity, msg.Request.Body.ToByteArray());
-        }
-
-        private void Connection_OnMessage(byte[] obj)
-        {
-            IncomingRequests.Add(obj);
-        }
-    }
-}
+        }
+
+        public async Task<ProvisioningUuid> GetProvisioningUuid(CancellationToken token)
+        {
+            await SignalWebSocket.ConnectAsync();
+            Stream raw = await TakeAsync(token);
+            return ProvisioningUuid.Parser.ParseFrom(WebSocketMessage.Parser.ParseFrom(raw).Request.Body);
+        }
+
+        public async Task<ProvisionMessage> GetProvisioningMessage(CancellationToken token, IdentityKeyPair tmpIdentity)
+        {
+            Stream raw = await TakeAsync(token);
+            WebSocketMessage msg = WebSocketMessage.Parser.ParseFrom(raw);
+            return new ProvisioningCipher(null).Decrypt(tmpIdentity, msg.Request.Body.ToByteArray());
+        }
+    }
+}