diff --git a/MQTTnet.sln b/MQTTnet.sln
index 45c7bf35..9a151b37 100644
--- a/MQTTnet.sln
+++ b/MQTTnet.sln
@@ -34,52 +34,164 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Extensions.TopicTem
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.Server", "Source\MQTTnet.Server\MQTTnet.Server.csproj", "{C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}"
EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Extensions.Socks5", "Source\MQTTnet.Extensions.Socks5\MQTTnet.Extensions.Socks5.csproj", "{C5D940EF-89E4-431F-BE28-AE4BD7689E24}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Tests.Extensions.Socks5", "Source\MQTTnet.Tests.Extensions.Socks5\MQTTnet.Tests.Extensions.Socks5.csproj", "{D8B6C53E-3F70-4B0C-9F61-2E2F8C00501A}"
+EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
+ Debug|x64 = Debug|x64
+ Debug|x86 = Debug|x86
Release|Any CPU = Release|Any CPU
+ Release|x64 = Release|x64
+ Release|x86 = Release|x86
EndGlobalSection
GlobalSection(ProjectConfigurationPlatforms) = postSolution
{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Debug|x64.ActiveCfg = Debug|Any CPU
+ {3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Debug|x64.Build.0 = Debug|Any CPU
+ {3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Debug|x86.ActiveCfg = Debug|Any CPU
+ {3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Debug|x86.Build.0 = Debug|Any CPU
{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Release|Any CPU.ActiveCfg = Release|Any CPU
{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Release|Any CPU.Build.0 = Release|Any CPU
+ {3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Release|x64.ActiveCfg = Release|Any CPU
+ {3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Release|x64.Build.0 = Release|Any CPU
+ {3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Release|x86.ActiveCfg = Release|Any CPU
+ {3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Release|x86.Build.0 = Release|Any CPU
{F10C4060-F7EE-4A83-919F-FF723E72F94A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{F10C4060-F7EE-4A83-919F-FF723E72F94A}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {F10C4060-F7EE-4A83-919F-FF723E72F94A}.Debug|x64.ActiveCfg = Debug|Any CPU
+ {F10C4060-F7EE-4A83-919F-FF723E72F94A}.Debug|x64.Build.0 = Debug|Any CPU
+ {F10C4060-F7EE-4A83-919F-FF723E72F94A}.Debug|x86.ActiveCfg = Debug|Any CPU
+ {F10C4060-F7EE-4A83-919F-FF723E72F94A}.Debug|x86.Build.0 = Debug|Any CPU
{F10C4060-F7EE-4A83-919F-FF723E72F94A}.Release|Any CPU.ActiveCfg = Release|Any CPU
{F10C4060-F7EE-4A83-919F-FF723E72F94A}.Release|Any CPU.Build.0 = Release|Any CPU
+ {F10C4060-F7EE-4A83-919F-FF723E72F94A}.Release|x64.ActiveCfg = Release|Any CPU
+ {F10C4060-F7EE-4A83-919F-FF723E72F94A}.Release|x64.Build.0 = Release|Any CPU
+ {F10C4060-F7EE-4A83-919F-FF723E72F94A}.Release|x86.ActiveCfg = Release|Any CPU
+ {F10C4060-F7EE-4A83-919F-FF723E72F94A}.Release|x86.Build.0 = Release|Any CPU
{C444E9C8-95FA-430E-9126-274129DE16CD}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{C444E9C8-95FA-430E-9126-274129DE16CD}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {C444E9C8-95FA-430E-9126-274129DE16CD}.Debug|x64.ActiveCfg = Debug|Any CPU
+ {C444E9C8-95FA-430E-9126-274129DE16CD}.Debug|x64.Build.0 = Debug|Any CPU
+ {C444E9C8-95FA-430E-9126-274129DE16CD}.Debug|x86.ActiveCfg = Debug|Any CPU
+ {C444E9C8-95FA-430E-9126-274129DE16CD}.Debug|x86.Build.0 = Debug|Any CPU
{C444E9C8-95FA-430E-9126-274129DE16CD}.Release|Any CPU.ActiveCfg = Release|Any CPU
{C444E9C8-95FA-430E-9126-274129DE16CD}.Release|Any CPU.Build.0 = Release|Any CPU
+ {C444E9C8-95FA-430E-9126-274129DE16CD}.Release|x64.ActiveCfg = Release|Any CPU
+ {C444E9C8-95FA-430E-9126-274129DE16CD}.Release|x64.Build.0 = Release|Any CPU
+ {C444E9C8-95FA-430E-9126-274129DE16CD}.Release|x86.ActiveCfg = Release|Any CPU
+ {C444E9C8-95FA-430E-9126-274129DE16CD}.Release|x86.Build.0 = Release|Any CPU
{71CF35F5-3327-4A91-AAF4-5340F6701771}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{71CF35F5-3327-4A91-AAF4-5340F6701771}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {71CF35F5-3327-4A91-AAF4-5340F6701771}.Debug|x64.ActiveCfg = Debug|Any CPU
+ {71CF35F5-3327-4A91-AAF4-5340F6701771}.Debug|x64.Build.0 = Debug|Any CPU
+ {71CF35F5-3327-4A91-AAF4-5340F6701771}.Debug|x86.ActiveCfg = Debug|Any CPU
+ {71CF35F5-3327-4A91-AAF4-5340F6701771}.Debug|x86.Build.0 = Debug|Any CPU
{71CF35F5-3327-4A91-AAF4-5340F6701771}.Release|Any CPU.ActiveCfg = Release|Any CPU
{71CF35F5-3327-4A91-AAF4-5340F6701771}.Release|Any CPU.Build.0 = Release|Any CPU
+ {71CF35F5-3327-4A91-AAF4-5340F6701771}.Release|x64.ActiveCfg = Release|Any CPU
+ {71CF35F5-3327-4A91-AAF4-5340F6701771}.Release|x64.Build.0 = Release|Any CPU
+ {71CF35F5-3327-4A91-AAF4-5340F6701771}.Release|x86.ActiveCfg = Release|Any CPU
+ {71CF35F5-3327-4A91-AAF4-5340F6701771}.Release|x86.Build.0 = Release|Any CPU
{B270F32A-9F3E-42EE-A989-813E35E29ADB}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{B270F32A-9F3E-42EE-A989-813E35E29ADB}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {B270F32A-9F3E-42EE-A989-813E35E29ADB}.Debug|x64.ActiveCfg = Debug|Any CPU
+ {B270F32A-9F3E-42EE-A989-813E35E29ADB}.Debug|x64.Build.0 = Debug|Any CPU
+ {B270F32A-9F3E-42EE-A989-813E35E29ADB}.Debug|x86.ActiveCfg = Debug|Any CPU
+ {B270F32A-9F3E-42EE-A989-813E35E29ADB}.Debug|x86.Build.0 = Debug|Any CPU
{B270F32A-9F3E-42EE-A989-813E35E29ADB}.Release|Any CPU.ActiveCfg = Release|Any CPU
{B270F32A-9F3E-42EE-A989-813E35E29ADB}.Release|Any CPU.Build.0 = Release|Any CPU
+ {B270F32A-9F3E-42EE-A989-813E35E29ADB}.Release|x64.ActiveCfg = Release|Any CPU
+ {B270F32A-9F3E-42EE-A989-813E35E29ADB}.Release|x64.Build.0 = Release|Any CPU
+ {B270F32A-9F3E-42EE-A989-813E35E29ADB}.Release|x86.ActiveCfg = Release|Any CPU
+ {B270F32A-9F3E-42EE-A989-813E35E29ADB}.Release|x86.Build.0 = Release|Any CPU
{2F516E76-AAC4-4219-B7D1-34CDD3CFF381}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{2F516E76-AAC4-4219-B7D1-34CDD3CFF381}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {2F516E76-AAC4-4219-B7D1-34CDD3CFF381}.Debug|x64.ActiveCfg = Debug|Any CPU
+ {2F516E76-AAC4-4219-B7D1-34CDD3CFF381}.Debug|x64.Build.0 = Debug|Any CPU
+ {2F516E76-AAC4-4219-B7D1-34CDD3CFF381}.Debug|x86.ActiveCfg = Debug|Any CPU
+ {2F516E76-AAC4-4219-B7D1-34CDD3CFF381}.Debug|x86.Build.0 = Debug|Any CPU
{2F516E76-AAC4-4219-B7D1-34CDD3CFF381}.Release|Any CPU.ActiveCfg = Release|Any CPU
{2F516E76-AAC4-4219-B7D1-34CDD3CFF381}.Release|Any CPU.Build.0 = Release|Any CPU
+ {2F516E76-AAC4-4219-B7D1-34CDD3CFF381}.Release|x64.ActiveCfg = Release|Any CPU
+ {2F516E76-AAC4-4219-B7D1-34CDD3CFF381}.Release|x64.Build.0 = Release|Any CPU
+ {2F516E76-AAC4-4219-B7D1-34CDD3CFF381}.Release|x86.ActiveCfg = Release|Any CPU
+ {2F516E76-AAC4-4219-B7D1-34CDD3CFF381}.Release|x86.Build.0 = Release|Any CPU
{175D5340-CC5B-4542-939D-4E7D15A0BC8D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{175D5340-CC5B-4542-939D-4E7D15A0BC8D}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {175D5340-CC5B-4542-939D-4E7D15A0BC8D}.Debug|x64.ActiveCfg = Debug|Any CPU
+ {175D5340-CC5B-4542-939D-4E7D15A0BC8D}.Debug|x64.Build.0 = Debug|Any CPU
+ {175D5340-CC5B-4542-939D-4E7D15A0BC8D}.Debug|x86.ActiveCfg = Debug|Any CPU
+ {175D5340-CC5B-4542-939D-4E7D15A0BC8D}.Debug|x86.Build.0 = Debug|Any CPU
{175D5340-CC5B-4542-939D-4E7D15A0BC8D}.Release|Any CPU.ActiveCfg = Release|Any CPU
{175D5340-CC5B-4542-939D-4E7D15A0BC8D}.Release|Any CPU.Build.0 = Release|Any CPU
+ {175D5340-CC5B-4542-939D-4E7D15A0BC8D}.Release|x64.ActiveCfg = Release|Any CPU
+ {175D5340-CC5B-4542-939D-4E7D15A0BC8D}.Release|x64.Build.0 = Release|Any CPU
+ {175D5340-CC5B-4542-939D-4E7D15A0BC8D}.Release|x86.ActiveCfg = Release|Any CPU
+ {175D5340-CC5B-4542-939D-4E7D15A0BC8D}.Release|x86.Build.0 = Release|Any CPU
{72867E4C-4E15-4E8E-8FAB-AE9253286BBC}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{72867E4C-4E15-4E8E-8FAB-AE9253286BBC}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {72867E4C-4E15-4E8E-8FAB-AE9253286BBC}.Debug|x64.ActiveCfg = Debug|Any CPU
+ {72867E4C-4E15-4E8E-8FAB-AE9253286BBC}.Debug|x64.Build.0 = Debug|Any CPU
+ {72867E4C-4E15-4E8E-8FAB-AE9253286BBC}.Debug|x86.ActiveCfg = Debug|Any CPU
+ {72867E4C-4E15-4E8E-8FAB-AE9253286BBC}.Debug|x86.Build.0 = Debug|Any CPU
{72867E4C-4E15-4E8E-8FAB-AE9253286BBC}.Release|Any CPU.ActiveCfg = Release|Any CPU
{72867E4C-4E15-4E8E-8FAB-AE9253286BBC}.Release|Any CPU.Build.0 = Release|Any CPU
+ {72867E4C-4E15-4E8E-8FAB-AE9253286BBC}.Release|x64.ActiveCfg = Release|Any CPU
+ {72867E4C-4E15-4E8E-8FAB-AE9253286BBC}.Release|x64.Build.0 = Release|Any CPU
+ {72867E4C-4E15-4E8E-8FAB-AE9253286BBC}.Release|x86.ActiveCfg = Release|Any CPU
+ {72867E4C-4E15-4E8E-8FAB-AE9253286BBC}.Release|x86.Build.0 = Release|Any CPU
{374D5861-F7A6-42A2-8CBE-040846000EF9}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{374D5861-F7A6-42A2-8CBE-040846000EF9}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {374D5861-F7A6-42A2-8CBE-040846000EF9}.Debug|x64.ActiveCfg = Debug|Any CPU
+ {374D5861-F7A6-42A2-8CBE-040846000EF9}.Debug|x64.Build.0 = Debug|Any CPU
+ {374D5861-F7A6-42A2-8CBE-040846000EF9}.Debug|x86.ActiveCfg = Debug|Any CPU
+ {374D5861-F7A6-42A2-8CBE-040846000EF9}.Debug|x86.Build.0 = Debug|Any CPU
{374D5861-F7A6-42A2-8CBE-040846000EF9}.Release|Any CPU.ActiveCfg = Release|Any CPU
{374D5861-F7A6-42A2-8CBE-040846000EF9}.Release|Any CPU.Build.0 = Release|Any CPU
+ {374D5861-F7A6-42A2-8CBE-040846000EF9}.Release|x64.ActiveCfg = Release|Any CPU
+ {374D5861-F7A6-42A2-8CBE-040846000EF9}.Release|x64.Build.0 = Release|Any CPU
+ {374D5861-F7A6-42A2-8CBE-040846000EF9}.Release|x86.ActiveCfg = Release|Any CPU
+ {374D5861-F7A6-42A2-8CBE-040846000EF9}.Release|x86.Build.0 = Release|Any CPU
{C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}.Debug|x64.ActiveCfg = Debug|Any CPU
+ {C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}.Debug|x64.Build.0 = Debug|Any CPU
+ {C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}.Debug|x86.ActiveCfg = Debug|Any CPU
+ {C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}.Debug|x86.Build.0 = Debug|Any CPU
{C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}.Release|Any CPU.ActiveCfg = Release|Any CPU
{C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}.Release|Any CPU.Build.0 = Release|Any CPU
+ {C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}.Release|x64.ActiveCfg = Release|Any CPU
+ {C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}.Release|x64.Build.0 = Release|Any CPU
+ {C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}.Release|x86.ActiveCfg = Release|Any CPU
+ {C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}.Release|x86.Build.0 = Release|Any CPU
+ {C5D940EF-89E4-431F-BE28-AE4BD7689E24}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {C5D940EF-89E4-431F-BE28-AE4BD7689E24}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {C5D940EF-89E4-431F-BE28-AE4BD7689E24}.Debug|x64.ActiveCfg = Debug|Any CPU
+ {C5D940EF-89E4-431F-BE28-AE4BD7689E24}.Debug|x64.Build.0 = Debug|Any CPU
+ {C5D940EF-89E4-431F-BE28-AE4BD7689E24}.Debug|x86.ActiveCfg = Debug|Any CPU
+ {C5D940EF-89E4-431F-BE28-AE4BD7689E24}.Debug|x86.Build.0 = Debug|Any CPU
+ {C5D940EF-89E4-431F-BE28-AE4BD7689E24}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {C5D940EF-89E4-431F-BE28-AE4BD7689E24}.Release|Any CPU.Build.0 = Release|Any CPU
+ {C5D940EF-89E4-431F-BE28-AE4BD7689E24}.Release|x64.ActiveCfg = Release|Any CPU
+ {C5D940EF-89E4-431F-BE28-AE4BD7689E24}.Release|x64.Build.0 = Release|Any CPU
+ {C5D940EF-89E4-431F-BE28-AE4BD7689E24}.Release|x86.ActiveCfg = Release|Any CPU
+ {C5D940EF-89E4-431F-BE28-AE4BD7689E24}.Release|x86.Build.0 = Release|Any CPU
+ {D8B6C53E-3F70-4B0C-9F61-2E2F8C00501A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {D8B6C53E-3F70-4B0C-9F61-2E2F8C00501A}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {D8B6C53E-3F70-4B0C-9F61-2E2F8C00501A}.Debug|x64.ActiveCfg = Debug|Any CPU
+ {D8B6C53E-3F70-4B0C-9F61-2E2F8C00501A}.Debug|x64.Build.0 = Debug|Any CPU
+ {D8B6C53E-3F70-4B0C-9F61-2E2F8C00501A}.Debug|x86.ActiveCfg = Debug|Any CPU
+ {D8B6C53E-3F70-4B0C-9F61-2E2F8C00501A}.Debug|x86.Build.0 = Debug|Any CPU
+ {D8B6C53E-3F70-4B0C-9F61-2E2F8C00501A}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {D8B6C53E-3F70-4B0C-9F61-2E2F8C00501A}.Release|Any CPU.Build.0 = Release|Any CPU
+ {D8B6C53E-3F70-4B0C-9F61-2E2F8C00501A}.Release|x64.ActiveCfg = Release|Any CPU
+ {D8B6C53E-3F70-4B0C-9F61-2E2F8C00501A}.Release|x64.Build.0 = Release|Any CPU
+ {D8B6C53E-3F70-4B0C-9F61-2E2F8C00501A}.Release|x86.ActiveCfg = Release|Any CPU
+ {D8B6C53E-3F70-4B0C-9F61-2E2F8C00501A}.Release|x86.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
diff --git a/Samples/Client/Client_Socks5_Samples.cs b/Samples/Client/Client_Socks5_Samples.cs
new file mode 100644
index 00000000..ba0fe9ee
--- /dev/null
+++ b/Samples/Client/Client_Socks5_Samples.cs
@@ -0,0 +1,86 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+// ReSharper disable UnusedType.Global
+// ReSharper disable UnusedMember.Global
+// ReSharper disable InconsistentNaming
+
+using MQTTnet.Extensions.Socks5;
+using MQTTnet.Samples.Helpers;
+
+namespace MQTTnet.Samples.Client;
+
+public static class Client_Socks5_Samples
+{
+ public static async Task Connect_Tcp_Via_Socks5_No_Auth()
+ {
+ /*
+ * This sample connects to an MQTT broker through a SOCKS5 proxy that does not
+ * require authentication. The SOCKS5 stream provider performs the proxy handshake
+ * (RFC 1928) and returns a connected stream that the MQTT TCP channel can use as if
+ * it were a direct TCP connection.
+ *
+ * Host name resolution is performed by the proxy by default (ATYP=DOMAINNAME), which
+ * is the desired behavior when the MQTT broker hostname is only reachable from inside
+ * the network the proxy lives in.
+ */
+
+ var mqttFactory = new MqttClientFactory();
+
+ using var mqttClient = mqttFactory.CreateMqttClient();
+
+ var mqttClientOptions = new MqttClientOptionsBuilder()
+ .WithTcpServer("broker.internal.example", 1883)
+ .WithSocks5Proxy(socks => socks
+ .WithHost("proxy.example.com")
+ .WithPort(1080))
+ .Build();
+
+ using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(15));
+ var response = await mqttClient.ConnectAsync(mqttClientOptions, timeout.Token);
+
+ Console.WriteLine("The MQTT client is connected through SOCKS5.");
+ response.DumpToConsole();
+
+ var disconnectOptions = mqttFactory.CreateClientDisconnectOptionsBuilder().Build();
+ await mqttClient.DisconnectAsync(disconnectOptions, CancellationToken.None);
+ }
+
+ public static async Task Connect_Tcp_With_Tls_Via_Socks5_With_Credentials()
+ {
+ /*
+ * This sample connects to a TLS-protected MQTT broker through a SOCKS5 proxy that
+ * requires username/password authentication (RFC 1929).
+ *
+ * Note that the TLS handshake is layered on top of the SOCKS5 tunnel: the SNI used
+ * by SslStream is derived from the broker hostname (or the value passed to
+ * WithTargetHost), not from the proxy hostname.
+ */
+
+ var mqttFactory = new MqttClientFactory();
+
+ using var mqttClient = mqttFactory.CreateMqttClient();
+
+ var mqttClientOptions = new MqttClientOptionsBuilder()
+ .WithTcpServer("broker.internal.example", 8883)
+ .WithTlsOptions(tls => tls
+ .WithTargetHost("broker.internal.example"))
+ .WithSocks5Proxy(socks => socks
+ .WithHost("proxy.example.com")
+ .WithPort(1080)
+ .WithCredentials("alice", "s3cr3t")
+ .WithResolveDnsRemotely(true)
+ .WithHandshakeTimeout(TimeSpan.FromSeconds(10)))
+ .Build();
+
+ using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(15));
+ var response = await mqttClient.ConnectAsync(mqttClientOptions, timeout.Token);
+
+ Console.WriteLine("The MQTT client is connected through SOCKS5 with TLS.");
+ response.DumpToConsole();
+
+ var disconnectOptions = mqttFactory.CreateClientDisconnectOptionsBuilder().Build();
+ await mqttClient.DisconnectAsync(disconnectOptions, CancellationToken.None);
+ }
+}
diff --git a/Samples/MQTTnet.Samples.csproj b/Samples/MQTTnet.Samples.csproj
index 3de761aa..a5ccf14f 100644
--- a/Samples/MQTTnet.Samples.csproj
+++ b/Samples/MQTTnet.Samples.csproj
@@ -12,6 +12,7 @@
+
diff --git a/Source/MQTTnet.Extensions.Socks5/MQTTnet.Extensions.Socks5.csproj b/Source/MQTTnet.Extensions.Socks5/MQTTnet.Extensions.Socks5.csproj
new file mode 100644
index 00000000..1c536b08
--- /dev/null
+++ b/Source/MQTTnet.Extensions.Socks5/MQTTnet.Extensions.Socks5.csproj
@@ -0,0 +1,21 @@
+
+
+
+ MQTTnet.Extensions.Socks5
+ MQTTnet.Extensions.Socks5
+ true
+
+ True
+ This is an extension library which adds SOCKS5 proxy support (RFC 1928 / RFC 1929) to the MQTTnet TCP transport.
+ MQTTnet.Extensions.Socks5
+ MQTT Message Queue Telemetry Transport MQTTClient SOCKS SOCKS5 Proxy Tunnel Networking IoT InternetOfThings
+ true
+ For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).
+ true
+
+
+
+
+
+
+
diff --git a/Source/MQTTnet.Extensions.Socks5/MqttClientOptionsBuilderExtensions.cs b/Source/MQTTnet.Extensions.Socks5/MqttClientOptionsBuilderExtensions.cs
new file mode 100644
index 00000000..4d33f361
--- /dev/null
+++ b/Source/MQTTnet.Extensions.Socks5/MqttClientOptionsBuilderExtensions.cs
@@ -0,0 +1,39 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+namespace MQTTnet.Extensions.Socks5;
+
+///
+/// Sugar extensions for configuring a SOCKS5 proxy on the MQTT client options builder.
+///
+public static class MqttClientOptionsBuilderExtensions
+{
+ ///
+ /// Configures the client to tunnel its TCP connection to the broker through a SOCKS5
+ /// proxy. A TCP transport (via WithTcpServer(...) or WithEndPoint(...))
+ /// must be configured first.
+ ///
+ public static MqttClientOptionsBuilder WithSocks5Proxy(this MqttClientOptionsBuilder builder, Socks5ProxyOptions options)
+ {
+ ArgumentNullException.ThrowIfNull(builder);
+ ArgumentNullException.ThrowIfNull(options);
+
+ return builder.WithStreamProvider(new Socks5StreamProvider(options));
+ }
+
+ ///
+ /// Configures the client to tunnel its TCP connection to the broker through a SOCKS5
+ /// proxy using the fluent .
+ ///
+ public static MqttClientOptionsBuilder WithSocks5Proxy(this MqttClientOptionsBuilder builder, Action configure)
+ {
+ ArgumentNullException.ThrowIfNull(builder);
+ ArgumentNullException.ThrowIfNull(configure);
+
+ var optionsBuilder = new Socks5ProxyOptionsBuilder();
+ configure(optionsBuilder);
+
+ return WithSocks5Proxy(builder, optionsBuilder.Build());
+ }
+}
diff --git a/Source/MQTTnet.Extensions.Socks5/Socks5NetworkStream.cs b/Source/MQTTnet.Extensions.Socks5/Socks5NetworkStream.cs
new file mode 100644
index 00000000..af5cbdc5
--- /dev/null
+++ b/Source/MQTTnet.Extensions.Socks5/Socks5NetworkStream.cs
@@ -0,0 +1,109 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Net;
+using System.Net.Sockets;
+using MQTTnet.Channel;
+
+namespace MQTTnet.Extensions.Socks5;
+
+///
+/// Wraps a opened by the SOCKS5 stream provider so
+/// MqttTcpChannel can read the original socket's endpoints via
+/// while the rest of the I/O delegates to the
+/// underlying stream.
+///
+sealed class Socks5NetworkStream : Stream, IMqttClientStreamEndPoints
+{
+ readonly NetworkStream _inner;
+
+ public Socks5NetworkStream(NetworkStream inner, EndPoint localEndPoint, EndPoint remoteEndPoint)
+ {
+ _inner = inner ?? throw new ArgumentNullException(nameof(inner));
+ LocalEndPoint = localEndPoint;
+ RemoteEndPoint = remoteEndPoint;
+ }
+
+ public EndPoint LocalEndPoint { get; }
+
+ public EndPoint RemoteEndPoint { get; }
+
+ public override bool CanRead => _inner.CanRead;
+
+ public override bool CanSeek => _inner.CanSeek;
+
+ public override bool CanWrite => _inner.CanWrite;
+
+ public override bool CanTimeout => _inner.CanTimeout;
+
+ public override long Length => _inner.Length;
+
+ public override long Position
+ {
+ get => _inner.Position;
+ set => _inner.Position = value;
+ }
+
+ public override int ReadTimeout
+ {
+ get => _inner.ReadTimeout;
+ set => _inner.ReadTimeout = value;
+ }
+
+ public override int WriteTimeout
+ {
+ get => _inner.WriteTimeout;
+ set => _inner.WriteTimeout = value;
+ }
+
+ public override void Flush() => _inner.Flush();
+
+ public override Task FlushAsync(CancellationToken cancellationToken) => _inner.FlushAsync(cancellationToken);
+
+ public override int Read(byte[] buffer, int offset, int count) => _inner.Read(buffer, offset, count);
+
+ public override int Read(Span buffer) => _inner.Read(buffer);
+
+ public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+ => _inner.ReadAsync(buffer, offset, count, cancellationToken);
+
+ public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default)
+ => _inner.ReadAsync(buffer, cancellationToken);
+
+ public override long Seek(long offset, SeekOrigin origin) => _inner.Seek(offset, origin);
+
+ public override void SetLength(long value) => _inner.SetLength(value);
+
+ public override void Write(byte[] buffer, int offset, int count) => _inner.Write(buffer, offset, count);
+
+ public override void Write(ReadOnlySpan buffer) => _inner.Write(buffer);
+
+ public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+ => _inner.WriteAsync(buffer, offset, count, cancellationToken);
+
+ public override ValueTask WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default)
+ => _inner.WriteAsync(buffer, cancellationToken);
+
+ public override void Close()
+ {
+ _inner.Close();
+ base.Close();
+ }
+
+ protected override void Dispose(bool disposing)
+ {
+ if (disposing)
+ {
+ _inner.Dispose();
+ }
+
+ base.Dispose(disposing);
+ }
+
+ public override async ValueTask DisposeAsync()
+ {
+ await _inner.DisposeAsync().ConfigureAwait(false);
+ await base.DisposeAsync().ConfigureAwait(false);
+ }
+}
diff --git a/Source/MQTTnet.Extensions.Socks5/Socks5Protocol.cs b/Source/MQTTnet.Extensions.Socks5/Socks5Protocol.cs
new file mode 100644
index 00000000..f61de19d
--- /dev/null
+++ b/Source/MQTTnet.Extensions.Socks5/Socks5Protocol.cs
@@ -0,0 +1,39 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+namespace MQTTnet.Extensions.Socks5;
+
+///
+/// Protocol constants for SOCKS5 (RFC 1928) and the username/password sub-negotiation
+/// (RFC 1929). Internal use only.
+///
+internal static class Socks5Protocol
+{
+ public const byte Version = 0x05;
+ public const byte AuthSubNegotiationVersion = 0x01;
+
+ public const byte MethodNoAuth = 0x00;
+ public const byte MethodGssApi = 0x01;
+ public const byte MethodUsernamePassword = 0x02;
+ public const byte MethodNoneAcceptable = 0xFF;
+
+ public const byte CommandConnect = 0x01;
+ public const byte Reserved = 0x00;
+
+ public const byte AddressTypeIPv4 = 0x01;
+ public const byte AddressTypeDomainName = 0x03;
+ public const byte AddressTypeIPv6 = 0x04;
+
+ public const byte ReplySucceeded = 0x00;
+ public const byte ReplyGeneralFailure = 0x01;
+ public const byte ReplyConnectionNotAllowed = 0x02;
+ public const byte ReplyNetworkUnreachable = 0x03;
+ public const byte ReplyHostUnreachable = 0x04;
+ public const byte ReplyConnectionRefused = 0x05;
+ public const byte ReplyTtlExpired = 0x06;
+ public const byte ReplyCommandNotSupported = 0x07;
+ public const byte ReplyAddressTypeNotSupported = 0x08;
+
+ public const int MaxStringLength = 255;
+}
diff --git a/Source/MQTTnet.Extensions.Socks5/Socks5ProxyOptions.cs b/Source/MQTTnet.Extensions.Socks5/Socks5ProxyOptions.cs
new file mode 100644
index 00000000..0fc52b44
--- /dev/null
+++ b/Source/MQTTnet.Extensions.Socks5/Socks5ProxyOptions.cs
@@ -0,0 +1,65 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Net.Sockets;
+
+namespace MQTTnet.Extensions.Socks5;
+
+///
+/// Options for the SOCKS5 stream provider (RFC 1928 / RFC 1929).
+/// A new provider can be constructed from these options via
+/// .
+///
+public sealed class Socks5ProxyOptions
+{
+ ///
+ /// The SOCKS5 proxy host name or IP literal. Must be set.
+ ///
+ public string Host { get; set; }
+
+ ///
+ /// The SOCKS5 proxy port. Defaults to 1080.
+ ///
+ public int Port { get; set; } = 1080;
+
+ ///
+ /// The user name used for the username/password authentication sub-negotiation (RFC 1929).
+ /// When null or empty, the client will only offer the "no authentication" method.
+ ///
+ public string Username { get; set; }
+
+ ///
+ /// The password (raw bytes per RFC 1929) used for the username/password authentication
+ /// sub-negotiation. Must be 1..255 bytes when is set.
+ ///
+ public byte[] Password { get; set; }
+
+ ///
+ /// When true (the default) the broker hostname is forwarded to the proxy as a
+ /// SOCKS5 DOMAINNAME address type, so DNS resolution happens at the proxy
+ /// (typical when the broker is behind the proxy in a private network).
+ /// When false, the hostname is resolved locally before sending the CONNECT request
+ /// and an IPv4 / IPv6 address type is used.
+ ///
+ public bool ResolveDnsRemotely { get; set; } = true;
+
+ ///
+ /// The maximum time allowed for the full SOCKS5 handshake (TCP connect to proxy + greeting
+ /// + optional auth + CONNECT). The handshake will be cancelled and a
+ /// will be thrown when this elapses.
+ /// Defaults to 30 seconds.
+ ///
+ public TimeSpan HandshakeTimeout { get; set; } = TimeSpan.FromSeconds(30);
+
+ ///
+ /// Address family preference for the socket connecting to the proxy and for local DNS
+ /// resolution when is false.
+ ///
+ public AddressFamily AddressFamily { get; set; } = AddressFamily.Unspecified;
+
+ public override string ToString()
+ {
+ return $"{Host}:{Port}";
+ }
+}
diff --git a/Source/MQTTnet.Extensions.Socks5/Socks5ProxyOptionsBuilder.cs b/Source/MQTTnet.Extensions.Socks5/Socks5ProxyOptionsBuilder.cs
new file mode 100644
index 00000000..f29c59f9
--- /dev/null
+++ b/Source/MQTTnet.Extensions.Socks5/Socks5ProxyOptionsBuilder.cs
@@ -0,0 +1,94 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Net.Sockets;
+using System.Text;
+
+namespace MQTTnet.Extensions.Socks5;
+
+///
+/// Fluent builder for .
+///
+public sealed class Socks5ProxyOptionsBuilder
+{
+ readonly Socks5ProxyOptions _options = new();
+
+ public Socks5ProxyOptions Build()
+ {
+ if (string.IsNullOrEmpty(_options.Host))
+ {
+ throw new InvalidOperationException("A SOCKS5 proxy host must be configured via WithHost.");
+ }
+
+ if (_options.Port <= 0 || _options.Port > 65535)
+ {
+ throw new InvalidOperationException("The SOCKS5 proxy port must be in the range 1..65535.");
+ }
+
+ if (_options.HandshakeTimeout <= TimeSpan.Zero)
+ {
+ throw new InvalidOperationException("The SOCKS5 handshake timeout must be greater than zero.");
+ }
+
+ if (string.IsNullOrEmpty(_options.Username) && _options.Password is { Length: > 0 })
+ {
+ throw new InvalidOperationException("A SOCKS5 password cannot be configured without a username.");
+ }
+
+ if (!string.IsNullOrEmpty(_options.Username) && (_options.Password == null || _options.Password.Length == 0))
+ {
+ throw new InvalidOperationException("A SOCKS5 password must be configured when a username is set.");
+ }
+
+ return _options;
+ }
+
+ public Socks5ProxyOptionsBuilder WithHost(string host)
+ {
+ _options.Host = host ?? throw new ArgumentNullException(nameof(host));
+ return this;
+ }
+
+ public Socks5ProxyOptionsBuilder WithPort(int port)
+ {
+ _options.Port = port;
+ return this;
+ }
+
+ public Socks5ProxyOptionsBuilder WithCredentials(string username, string password)
+ {
+ _options.Username = username;
+ _options.Password = password != null ? Encoding.UTF8.GetBytes(password) : null;
+ return this;
+ }
+
+ public Socks5ProxyOptionsBuilder WithCredentials(string username, byte[] password)
+ {
+ _options.Username = username;
+ _options.Password = password;
+ return this;
+ }
+
+ ///
+ /// Controls whether the broker hostname is forwarded to the proxy (remote DNS, default) or
+ /// resolved locally before issuing the CONNECT request.
+ ///
+ public Socks5ProxyOptionsBuilder WithResolveDnsRemotely(bool resolveDnsRemotely = true)
+ {
+ _options.ResolveDnsRemotely = resolveDnsRemotely;
+ return this;
+ }
+
+ public Socks5ProxyOptionsBuilder WithHandshakeTimeout(TimeSpan timeout)
+ {
+ _options.HandshakeTimeout = timeout;
+ return this;
+ }
+
+ public Socks5ProxyOptionsBuilder WithAddressFamily(AddressFamily addressFamily)
+ {
+ _options.AddressFamily = addressFamily;
+ return this;
+ }
+}
diff --git a/Source/MQTTnet.Extensions.Socks5/Socks5StreamProvider.cs b/Source/MQTTnet.Extensions.Socks5/Socks5StreamProvider.cs
new file mode 100644
index 00000000..8c1e1cab
--- /dev/null
+++ b/Source/MQTTnet.Extensions.Socks5/Socks5StreamProvider.cs
@@ -0,0 +1,507 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Buffers.Binary;
+using System.Net;
+using System.Net.Sockets;
+using System.Text;
+using MQTTnet.Channel;
+using MQTTnet.Exceptions;
+
+namespace MQTTnet.Extensions.Socks5;
+
+///
+/// implementation that tunnels the connection through
+/// a SOCKS5 proxy (RFC 1928) with optional username/password authentication (RFC 1929).
+///
+public sealed class Socks5StreamProvider : IMqttClientStreamProvider
+{
+ readonly Socks5ProxyOptions _options;
+
+ public Socks5StreamProvider(Socks5ProxyOptions options)
+ {
+ _options = options ?? throw new ArgumentNullException(nameof(options));
+
+ if (string.IsNullOrEmpty(_options.Host))
+ {
+ throw new ArgumentException("Socks5ProxyOptions.Host must be set.", nameof(options));
+ }
+
+ if (_options.Port <= 0 || _options.Port > 65535)
+ {
+ throw new ArgumentOutOfRangeException(nameof(options), _options.Port, "Socks5ProxyOptions.Port must be in the range 1..65535.");
+ }
+
+ if (_options.HandshakeTimeout <= TimeSpan.Zero)
+ {
+ throw new ArgumentOutOfRangeException(nameof(options), _options.HandshakeTimeout, "Socks5ProxyOptions.HandshakeTimeout must be greater than zero.");
+ }
+
+ if (string.IsNullOrEmpty(_options.Username) && _options.Password is { Length: > 0 })
+ {
+ throw new ArgumentException("Socks5ProxyOptions.Password cannot be set without Socks5ProxyOptions.Username.", nameof(options));
+ }
+
+ if (!string.IsNullOrEmpty(_options.Username) && (_options.Password == null || _options.Password.Length == 0))
+ {
+ throw new ArgumentException("Socks5ProxyOptions.Password must be set when Socks5ProxyOptions.Username is set.", nameof(options));
+ }
+ }
+
+ public async Task ConnectAsync(EndPoint brokerEndPoint, CancellationToken cancellationToken)
+ {
+ ArgumentNullException.ThrowIfNull(brokerEndPoint);
+
+ using var timeoutCts = new CancellationTokenSource(_options.HandshakeTimeout);
+ using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeoutCts.Token);
+ var handshakeToken = linkedCts.Token;
+
+ var addressFamily = _options.AddressFamily;
+ Socket socket = addressFamily == AddressFamily.Unspecified
+ ? new Socket(SocketType.Stream, ProtocolType.Tcp)
+ : new Socket(addressFamily, SocketType.Stream, ProtocolType.Tcp);
+
+ NetworkStream networkStream = null;
+ try
+ {
+ try
+ {
+ await socket.ConnectAsync(_options.Host, _options.Port, handshakeToken).ConfigureAwait(false);
+ }
+ catch (OperationCanceledException) when (timeoutCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested)
+ {
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyUnreachable,
+ $"Timed out connecting to SOCKS5 proxy after {_options.HandshakeTimeout}.",
+ _options.ToString());
+ }
+ catch (Exception ex) when (ex is SocketException or IOException)
+ {
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyUnreachable,
+ $"Could not connect to SOCKS5 proxy: {ex.Message}",
+ _options.ToString(),
+ ex);
+ }
+
+ networkStream = new NetworkStream(socket, ownsSocket: true);
+
+ try
+ {
+ await PerformGreetingAndAuthAsync(networkStream, handshakeToken).ConfigureAwait(false);
+ await PerformConnectAsync(networkStream, brokerEndPoint, handshakeToken).ConfigureAwait(false);
+ }
+ catch (OperationCanceledException) when (timeoutCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested)
+ {
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyProtocolError,
+ $"Timed out during SOCKS5 handshake after {_options.HandshakeTimeout}.",
+ _options.ToString());
+ }
+ catch (EndOfStreamException ex)
+ {
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyProtocolError,
+ "The SOCKS5 proxy closed the connection during the handshake.",
+ _options.ToString(),
+ ex);
+ }
+ catch (Exception ex) when (ex is IOException or SocketException)
+ {
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyProtocolError,
+ $"I/O error during SOCKS5 handshake: {ex.Message}",
+ _options.ToString(),
+ ex);
+ }
+
+ // Hand off the stream — ownership transfers to the caller.
+ var result = new Socks5NetworkStream(networkStream, socket.LocalEndPoint, socket.RemoteEndPoint);
+ networkStream = null;
+ socket = null;
+ return result;
+ }
+ finally
+ {
+ networkStream?.Dispose();
+ socket?.Dispose();
+ }
+ }
+
+ async Task PerformGreetingAndAuthAsync(NetworkStream stream, CancellationToken cancellationToken)
+ {
+ var hasCredentials = !string.IsNullOrEmpty(_options.Username);
+
+ // VER NMETHODS METHOD[1..n]
+ // We always offer no-auth; we additionally offer user/pass when credentials are configured.
+ byte[] greeting;
+ if (hasCredentials)
+ {
+ greeting = [Socks5Protocol.Version, 0x02, Socks5Protocol.MethodNoAuth, Socks5Protocol.MethodUsernamePassword];
+ }
+ else
+ {
+ greeting = [Socks5Protocol.Version, 0x01, Socks5Protocol.MethodNoAuth];
+ }
+
+ await stream.WriteAsync(greeting, cancellationToken).ConfigureAwait(false);
+
+ // VER METHOD
+ var greetingResponse = new byte[2];
+ await ReadExactlyAsync(stream, greetingResponse, cancellationToken).ConfigureAwait(false);
+
+ if (greetingResponse[0] != Socks5Protocol.Version)
+ {
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyProtocolError,
+ $"Unexpected SOCKS version in greeting response: 0x{greetingResponse[0]:X2}.",
+ _options.ToString());
+ }
+
+ var chosenMethod = greetingResponse[1];
+ switch (chosenMethod)
+ {
+ case Socks5Protocol.MethodNoAuth:
+ if (hasCredentials)
+ {
+ // Server accepted no-auth even though we offered credentials: fine, continue.
+ }
+
+ return;
+
+ case Socks5Protocol.MethodUsernamePassword:
+ if (!hasCredentials)
+ {
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyProtocolError,
+ "The SOCKS5 server selected username/password authentication but no credentials are configured.",
+ _options.ToString());
+ }
+
+ await PerformUsernamePasswordAuthAsync(stream, cancellationToken).ConfigureAwait(false);
+ return;
+
+ case Socks5Protocol.MethodNoneAcceptable:
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyAuthMethodRejected,
+ "The SOCKS5 server rejected all offered authentication methods.",
+ _options.ToString());
+
+ default:
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyProtocolError,
+ $"The SOCKS5 server selected an unsupported authentication method: 0x{chosenMethod:X2}.",
+ _options.ToString());
+ }
+ }
+
+ async Task PerformUsernamePasswordAuthAsync(NetworkStream stream, CancellationToken cancellationToken)
+ {
+ var usernameBytes = Encoding.UTF8.GetBytes(_options.Username ?? string.Empty);
+ var passwordBytes = _options.Password ?? [];
+
+ if (usernameBytes.Length > Socks5Protocol.MaxStringLength)
+ {
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyAuthFailed,
+ $"Username exceeds {Socks5Protocol.MaxStringLength} bytes (RFC 1929).",
+ _options.ToString());
+ }
+
+ if (usernameBytes.Length == 0)
+ {
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyAuthFailed,
+ "Username must be 1..255 bytes (RFC 1929).",
+ _options.ToString());
+ }
+
+ if (passwordBytes.Length > Socks5Protocol.MaxStringLength)
+ {
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyAuthFailed,
+ $"Password exceeds {Socks5Protocol.MaxStringLength} bytes (RFC 1929).",
+ _options.ToString());
+ }
+
+ if (passwordBytes.Length == 0)
+ {
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyAuthFailed,
+ "Password must be 1..255 bytes (RFC 1929).",
+ _options.ToString());
+ }
+
+ var request = new byte[3 + usernameBytes.Length + passwordBytes.Length];
+ var offset = 0;
+ request[offset++] = Socks5Protocol.AuthSubNegotiationVersion;
+ request[offset++] = (byte)usernameBytes.Length;
+ Buffer.BlockCopy(usernameBytes, 0, request, offset, usernameBytes.Length);
+ offset += usernameBytes.Length;
+ request[offset++] = (byte)passwordBytes.Length;
+ Buffer.BlockCopy(passwordBytes, 0, request, offset, passwordBytes.Length);
+
+ await stream.WriteAsync(request, cancellationToken).ConfigureAwait(false);
+
+ var response = new byte[2];
+ await ReadExactlyAsync(stream, response, cancellationToken).ConfigureAwait(false);
+
+ if (response[0] != Socks5Protocol.AuthSubNegotiationVersion)
+ {
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyProtocolError,
+ $"Unexpected SOCKS5 auth sub-negotiation version: 0x{response[0]:X2}.",
+ _options.ToString());
+ }
+
+ if (response[1] != 0x00)
+ {
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyAuthFailed,
+ $"The SOCKS5 server rejected the supplied credentials (status 0x{response[1]:X2}).",
+ _options.ToString());
+ }
+ }
+
+ async Task PerformConnectAsync(NetworkStream stream, EndPoint brokerEndPoint, CancellationToken cancellationToken)
+ {
+ var request = BuildConnectRequest(brokerEndPoint, out var targetDescription);
+
+ await stream.WriteAsync(request, cancellationToken).ConfigureAwait(false);
+
+ // Reply: VER REP RSV ATYP BND.ADDR BND.PORT
+ var header = new byte[4];
+ await ReadExactlyAsync(stream, header, cancellationToken).ConfigureAwait(false);
+
+ if (header[0] != Socks5Protocol.Version)
+ {
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyProtocolError,
+ $"Unexpected SOCKS version in CONNECT response: 0x{header[0]:X2}.",
+ _options.ToString());
+ }
+
+ if (header[2] != Socks5Protocol.Reserved)
+ {
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyProtocolError,
+ $"Unexpected SOCKS5 reserved byte in CONNECT response: 0x{header[2]:X2}.",
+ _options.ToString());
+ }
+
+ var rep = header[1];
+ if (rep != Socks5Protocol.ReplySucceeded)
+ {
+ var (code, description) = MapReplyCode(rep);
+ throw new MqttProxyException(
+ code,
+ $"SOCKS5 CONNECT to '{targetDescription}' failed: {description}.",
+ _options.ToString());
+ }
+
+ // Discard BND.ADDR / BND.PORT. The length depends on ATYP and we tolerate unexpected lengths.
+ var atyp = header[3];
+ switch (atyp)
+ {
+ case Socks5Protocol.AddressTypeIPv4:
+ await SkipAsync(stream, 4 + 2, cancellationToken).ConfigureAwait(false);
+ break;
+
+ case Socks5Protocol.AddressTypeIPv6:
+ await SkipAsync(stream, 16 + 2, cancellationToken).ConfigureAwait(false);
+ break;
+
+ case Socks5Protocol.AddressTypeDomainName:
+ var lenBuf = new byte[1];
+ await ReadExactlyAsync(stream, lenBuf, cancellationToken).ConfigureAwait(false);
+ await SkipAsync(stream, lenBuf[0] + 2, cancellationToken).ConfigureAwait(false);
+ break;
+
+ default:
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyProtocolError,
+ $"Unexpected SOCKS5 address type in CONNECT response: 0x{atyp:X2}.",
+ _options.ToString());
+ }
+ }
+
+ byte[] BuildConnectRequest(EndPoint brokerEndPoint, out string targetDescription)
+ {
+ // Layout: VER CMD RSV ATYP DST.ADDR DST.PORT
+ if (brokerEndPoint is DnsEndPoint dns)
+ {
+ if (_options.ResolveDnsRemotely)
+ {
+ return BuildDomainConnectRequest(dns.Host, dns.Port, out targetDescription);
+ }
+
+ var addresses = ResolveDns(dns.Host, dns.Port);
+ return BuildIpConnectRequest(addresses, dns.Port, out targetDescription);
+ }
+
+ if (brokerEndPoint is IPEndPoint ip)
+ {
+ return BuildIpConnectRequest([ip.Address], ip.Port, out targetDescription);
+ }
+
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyAddressTypeNotSupported,
+ $"Broker endpoint of type '{brokerEndPoint.GetType().Name}' is not supported by the SOCKS5 stream provider.",
+ _options.ToString());
+ }
+
+ IPAddress[] ResolveDns(string host, int port)
+ {
+ try
+ {
+ var addresses = _options.AddressFamily == AddressFamily.Unspecified
+ ? Dns.GetHostAddresses(host)
+ : Dns.GetHostAddresses(host, _options.AddressFamily);
+
+ if (addresses == null || addresses.Length == 0)
+ {
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyHostUnreachable,
+ $"DNS resolution for '{host}:{port}' returned no addresses.",
+ _options.ToString());
+ }
+
+ return addresses;
+ }
+ catch (SocketException ex)
+ {
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyHostUnreachable,
+ $"Local DNS resolution failed for '{host}:{port}': {ex.Message}",
+ _options.ToString(),
+ ex);
+ }
+ }
+
+ byte[] BuildIpConnectRequest(IPAddress[] addresses, int port, out string targetDescription)
+ {
+ var preferredFamily = _options.AddressFamily == AddressFamily.Unspecified
+ ? AddressFamily.InterNetwork
+ : _options.AddressFamily;
+
+ IPAddress chosen = null;
+ foreach (var address in addresses)
+ {
+ if (address.AddressFamily == preferredFamily)
+ {
+ chosen = address;
+ break;
+ }
+ }
+
+ chosen ??= addresses[0];
+
+ targetDescription = $"{chosen}:{port}";
+
+ if (chosen.AddressFamily == AddressFamily.InterNetwork)
+ {
+ var addr = chosen.GetAddressBytes();
+ var request = new byte[4 + 4 + 2];
+ request[0] = Socks5Protocol.Version;
+ request[1] = Socks5Protocol.CommandConnect;
+ request[2] = Socks5Protocol.Reserved;
+ request[3] = Socks5Protocol.AddressTypeIPv4;
+ Buffer.BlockCopy(addr, 0, request, 4, 4);
+ BinaryPrimitives.WriteUInt16BigEndian(request.AsSpan(8, 2), (ushort)port);
+ return request;
+ }
+
+ if (chosen.AddressFamily == AddressFamily.InterNetworkV6)
+ {
+ var addr = chosen.GetAddressBytes();
+ var request = new byte[4 + 16 + 2];
+ request[0] = Socks5Protocol.Version;
+ request[1] = Socks5Protocol.CommandConnect;
+ request[2] = Socks5Protocol.Reserved;
+ request[3] = Socks5Protocol.AddressTypeIPv6;
+ Buffer.BlockCopy(addr, 0, request, 4, 16);
+ BinaryPrimitives.WriteUInt16BigEndian(request.AsSpan(20, 2), (ushort)port);
+ return request;
+ }
+
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyAddressTypeNotSupported,
+ $"IP address family '{chosen.AddressFamily}' is not supported by SOCKS5.",
+ _options.ToString());
+ }
+
+ byte[] BuildDomainConnectRequest(string host, int port, out string targetDescription)
+ {
+ var hostBytes = Encoding.UTF8.GetBytes(host ?? string.Empty);
+ if (hostBytes.Length == 0)
+ {
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyHostUnreachable,
+ "Broker host is empty.",
+ _options.ToString());
+ }
+
+ if (hostBytes.Length > Socks5Protocol.MaxStringLength)
+ {
+ throw new MqttProxyException(
+ MqttProxyErrorCode.ProxyAddressTypeNotSupported,
+ $"Broker host exceeds the SOCKS5 DOMAINNAME limit of {Socks5Protocol.MaxStringLength} bytes.",
+ _options.ToString());
+ }
+
+ targetDescription = $"{host}:{port}";
+
+ var request = new byte[4 + 1 + hostBytes.Length + 2];
+ request[0] = Socks5Protocol.Version;
+ request[1] = Socks5Protocol.CommandConnect;
+ request[2] = Socks5Protocol.Reserved;
+ request[3] = Socks5Protocol.AddressTypeDomainName;
+ request[4] = (byte)hostBytes.Length;
+ Buffer.BlockCopy(hostBytes, 0, request, 5, hostBytes.Length);
+ BinaryPrimitives.WriteUInt16BigEndian(request.AsSpan(5 + hostBytes.Length, 2), (ushort)port);
+ return request;
+ }
+
+ static (MqttProxyErrorCode Code, string Description) MapReplyCode(byte rep)
+ {
+ return rep switch
+ {
+ Socks5Protocol.ReplyGeneralFailure => (MqttProxyErrorCode.ProxyGeneralFailure, "general SOCKS server failure"),
+ Socks5Protocol.ReplyConnectionNotAllowed => (MqttProxyErrorCode.ProxyConnectionNotAllowed, "connection not allowed by ruleset"),
+ Socks5Protocol.ReplyNetworkUnreachable => (MqttProxyErrorCode.ProxyNetworkUnreachable, "network unreachable"),
+ Socks5Protocol.ReplyHostUnreachable => (MqttProxyErrorCode.ProxyHostUnreachable, "host unreachable"),
+ Socks5Protocol.ReplyConnectionRefused => (MqttProxyErrorCode.ProxyConnectionRefused, "connection refused"),
+ Socks5Protocol.ReplyTtlExpired => (MqttProxyErrorCode.ProxyTtlExpired, "TTL expired"),
+ Socks5Protocol.ReplyCommandNotSupported => (MqttProxyErrorCode.ProxyCommandNotSupported, "command not supported"),
+ Socks5Protocol.ReplyAddressTypeNotSupported => (MqttProxyErrorCode.ProxyAddressTypeNotSupported, "address type not supported"),
+ _ => (MqttProxyErrorCode.ProxyUnknownError, $"unknown reply code 0x{rep:X2}")
+ };
+ }
+
+ static async Task ReadExactlyAsync(NetworkStream stream, byte[] buffer, CancellationToken cancellationToken)
+ {
+ var offset = 0;
+ while (offset < buffer.Length)
+ {
+ var read = await stream.ReadAsync(buffer.AsMemory(offset, buffer.Length - offset), cancellationToken).ConfigureAwait(false);
+ if (read <= 0)
+ {
+ throw new EndOfStreamException();
+ }
+
+ offset += read;
+ }
+ }
+
+ static async Task SkipAsync(NetworkStream stream, int count, CancellationToken cancellationToken)
+ {
+ if (count <= 0)
+ {
+ return;
+ }
+
+ var buffer = new byte[count];
+ await ReadExactlyAsync(stream, buffer, cancellationToken).ConfigureAwait(false);
+ }
+}
diff --git a/Source/MQTTnet.Tests.Extensions.Socks5/AssemblyInfo.cs b/Source/MQTTnet.Tests.Extensions.Socks5/AssemblyInfo.cs
new file mode 100644
index 00000000..4e33b9de
--- /dev/null
+++ b/Source/MQTTnet.Tests.Extensions.Socks5/AssemblyInfo.cs
@@ -0,0 +1,5 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+[assembly: DoNotParallelize]
diff --git a/Source/MQTTnet.Tests.Extensions.Socks5/FakeSocks5Server.cs b/Source/MQTTnet.Tests.Extensions.Socks5/FakeSocks5Server.cs
new file mode 100644
index 00000000..6c690553
--- /dev/null
+++ b/Source/MQTTnet.Tests.Extensions.Socks5/FakeSocks5Server.cs
@@ -0,0 +1,529 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.Buffers.Binary;
+using System.Collections.Generic;
+using System.IO;
+using System.Net;
+using System.Net.Sockets;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Tests.Extensions.Socks5;
+
+///
+/// Scriptable in-process SOCKS5 server used by the test suite.
+/// Listens on 127.0.0.1 on a random port and runs a single, configurable handshake per
+/// accepted connection. On a successful CONNECT, it opens a TCP connection to the destination
+/// described by the CONNECT request and bidirectionally relays bytes between the client and
+/// the destination, allowing tests to point the SOCKS5 client at the in-memory MQTTnet
+/// broker.
+///
+internal sealed class FakeSocks5Server : IAsyncDisposable
+{
+ readonly TcpListener _listener;
+ readonly CancellationTokenSource _cts = new();
+ readonly Task _acceptLoop;
+ readonly List _connectionTasks = new();
+ readonly object _connectionsLock = new();
+ readonly List _recordedConnects = new();
+ readonly object _recordedLock = new();
+
+ public FakeSocks5Server(FakeSocks5ServerOptions options)
+ {
+ Options = options ?? throw new ArgumentNullException(nameof(options));
+
+ _listener = new TcpListener(IPAddress.Loopback, 0);
+ _listener.Start();
+
+ Port = ((IPEndPoint)_listener.LocalEndpoint).Port;
+
+ _acceptLoop = Task.Run(AcceptLoopAsync);
+ }
+
+ public int Port { get; }
+
+ public FakeSocks5ServerOptions Options { get; }
+
+ public IReadOnlyList RecordedConnects
+ {
+ get
+ {
+ lock (_recordedLock)
+ {
+ return _recordedConnects.ToArray();
+ }
+ }
+ }
+
+ public async ValueTask DisposeAsync()
+ {
+ try
+ {
+ _cts.Cancel();
+ }
+ catch (ObjectDisposedException)
+ {
+ }
+
+ try
+ {
+ _listener.Stop();
+ }
+ catch (Exception)
+ {
+ // ignore
+ }
+
+ try
+ {
+ await _acceptLoop.ConfigureAwait(false);
+ }
+ catch (Exception)
+ {
+ // ignore
+ }
+
+ Task[] pending;
+ lock (_connectionsLock)
+ {
+ pending = _connectionTasks.ToArray();
+ }
+
+ try
+ {
+ await Task.WhenAll(pending).ConfigureAwait(false);
+ }
+ catch (Exception)
+ {
+ // ignore
+ }
+
+ _cts.Dispose();
+ }
+
+ async Task AcceptLoopAsync()
+ {
+ try
+ {
+ while (!_cts.IsCancellationRequested)
+ {
+ TcpClient client;
+ try
+ {
+ client = await _listener.AcceptTcpClientAsync(_cts.Token).ConfigureAwait(false);
+ }
+ catch (OperationCanceledException)
+ {
+ return;
+ }
+ catch (SocketException)
+ {
+ return;
+ }
+ catch (ObjectDisposedException)
+ {
+ return;
+ }
+
+ var task = Task.Run(() => HandleClientAsync(client));
+ lock (_connectionsLock)
+ {
+ _connectionTasks.Add(task);
+ }
+ }
+ }
+ catch (Exception)
+ {
+ // swallow — diagnostic-only
+ }
+ }
+
+ async Task HandleClientAsync(TcpClient client)
+ {
+ using (client)
+ using (var clientStream = client.GetStream())
+ {
+ try
+ {
+ if (Options.DropAfterAccept)
+ {
+ return;
+ }
+
+ // Greeting: VER NMETHODS METHODS[]
+ var versionAndCount = new byte[2];
+ await ReadExactlyAsync(clientStream, versionAndCount, _cts.Token).ConfigureAwait(false);
+ if (versionAndCount[0] != 0x05)
+ {
+ return;
+ }
+
+ var methods = new byte[versionAndCount[1]];
+ await ReadExactlyAsync(clientStream, methods, _cts.Token).ConfigureAwait(false);
+
+ if (Options.GreetingResponseOverride != null)
+ {
+ await clientStream.WriteAsync(Options.GreetingResponseOverride, _cts.Token).ConfigureAwait(false);
+ if (Options.DropAfterGreetingResponse)
+ {
+ return;
+ }
+ }
+ else
+ {
+ byte method = SelectMethod(methods);
+ await clientStream.WriteAsync(new byte[] { 0x05, method }, _cts.Token).ConfigureAwait(false);
+
+ if (method == 0xFF)
+ {
+ return;
+ }
+
+ if (method == 0x02)
+ {
+ if (!await HandleAuthAsync(clientStream).ConfigureAwait(false))
+ {
+ return;
+ }
+ }
+ }
+
+ // CONNECT request: VER CMD RSV ATYP DST.ADDR DST.PORT
+ var header = new byte[4];
+ await ReadExactlyAsync(clientStream, header, _cts.Token).ConfigureAwait(false);
+
+ if (header[0] != 0x05)
+ {
+ // RFC 1928 §4: VER must be X'05' in every message.
+ return;
+ }
+
+ if (header[1] != 0x01)
+ {
+ // RFC 1928 §4: only CONNECT (0x01) is supported by this fake server. BIND (0x02)
+ // and UDP ASSOCIATE (0x03) are signaled as unsupported per spec via REP=0x07.
+ await WriteReplyAsync(clientStream, 0x07, Options.ReplyReservedByte, Options.ReplyAddressType, Options.ReplyBindHost, Options.ReplyBindPort).ConfigureAwait(false);
+ return;
+ }
+
+ string targetHost;
+ IPAddress targetIp = null;
+ byte atyp = header[3];
+ switch (atyp)
+ {
+ case 0x01:
+ {
+ var addr = new byte[4];
+ await ReadExactlyAsync(clientStream, addr, _cts.Token).ConfigureAwait(false);
+ targetIp = new IPAddress(addr);
+ targetHost = targetIp.ToString();
+ break;
+ }
+
+ case 0x03:
+ {
+ var lenBuf = new byte[1];
+ await ReadExactlyAsync(clientStream, lenBuf, _cts.Token).ConfigureAwait(false);
+ var hostBytes = new byte[lenBuf[0]];
+ await ReadExactlyAsync(clientStream, hostBytes, _cts.Token).ConfigureAwait(false);
+ targetHost = Encoding.UTF8.GetString(hostBytes);
+ break;
+ }
+
+ case 0x04:
+ {
+ var addr = new byte[16];
+ await ReadExactlyAsync(clientStream, addr, _cts.Token).ConfigureAwait(false);
+ targetIp = new IPAddress(addr);
+ targetHost = targetIp.ToString();
+ break;
+ }
+
+ default:
+ await WriteReplyAsync(clientStream, 0x08, Options.ReplyReservedByte, Options.ReplyAddressType, Options.ReplyBindHost, Options.ReplyBindPort).ConfigureAwait(false);
+ return;
+ }
+
+ var portBuf = new byte[2];
+ await ReadExactlyAsync(clientStream, portBuf, _cts.Token).ConfigureAwait(false);
+ var targetPort = BinaryPrimitives.ReadUInt16BigEndian(portBuf);
+
+ lock (_recordedLock)
+ {
+ _recordedConnects.Add(new RecordedConnect(atyp, targetHost, targetPort));
+ }
+
+ if (Options.ConnectReplyCode != 0x00)
+ {
+ await WriteReplyAsync(clientStream, Options.ConnectReplyCode, Options.ReplyReservedByte, Options.ReplyAddressType, Options.ReplyBindHost, Options.ReplyBindPort).ConfigureAwait(false);
+ return;
+ }
+
+ // Open the destination connection and start relaying.
+ TcpClient upstream;
+ try
+ {
+ upstream = new TcpClient();
+ if (targetIp != null)
+ {
+ await upstream.ConnectAsync(targetIp, targetPort, _cts.Token).ConfigureAwait(false);
+ }
+ else
+ {
+ await upstream.ConnectAsync(targetHost, targetPort, _cts.Token).ConfigureAwait(false);
+ }
+ }
+ catch (Exception)
+ {
+ await WriteReplyAsync(clientStream, 0x04, Options.ReplyReservedByte, Options.ReplyAddressType, Options.ReplyBindHost, Options.ReplyBindPort).ConfigureAwait(false);
+ return;
+ }
+
+ using (upstream)
+ using (var upstreamStream = upstream.GetStream())
+ {
+ await WriteReplyAsync(clientStream, 0x00, Options.ReplyReservedByte, Options.ReplyAddressType, Options.ReplyBindHost, Options.ReplyBindPort).ConfigureAwait(false);
+
+ var c2u = RelayAsync(clientStream, upstreamStream, _cts.Token);
+ var u2c = RelayAsync(upstreamStream, clientStream, _cts.Token);
+ await Task.WhenAny(c2u, u2c).ConfigureAwait(false);
+ }
+ }
+ catch (Exception)
+ {
+ // swallow — diagnostic-only
+ }
+ }
+ }
+
+ byte SelectMethod(byte[] offered)
+ {
+ var allowedMethods = Options.AllowedMethods;
+ foreach (var m in offered)
+ {
+ foreach (var allowed in allowedMethods)
+ {
+ if (m == allowed)
+ {
+ return m;
+ }
+ }
+ }
+
+ return 0xFF;
+ }
+
+ async Task HandleAuthAsync(NetworkStream stream)
+ {
+ var header = new byte[2];
+ await ReadExactlyAsync(stream, header, _cts.Token).ConfigureAwait(false);
+ if (header[0] != 0x01)
+ {
+ return false;
+ }
+
+ var ulen = header[1];
+ var userBytes = new byte[ulen];
+ await ReadExactlyAsync(stream, userBytes, _cts.Token).ConfigureAwait(false);
+
+ var plenBuf = new byte[1];
+ await ReadExactlyAsync(stream, plenBuf, _cts.Token).ConfigureAwait(false);
+ var passBytes = new byte[plenBuf[0]];
+ await ReadExactlyAsync(stream, passBytes, _cts.Token).ConfigureAwait(false);
+
+ var user = Encoding.UTF8.GetString(userBytes);
+ var pass = Encoding.UTF8.GetString(passBytes);
+
+ var ok = Options.ExpectedUsername == user && Options.ExpectedPassword == pass;
+ if (Options.ForceAuthFailure)
+ {
+ ok = false;
+ }
+
+ await stream.WriteAsync(new byte[] { 0x01, ok ? (byte)0x00 : (byte)0x01 }, _cts.Token).ConfigureAwait(false);
+ return ok;
+ }
+
+ static async Task WriteReplyAsync(NetworkStream stream, byte rep, byte replyReserved, byte replyAtyp, string replyHost, ushort replyPort)
+ {
+ // VER REP RSV ATYP BND.ADDR BND.PORT
+ //
+ // RFC 1928 §6 says BND.ADDR contains the "associated IP address" (i.e. the server-side
+ // endpoint it used to reach the destination). Most real SOCKS5 servers reply with
+ // ATYP=0x01 (IPv4) + a zeroed address when they have nothing meaningful to report,
+ // which is the safe default we use here. Tests can override the reply ATYP via
+ // FakeSocks5ServerOptions.ReplyAddressType to exercise the client's reply parser.
+
+ switch (replyAtyp)
+ {
+ case 0x04:
+ {
+ var reply = new byte[4 + 16 + 2];
+ reply[0] = 0x05;
+ reply[1] = rep;
+ reply[2] = replyReserved;
+ reply[3] = 0x04;
+ if (!string.IsNullOrEmpty(replyHost) && IPAddress.TryParse(replyHost, out var ip) && ip.AddressFamily == AddressFamily.InterNetworkV6)
+ {
+ var bytes = ip.GetAddressBytes();
+ Buffer.BlockCopy(bytes, 0, reply, 4, 16);
+ }
+ BinaryPrimitives.WriteUInt16BigEndian(reply.AsSpan(20, 2), replyPort);
+ await stream.WriteAsync(reply).ConfigureAwait(false);
+ return;
+ }
+
+ case 0x03:
+ {
+ var host = string.IsNullOrEmpty(replyHost) ? "0.0.0.0" : replyHost;
+ var hostBytes = Encoding.UTF8.GetBytes(host);
+ if (hostBytes.Length > 255)
+ {
+ Array.Resize(ref hostBytes, 255);
+ }
+
+ var reply = new byte[4 + 1 + hostBytes.Length + 2];
+ reply[0] = 0x05;
+ reply[1] = rep;
+ reply[2] = replyReserved;
+ reply[3] = 0x03;
+ reply[4] = (byte)hostBytes.Length;
+ Buffer.BlockCopy(hostBytes, 0, reply, 5, hostBytes.Length);
+ BinaryPrimitives.WriteUInt16BigEndian(reply.AsSpan(5 + hostBytes.Length, 2), replyPort);
+ await stream.WriteAsync(reply).ConfigureAwait(false);
+ return;
+ }
+
+ default:
+ {
+ var reply = new byte[4 + 4 + 2];
+ reply[0] = 0x05;
+ reply[1] = rep;
+ reply[2] = replyReserved;
+ reply[3] = 0x01;
+ if (!string.IsNullOrEmpty(replyHost) && IPAddress.TryParse(replyHost, out var ip) && ip.AddressFamily == AddressFamily.InterNetwork)
+ {
+ var bytes = ip.GetAddressBytes();
+ Buffer.BlockCopy(bytes, 0, reply, 4, 4);
+ }
+ BinaryPrimitives.WriteUInt16BigEndian(reply.AsSpan(8, 2), replyPort);
+ await stream.WriteAsync(reply).ConfigureAwait(false);
+ return;
+ }
+ }
+ }
+
+ static async Task RelayAsync(NetworkStream source, NetworkStream destination, CancellationToken cancellationToken)
+ {
+ var buffer = new byte[16 * 1024];
+ try
+ {
+ while (true)
+ {
+ var read = await source.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
+ if (read <= 0)
+ {
+ return;
+ }
+
+ await destination.WriteAsync(buffer.AsMemory(0, read), cancellationToken).ConfigureAwait(false);
+ }
+ }
+ catch (Exception)
+ {
+ // ignore
+ }
+ }
+
+ static async Task ReadExactlyAsync(NetworkStream stream, byte[] buffer, CancellationToken cancellationToken)
+ {
+ var offset = 0;
+ while (offset < buffer.Length)
+ {
+ var read = await stream.ReadAsync(buffer.AsMemory(offset, buffer.Length - offset), cancellationToken).ConfigureAwait(false);
+ if (read <= 0)
+ {
+ throw new EndOfStreamException();
+ }
+
+ offset += read;
+ }
+ }
+
+ public sealed record RecordedConnect(byte AddressType, string Host, int Port);
+}
+
+internal sealed class FakeSocks5ServerOptions
+{
+ ///
+ /// The SOCKS5 auth methods the server will accept, in preference order matched against the
+ /// client's offered methods. Defaults to no-auth only.
+ ///
+ public byte[] AllowedMethods { get; set; } = [0x00];
+
+ public string ExpectedUsername { get; set; }
+
+ public string ExpectedPassword { get; set; }
+
+ ///
+ /// Forces the username/password sub-negotiation to return STATUS=0x01 regardless of the
+ /// supplied credentials. Useful for tests that exercise the auth-failure path.
+ ///
+ public bool ForceAuthFailure { get; set; }
+
+ ///
+ /// SOCKS5 reply code for the CONNECT response. 0x00 means "succeeded" and the
+ /// server will additionally bridge bytes to the destination.
+ ///
+ public byte ConnectReplyCode { get; set; }
+
+ ///
+ /// Reserved byte included in the CONNECT reply. Defaults to 0x00 per RFC 1928.
+ ///
+ public byte ReplyReservedByte { get; set; }
+
+ ///
+ /// When set, the server sends this byte sequence instead of the regular greeting response.
+ /// Used to simulate malformed responses (e.g. wrong version).
+ ///
+ public byte[] GreetingResponseOverride { get; set; }
+
+ ///
+ /// If true, closes the connection right after writing the greeting response
+ /// override. Used together with for protocol-error
+ /// tests.
+ ///
+ public bool DropAfterGreetingResponse { get; set; }
+
+ ///
+ /// If true, the server closes the connection immediately after accepting (before
+ /// reading the greeting). Used to simulate proxy mid-handshake disconnects.
+ ///
+ public bool DropAfterAccept { get; set; }
+
+ ///
+ /// Address type used when constructing CONNECT replies. Defaults to 0x01 (IPv4),
+ /// which matches the behavior of real SOCKS5 servers — they always reply with a resolved
+ /// address, regardless of how the client expressed the destination. Override to
+ /// 0x03 or 0x04 to exercise the client's reply parser.
+ ///
+ public byte ReplyAddressType { get; set; } = 0x01;
+
+ ///
+ /// Bound host included in the CONNECT reply (BND.ADDR). Format depends on
+ /// : IPv4/IPv6 textual representation for ATYP 0x01/0x04,
+ /// a domain name for ATYP 0x03. When null or empty the server writes a zeroed
+ /// address (0.0.0.0, ::, or "0.0.0.0" respectively).
+ ///
+ public string ReplyBindHost { get; set; }
+
+ ///
+ /// Bound port included in the CONNECT reply (BND.PORT). Defaults to 0, which
+ /// is what real SOCKS5 servers return when no meaningful port is available.
+ ///
+ public ushort ReplyBindPort { get; set; }
+}
diff --git a/Source/MQTTnet.Tests.Extensions.Socks5/GlobalUsings.cs b/Source/MQTTnet.Tests.Extensions.Socks5/GlobalUsings.cs
new file mode 100644
index 00000000..1b74e07f
--- /dev/null
+++ b/Source/MQTTnet.Tests.Extensions.Socks5/GlobalUsings.cs
@@ -0,0 +1,5 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+global using Microsoft.VisualStudio.TestTools.UnitTesting;
diff --git a/Source/MQTTnet.Tests.Extensions.Socks5/MQTTnet.Tests.Extensions.Socks5.csproj b/Source/MQTTnet.Tests.Extensions.Socks5/MQTTnet.Tests.Extensions.Socks5.csproj
new file mode 100644
index 00000000..6c532c7f
--- /dev/null
+++ b/Source/MQTTnet.Tests.Extensions.Socks5/MQTTnet.Tests.Extensions.Socks5.csproj
@@ -0,0 +1,28 @@
+
+
+
+ Exe
+ false
+ default
+ disable
+ $(NoWarn);CA1707;CA1859
+ true
+ MQTTnet.Tests.Extensions.Socks5
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/Source/MQTTnet.Tests.Extensions.Socks5/Socks5MqttClient_Tests.cs b/Source/MQTTnet.Tests.Extensions.Socks5/Socks5MqttClient_Tests.cs
new file mode 100644
index 00000000..1fa260ed
--- /dev/null
+++ b/Source/MQTTnet.Tests.Extensions.Socks5/Socks5MqttClient_Tests.cs
@@ -0,0 +1,566 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.Buffers;
+using System.Linq;
+using System.Net;
+using System.Net.Sockets;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using MQTTnet.Exceptions;
+using MQTTnet.Extensions.Socks5;
+using MQTTnet.Server;
+
+namespace MQTTnet.Tests.Extensions.Socks5;
+
+[TestClass]
+public sealed class Socks5MqttClient_Tests
+{
+ [TestMethod]
+ public async Task Connect_Tcp_Via_Socks5_No_Auth_PubSub_Works()
+ {
+ Socks5TestHelper.BrokerHandle broker = null;
+ FakeSocks5Server proxy = null;
+ IMqttClient client = null;
+
+ try
+ {
+ broker = await Socks5TestHelper.StartBrokerAsync();
+ var brokerPort = Socks5TestHelper.GetBrokerPort(broker);
+
+ proxy = new FakeSocks5Server(new FakeSocks5ServerOptions());
+
+ client = new MqttClientFactory().CreateMqttClient();
+
+ var receivedTopic = string.Empty;
+ var received = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ client.ApplicationMessageReceivedAsync += e =>
+ {
+ receivedTopic = e.ApplicationMessage.Topic;
+ received.TrySetResult(Encoding.UTF8.GetString(e.ApplicationMessage.Payload.ToArray()));
+ return Task.CompletedTask;
+ };
+
+ var options = new MqttClientOptionsBuilder()
+ .WithTcpServer("127.0.0.1", brokerPort)
+ .WithSocks5Proxy(p => p.WithHost("127.0.0.1").WithPort(proxy.Port))
+ .Build();
+
+ await client.ConnectAsync(options, CancellationToken.None);
+ Assert.IsTrue(client.IsConnected);
+
+ await client.SubscribeAsync("topic/socks5");
+ await client.PublishStringAsync("topic/socks5", "hello-socks5");
+
+ var payload = await received.Task.WaitAsync(TimeSpan.FromSeconds(5));
+ Assert.AreEqual("topic/socks5", receivedTopic);
+ Assert.AreEqual("hello-socks5", payload);
+
+ Assert.HasCount(1, proxy.RecordedConnects);
+ Assert.AreEqual(brokerPort, proxy.RecordedConnects[0].Port);
+ }
+ finally
+ {
+ if (client != null)
+ {
+ try { await client.DisconnectAsync(); } catch { /* ignore */ }
+ client.Dispose();
+ }
+
+ if (proxy != null)
+ {
+ await proxy.DisposeAsync();
+ }
+
+ if (broker != null)
+ {
+ await broker.Server.StopAsync();
+ broker.Server.Dispose();
+ }
+ }
+ }
+
+ [TestMethod]
+ public async Task Connect_Tcp_Via_Socks5_With_UsernamePassword_Works()
+ {
+ var broker = await Socks5TestHelper.StartBrokerAsync();
+ var brokerPort = Socks5TestHelper.GetBrokerPort(broker);
+
+ await using var proxy = new FakeSocks5Server(new FakeSocks5ServerOptions
+ {
+ AllowedMethods = [0x02],
+ ExpectedUsername = "alice",
+ ExpectedPassword = "s3cr3t"
+ });
+
+ using var client = new MqttClientFactory().CreateMqttClient();
+
+ var options = new MqttClientOptionsBuilder()
+ .WithTcpServer("127.0.0.1", brokerPort)
+ .WithSocks5Proxy(p => p.WithHost("127.0.0.1").WithPort(proxy.Port).WithCredentials("alice", "s3cr3t"))
+ .Build();
+
+ try
+ {
+ await client.ConnectAsync(options, CancellationToken.None);
+ Assert.IsTrue(client.IsConnected);
+ }
+ finally
+ {
+ try { await client.DisconnectAsync(); } catch { /* ignore */ }
+ await broker.Server.StopAsync();
+ broker.Server.Dispose();
+ }
+ }
+
+ [TestMethod]
+ public async Task Connect_Tcp_Via_Socks5_With_Wrong_Credentials_Throws_AuthFailed()
+ {
+ var broker = await Socks5TestHelper.StartBrokerAsync();
+ var brokerPort = Socks5TestHelper.GetBrokerPort(broker);
+
+ await using var proxy = new FakeSocks5Server(new FakeSocks5ServerOptions
+ {
+ AllowedMethods = [0x02],
+ ForceAuthFailure = true
+ });
+
+ using var client = new MqttClientFactory().CreateMqttClient();
+
+ var options = new MqttClientOptionsBuilder()
+ .WithTcpServer("127.0.0.1", brokerPort)
+ .WithSocks5Proxy(p => p.WithHost("127.0.0.1").WithPort(proxy.Port).WithCredentials("alice", "wrong"))
+ .Build();
+
+ try
+ {
+ var ex = await Assert.ThrowsExactlyAsync(() => client.ConnectAsync(options));
+ Assert.AreEqual(MqttProxyErrorCode.ProxyAuthFailed, ex.ErrorCode);
+ }
+ finally
+ {
+ await broker.Server.StopAsync();
+ broker.Server.Dispose();
+ }
+ }
+
+ [TestMethod]
+ public async Task Connect_Tcp_Via_Socks5_When_Method_Rejected_Throws_AuthMethodRejected()
+ {
+ var broker = await Socks5TestHelper.StartBrokerAsync();
+ var brokerPort = Socks5TestHelper.GetBrokerPort(broker);
+
+ await using var proxy = new FakeSocks5Server(new FakeSocks5ServerOptions
+ {
+ AllowedMethods = [0x01] // GSSAPI which the client never offers.
+ });
+
+ using var client = new MqttClientFactory().CreateMqttClient();
+
+ var options = new MqttClientOptionsBuilder()
+ .WithTcpServer("127.0.0.1", brokerPort)
+ .WithSocks5Proxy(p => p.WithHost("127.0.0.1").WithPort(proxy.Port))
+ .Build();
+
+ try
+ {
+ var ex = await Assert.ThrowsExactlyAsync(() => client.ConnectAsync(options));
+ Assert.AreEqual(MqttProxyErrorCode.ProxyAuthMethodRejected, ex.ErrorCode);
+ }
+ finally
+ {
+ await broker.Server.StopAsync();
+ broker.Server.Dispose();
+ }
+ }
+
+ [TestMethod]
+ public async Task Connect_Tcp_Via_Socks5_When_Broker_Host_Unreachable_Maps_To_HostUnreachable()
+ {
+ await using var proxy = new FakeSocks5Server(new FakeSocks5ServerOptions
+ {
+ ConnectReplyCode = 0x04
+ });
+
+ using var client = new MqttClientFactory().CreateMqttClient();
+
+ var options = new MqttClientOptionsBuilder()
+ .WithTcpServer("broker.invalid", 1883)
+ .WithSocks5Proxy(p => p.WithHost("127.0.0.1").WithPort(proxy.Port))
+ .Build();
+
+ var ex = await Assert.ThrowsExactlyAsync(() => client.ConnectAsync(options));
+ Assert.AreEqual(MqttProxyErrorCode.ProxyHostUnreachable, ex.ErrorCode);
+ }
+
+ [TestMethod]
+ public async Task Connect_Tcp_Via_Socks5_When_Broker_Connection_Refused_Maps_To_ConnectionRefused()
+ {
+ await using var proxy = new FakeSocks5Server(new FakeSocks5ServerOptions
+ {
+ ConnectReplyCode = 0x05
+ });
+
+ using var client = new MqttClientFactory().CreateMqttClient();
+
+ var options = new MqttClientOptionsBuilder()
+ .WithTcpServer("127.0.0.1", 1)
+ .WithSocks5Proxy(p => p.WithHost("127.0.0.1").WithPort(proxy.Port))
+ .Build();
+
+ var ex = await Assert.ThrowsExactlyAsync(() => client.ConnectAsync(options));
+ Assert.AreEqual(MqttProxyErrorCode.ProxyConnectionRefused, ex.ErrorCode);
+ }
+
+ [TestMethod]
+ public async Task Connect_Tcp_Via_Socks5_When_Proxy_Port_Closed_Throws_ProxyUnreachable()
+ {
+ // Bind a temporary listener just to grab a free port and immediately release it.
+ var probe = new TcpListener(IPAddress.Loopback, 0);
+ probe.Start();
+ var freePort = ((IPEndPoint)probe.LocalEndpoint).Port;
+ probe.Stop();
+
+ using var client = new MqttClientFactory().CreateMqttClient();
+
+ var options = new MqttClientOptionsBuilder()
+ .WithTcpServer("127.0.0.1", 1883)
+ .WithSocks5Proxy(p => p.WithHost("127.0.0.1").WithPort(freePort).WithHandshakeTimeout(TimeSpan.FromSeconds(5)))
+ .Build();
+
+ var ex = await Assert.ThrowsExactlyAsync(() => client.ConnectAsync(options));
+ Assert.AreEqual(MqttProxyErrorCode.ProxyUnreachable, ex.ErrorCode);
+ }
+
+ [TestMethod]
+ public async Task Connect_Tcp_Via_Socks5_When_Proxy_Host_Does_Not_Resolve_Throws_ProxyUnreachable()
+ {
+ using var client = new MqttClientFactory().CreateMqttClient();
+
+ var options = new MqttClientOptionsBuilder()
+ .WithTcpServer("127.0.0.1", 1883)
+ .WithSocks5Proxy(p => p.WithHost("nonexistent.proxy.invalid").WithPort(1080).WithHandshakeTimeout(TimeSpan.FromSeconds(5)))
+ .Build();
+
+ var ex = await Assert.ThrowsExactlyAsync(() => client.ConnectAsync(options));
+ Assert.AreEqual(MqttProxyErrorCode.ProxyUnreachable, ex.ErrorCode);
+ }
+
+ [TestMethod]
+ public async Task Remote_Dns_Enabled_Sends_DomainName_To_Proxy()
+ {
+ var broker = await Socks5TestHelper.StartBrokerAsync();
+ var brokerPort = Socks5TestHelper.GetBrokerPort(broker);
+
+ await using var proxy = new FakeSocks5Server(new FakeSocks5ServerOptions());
+
+ using var client = new MqttClientFactory().CreateMqttClient();
+
+ var options = new MqttClientOptionsBuilder()
+ .WithTcpServer("localhost", brokerPort)
+ .WithSocks5Proxy(p => p.WithHost("127.0.0.1").WithPort(proxy.Port).WithResolveDnsRemotely(true))
+ .Build();
+
+ try
+ {
+ await client.ConnectAsync(options);
+ Assert.IsTrue(client.IsConnected);
+ Assert.HasCount(1, proxy.RecordedConnects);
+ Assert.AreEqual(0x03, proxy.RecordedConnects[0].AddressType);
+ Assert.AreEqual("localhost", proxy.RecordedConnects[0].Host);
+ }
+ finally
+ {
+ try { await client.DisconnectAsync(); } catch { /* ignore */ }
+ await broker.Server.StopAsync();
+ broker.Server.Dispose();
+ }
+ }
+
+ [TestMethod]
+ public async Task Remote_Dns_Disabled_Sends_Ipv4_To_Proxy()
+ {
+ var broker = await Socks5TestHelper.StartBrokerAsync();
+ var brokerPort = Socks5TestHelper.GetBrokerPort(broker);
+
+ await using var proxy = new FakeSocks5Server(new FakeSocks5ServerOptions());
+
+ using var client = new MqttClientFactory().CreateMqttClient();
+
+ var options = new MqttClientOptionsBuilder()
+ .WithTcpServer("localhost", brokerPort)
+ .WithSocks5Proxy(p => p
+ .WithHost("127.0.0.1")
+ .WithPort(proxy.Port)
+ .WithResolveDnsRemotely(false)
+ .WithAddressFamily(AddressFamily.InterNetwork))
+ .Build();
+
+ try
+ {
+ await client.ConnectAsync(options);
+ Assert.IsTrue(client.IsConnected);
+ Assert.HasCount(1, proxy.RecordedConnects);
+ Assert.AreEqual(0x01, proxy.RecordedConnects[0].AddressType);
+ Assert.AreEqual("127.0.0.1", proxy.RecordedConnects[0].Host);
+ }
+ finally
+ {
+ try { await client.DisconnectAsync(); } catch { /* ignore */ }
+ await broker.Server.StopAsync();
+ broker.Server.Dispose();
+ }
+ }
+
+ [TestMethod]
+ public async Task Ipv6_IpEndPoint_Sends_Ipv6_AddressType()
+ {
+ var broker = await Socks5TestHelper.StartBrokerAsync();
+ var brokerPort = Socks5TestHelper.GetBrokerPort(broker);
+
+ await using var proxy = new FakeSocks5Server(new FakeSocks5ServerOptions());
+
+ using var client = new MqttClientFactory().CreateMqttClient();
+
+ // Loopback (::1) is reachable on practically every CI runner.
+ var options = new MqttClientOptionsBuilder()
+ .WithEndPoint(new IPEndPoint(IPAddress.IPv6Loopback, brokerPort))
+ .WithSocks5Proxy(p => p.WithHost("127.0.0.1").WithPort(proxy.Port))
+ .Build();
+
+ try
+ {
+ await client.ConnectAsync(options);
+ Assert.IsTrue(client.IsConnected);
+ Assert.HasCount(1, proxy.RecordedConnects);
+ Assert.AreEqual(0x04, proxy.RecordedConnects[0].AddressType);
+ }
+ finally
+ {
+ try { await client.DisconnectAsync(); } catch { /* ignore */ }
+ await broker.Server.StopAsync();
+ broker.Server.Dispose();
+ }
+ }
+
+ [TestMethod]
+ public async Task Cancellation_During_Handshake_Throws_OperationCanceled()
+ {
+ // Proxy that accepts the TCP connection but never replies to the greeting.
+ await using var proxy = new FakeSocks5Server(new FakeSocks5ServerOptions
+ {
+ DropAfterAccept = true
+ });
+
+ using var client = new MqttClientFactory().CreateMqttClient();
+ using var cts = new CancellationTokenSource();
+ await cts.CancelAsync();
+
+ var options = new MqttClientOptionsBuilder()
+ .WithTcpServer("127.0.0.1", 1883)
+ .WithSocks5Proxy(p => p.WithHost("127.0.0.1").WithPort(proxy.Port))
+ .Build();
+
+ await Assert.ThrowsAsync(() => client.ConnectAsync(options, cts.Token));
+ }
+
+ [TestMethod]
+ public async Task Greeting_Returns_Wrong_Version_Throws_ProtocolError()
+ {
+ await using var proxy = new FakeSocks5Server(new FakeSocks5ServerOptions
+ {
+ GreetingResponseOverride = [0x04, 0x00],
+ DropAfterGreetingResponse = true
+ });
+
+ using var client = new MqttClientFactory().CreateMqttClient();
+
+ var options = new MqttClientOptionsBuilder()
+ .WithTcpServer("127.0.0.1", 1883)
+ .WithSocks5Proxy(p => p.WithHost("127.0.0.1").WithPort(proxy.Port))
+ .Build();
+
+ var ex = await Assert.ThrowsExactlyAsync(() => client.ConnectAsync(options));
+ Assert.AreEqual(MqttProxyErrorCode.ProxyProtocolError, ex.ErrorCode);
+ }
+
+ [TestMethod]
+ public async Task Connect_Response_With_NonZero_Reserved_Byte_Throws_ProtocolError()
+ {
+ await using var proxy = new FakeSocks5Server(new FakeSocks5ServerOptions
+ {
+ ReplyReservedByte = 0x01
+ });
+
+ using var client = new MqttClientFactory().CreateMqttClient();
+
+ var options = new MqttClientOptionsBuilder()
+ .WithTcpServer("127.0.0.1", 1883)
+ .WithSocks5Proxy(p => p.WithHost("127.0.0.1").WithPort(proxy.Port))
+ .Build();
+
+ var ex = await Assert.ThrowsExactlyAsync(() => client.ConnectAsync(options));
+ Assert.AreEqual(MqttProxyErrorCode.ProxyProtocolError, ex.ErrorCode);
+ }
+
+ [TestMethod]
+ public async Task Reconnect_After_Proxy_Disconnect_Works()
+ {
+ var broker = await Socks5TestHelper.StartBrokerAsync();
+ var brokerPort = Socks5TestHelper.GetBrokerPort(broker);
+
+ await using var proxy = new FakeSocks5Server(new FakeSocks5ServerOptions());
+
+ using var client = new MqttClientFactory().CreateMqttClient();
+
+ var options = new MqttClientOptionsBuilder()
+ .WithTcpServer("127.0.0.1", brokerPort)
+ .WithSocks5Proxy(p => p.WithHost("127.0.0.1").WithPort(proxy.Port))
+ .Build();
+
+ try
+ {
+ await client.ConnectAsync(options);
+ Assert.IsTrue(client.IsConnected);
+
+ // Force the broker side to drop the client by stopping & restarting it (proxy
+ // forwarding closes when the upstream connection closes).
+ await broker.Server.StopAsync();
+ broker.Server.Dispose();
+
+ await Task.Delay(500);
+
+ broker = await Socks5TestHelper.StartBrokerAsync();
+ var newPort = Socks5TestHelper.GetBrokerPort(broker);
+
+ options = new MqttClientOptionsBuilder()
+ .WithTcpServer("127.0.0.1", newPort)
+ .WithSocks5Proxy(p => p.WithHost("127.0.0.1").WithPort(proxy.Port))
+ .Build();
+
+ await client.ConnectAsync(options);
+ Assert.IsTrue(client.IsConnected);
+
+ // Two distinct CONNECT requests should have hit the proxy.
+ Assert.IsGreaterThanOrEqualTo(2, proxy.RecordedConnects.Count);
+ }
+ finally
+ {
+ try { await client.DisconnectAsync(); } catch { /* ignore */ }
+ await broker.Server.StopAsync();
+ broker.Server.Dispose();
+ }
+ }
+
+ [TestMethod]
+ public async Task Connect_Tls_Via_Socks5_PubSub_Works()
+ {
+ var tlsBroker = await Socks5TestHelper.StartTlsBrokerAsync();
+ try
+ {
+ using (tlsBroker.Certificate)
+ {
+ var brokerPort = Socks5TestHelper.GetTlsBrokerPort(tlsBroker);
+
+ await using var proxy = new FakeSocks5Server(new FakeSocks5ServerOptions());
+
+ using var client = new MqttClientFactory().CreateMqttClient();
+
+ var receivedTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ client.ApplicationMessageReceivedAsync += e =>
+ {
+ receivedTcs.TrySetResult(Encoding.UTF8.GetString(e.ApplicationMessage.Payload.ToArray()));
+ return Task.CompletedTask;
+ };
+
+ var options = new MqttClientOptionsBuilder()
+ .WithTcpServer("localhost", brokerPort)
+ .WithTlsOptions(o => o
+ .WithSslProtocols(System.Security.Authentication.SslProtocols.Tls12)
+ .WithCertificateValidationHandler(_ => true))
+ .WithSocks5Proxy(p => p.WithHost("127.0.0.1").WithPort(proxy.Port))
+ .Build();
+
+ await client.ConnectAsync(options);
+ Assert.IsTrue(client.IsConnected);
+
+ await client.SubscribeAsync("tls/socks5");
+ await client.PublishStringAsync("tls/socks5", "tls-hello");
+
+ var payload = await receivedTcs.Task.WaitAsync(TimeSpan.FromSeconds(5));
+ Assert.AreEqual("tls-hello", payload);
+
+ // Validate that the proxy saw a CONNECT with domain forwarding.
+ Assert.HasCount(1, proxy.RecordedConnects);
+ Assert.AreEqual(0x03, proxy.RecordedConnects[0].AddressType);
+ Assert.AreEqual("localhost", proxy.RecordedConnects[0].Host);
+
+ try { await client.DisconnectAsync(); } catch { /* ignore */ }
+ }
+ }
+ finally
+ {
+ await tlsBroker.Server.StopAsync();
+ tlsBroker.Server.Dispose();
+ }
+ }
+
+ [TestMethod]
+ public async Task Connect_Tcp_Via_Socks5_When_Command_Not_Supported_Maps_To_CommandNotSupported()
+ {
+ // Server replies with REP=0x07 ("Command not supported / protocol error") per RFC 1928 §6.
+ await using var proxy = new FakeSocks5Server(new FakeSocks5ServerOptions
+ {
+ ConnectReplyCode = 0x07
+ });
+
+ using var client = new MqttClientFactory().CreateMqttClient();
+
+ var options = new MqttClientOptionsBuilder()
+ .WithTcpServer("broker.example", 1883)
+ .WithSocks5Proxy(p => p.WithHost("127.0.0.1").WithPort(proxy.Port))
+ .Build();
+
+ var ex = await Assert.ThrowsExactlyAsync(() => client.ConnectAsync(options));
+ Assert.AreEqual(MqttProxyErrorCode.ProxyCommandNotSupported, ex.ErrorCode);
+ }
+
+ [TestMethod]
+ public async Task Connect_Tcp_Via_Socks5_Parses_DomainName_BindAddress_In_Reply()
+ {
+ // RFC 1928 §6 allows ATYP=0x03 (DOMAINNAME) in the CONNECT reply with a variable-length
+ // BND.ADDR. Verify the client correctly parses the length-prefixed bind address and
+ // completes the handshake even when the server uses a non-IP BND.ADDR.
+ var broker = await Socks5TestHelper.StartBrokerAsync();
+ var brokerPort = Socks5TestHelper.GetBrokerPort(broker);
+
+ await using var proxy = new FakeSocks5Server(new FakeSocks5ServerOptions
+ {
+ ReplyAddressType = 0x03,
+ ReplyBindHost = "proxy.example.internal",
+ ReplyBindPort = 4242
+ });
+
+ using var client = new MqttClientFactory().CreateMqttClient();
+
+ var connectOptions = new MqttClientOptionsBuilder()
+ .WithTcpServer("127.0.0.1", brokerPort)
+ .WithSocks5Proxy(p => p.WithHost("127.0.0.1").WithPort(proxy.Port))
+ .Build();
+
+ try
+ {
+ await client.ConnectAsync(connectOptions);
+ Assert.IsTrue(client.IsConnected);
+ }
+ finally
+ {
+ try { await client.DisconnectAsync(); } catch { /* ignore */ }
+ await broker.Server.StopAsync();
+ broker.Server.Dispose();
+ }
+ }
+}
diff --git a/Source/MQTTnet.Tests.Extensions.Socks5/Socks5ProxyOptionsValidation_Tests.cs b/Source/MQTTnet.Tests.Extensions.Socks5/Socks5ProxyOptionsValidation_Tests.cs
new file mode 100644
index 00000000..06d82cdc
--- /dev/null
+++ b/Source/MQTTnet.Tests.Extensions.Socks5/Socks5ProxyOptionsValidation_Tests.cs
@@ -0,0 +1,112 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using MQTTnet.Extensions.Socks5;
+
+namespace MQTTnet.Tests.Extensions.Socks5;
+
+[TestClass]
+public sealed class Socks5ProxyOptionsValidation_Tests
+{
+ [TestMethod]
+ public void Builder_Build_Without_Host_Throws()
+ {
+ var builder = new Socks5ProxyOptionsBuilder();
+
+ Assert.ThrowsExactly(() => builder.Build());
+ }
+
+ [TestMethod]
+ public void Builder_Build_With_Invalid_Port_Throws()
+ {
+ var builder = new Socks5ProxyOptionsBuilder()
+ .WithHost("127.0.0.1")
+ .WithPort(0);
+
+ Assert.ThrowsExactly(() => builder.Build());
+ }
+
+ [TestMethod]
+ public void Builder_Build_With_NonPositive_HandshakeTimeout_Throws()
+ {
+ var builder = new Socks5ProxyOptionsBuilder()
+ .WithHost("127.0.0.1")
+ .WithHandshakeTimeout(TimeSpan.Zero);
+
+ Assert.ThrowsExactly(() => builder.Build());
+ }
+
+ [TestMethod]
+ public void Builder_Build_With_Password_But_Without_Username_Throws()
+ {
+ var builder = new Socks5ProxyOptionsBuilder()
+ .WithHost("127.0.0.1")
+ .WithCredentials(null, new byte[] { 0x01, 0x02 });
+
+ Assert.ThrowsExactly(() => builder.Build());
+ }
+
+ [TestMethod]
+ public void Builder_Build_With_Username_But_Without_Password_Throws()
+ {
+ var builder = new Socks5ProxyOptionsBuilder()
+ .WithHost("127.0.0.1")
+ .WithCredentials("alice", Array.Empty());
+
+ Assert.ThrowsExactly(() => builder.Build());
+ }
+
+ [TestMethod]
+ public void StreamProvider_Ctor_With_Invalid_Port_Throws()
+ {
+ var options = new Socks5ProxyOptions
+ {
+ Host = "127.0.0.1",
+ Port = 70000
+ };
+
+ Assert.ThrowsExactly(() => _ = new Socks5StreamProvider(options));
+ }
+
+ [TestMethod]
+ public void StreamProvider_Ctor_With_NonPositive_HandshakeTimeout_Throws()
+ {
+ var options = new Socks5ProxyOptions
+ {
+ Host = "127.0.0.1",
+ Port = 1080,
+ HandshakeTimeout = TimeSpan.Zero
+ };
+
+ Assert.ThrowsExactly(() => _ = new Socks5StreamProvider(options));
+ }
+
+ [TestMethod]
+ public void StreamProvider_Ctor_With_Password_But_Without_Username_Throws()
+ {
+ var options = new Socks5ProxyOptions
+ {
+ Host = "127.0.0.1",
+ Port = 1080,
+ Password = new byte[] { 0x01 }
+ };
+
+ Assert.ThrowsExactly(() => _ = new Socks5StreamProvider(options));
+ }
+
+ [TestMethod]
+ public void StreamProvider_Ctor_With_Username_But_Without_Password_Throws()
+ {
+ var options = new Socks5ProxyOptions
+ {
+ Host = "127.0.0.1",
+ Port = 1080,
+ Username = "alice",
+ Password = Array.Empty()
+ };
+
+ Assert.ThrowsExactly(() => _ = new Socks5StreamProvider(options));
+ }
+}
diff --git a/Source/MQTTnet.Tests.Extensions.Socks5/Socks5TestHelper.cs b/Source/MQTTnet.Tests.Extensions.Socks5/Socks5TestHelper.cs
new file mode 100644
index 00000000..0a70c3b3
--- /dev/null
+++ b/Source/MQTTnet.Tests.Extensions.Socks5/Socks5TestHelper.cs
@@ -0,0 +1,97 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.IO;
+using System.Net;
+using System.Net.Security;
+using System.Net.Sockets;
+using System.Security.Cryptography;
+using System.Security.Cryptography.X509Certificates;
+using System.Threading.Tasks;
+using MQTTnet.Certificates;
+using MQTTnet.Server;
+
+namespace MQTTnet.Tests.Extensions.Socks5;
+
+internal static class Socks5TestHelper
+{
+ public static async Task StartBrokerAsync()
+ {
+ var factory = new MqttServerFactory();
+ var options = factory.CreateServerOptionsBuilder()
+ .WithDefaultEndpoint()
+ .WithDefaultEndpointPort(0)
+ .Build();
+
+ var server = factory.CreateMqttServer(options);
+ await server.StartAsync();
+ return new BrokerHandle(server, options);
+ }
+
+ public static int GetBrokerPort(BrokerHandle handle)
+ {
+ return handle.Options.DefaultEndpointOptions.Port;
+ }
+
+ public static async Task StartTlsBrokerAsync()
+ {
+ var certificate = CreateSelfSignedServerCertificate();
+
+ var factory = new MqttServerFactory();
+ var options = factory.CreateServerOptionsBuilder()
+ .WithoutDefaultEndpoint()
+ .WithEncryptedEndpoint()
+ .WithEncryptedEndpointPort(0)
+ .WithEncryptionCertificate(new InMemoryCertificateProvider(certificate))
+ .Build();
+
+ var server = factory.CreateMqttServer(options);
+ await server.StartAsync();
+ return new TlsBrokerHandle(server, options, certificate);
+ }
+
+ public static int GetTlsBrokerPort(TlsBrokerHandle handle)
+ {
+ return handle.Options.TlsEndpointOptions.Port;
+ }
+
+ public static X509Certificate2 CreateSelfSignedServerCertificate()
+ {
+ var sanBuilder = new SubjectAlternativeNameBuilder();
+ sanBuilder.AddIpAddress(IPAddress.Loopback);
+ sanBuilder.AddIpAddress(IPAddress.IPv6Loopback);
+ sanBuilder.AddDnsName("localhost");
+
+ using var rsa = RSA.Create(2048);
+ var certRequest = new CertificateRequest("CN=localhost", rsa, HashAlgorithmName.SHA256, RSASignaturePadding.Pkcs1);
+ certRequest.CertificateExtensions.Add(
+ new X509KeyUsageExtension(X509KeyUsageFlags.DataEncipherment | X509KeyUsageFlags.KeyEncipherment | X509KeyUsageFlags.DigitalSignature, false));
+ certRequest.CertificateExtensions.Add(sanBuilder.Build());
+
+ using var cert = certRequest.CreateSelfSigned(DateTimeOffset.UtcNow.AddMinutes(-10), DateTimeOffset.UtcNow.AddMinutes(60));
+#pragma warning disable SYSLIB0057
+ return new X509Certificate2(
+ cert.Export(X509ContentType.Pfx),
+ (string)null,
+ X509KeyStorageFlags.MachineKeySet | X509KeyStorageFlags.Exportable);
+#pragma warning restore SYSLIB0057
+ }
+
+ internal sealed record BrokerHandle(MqttServer Server, MqttServerOptions Options);
+
+ internal sealed record TlsBrokerHandle(MqttServer Server, MqttServerOptions Options, X509Certificate2 Certificate);
+
+ sealed class InMemoryCertificateProvider : ICertificateProvider
+ {
+ readonly X509Certificate2 _certificate;
+
+ public InMemoryCertificateProvider(X509Certificate2 certificate)
+ {
+ _certificate = certificate;
+ }
+
+ public X509Certificate2 GetCertificate() => _certificate;
+ }
+}
diff --git a/Source/MQTTnet/Channel/IMqttClientStreamEndPoints.cs b/Source/MQTTnet/Channel/IMqttClientStreamEndPoints.cs
new file mode 100644
index 00000000..ef9f8d58
--- /dev/null
+++ b/Source/MQTTnet/Channel/IMqttClientStreamEndPoints.cs
@@ -0,0 +1,27 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Net;
+
+namespace MQTTnet.Channel;
+
+///
+/// Optional marker interface that a returned by an
+/// may implement to expose the local and remote
+/// socket endpoints to the MQTT TCP channel (used for diagnostics only).
+///
+public interface IMqttClientStreamEndPoints
+{
+ ///
+ /// Gets the local endpoint of the socket actually opened by the provider
+ /// (e.g. the local interface used to reach the proxy). May be null if not known.
+ ///
+ EndPoint LocalEndPoint { get; }
+
+ ///
+ /// Gets the remote endpoint of the socket actually opened by the provider
+ /// (e.g. the proxy address). May be null if not known.
+ ///
+ EndPoint RemoteEndPoint { get; }
+}
diff --git a/Source/MQTTnet/Channel/IMqttClientStreamProvider.cs b/Source/MQTTnet/Channel/IMqttClientStreamProvider.cs
new file mode 100644
index 00000000..d8490116
--- /dev/null
+++ b/Source/MQTTnet/Channel/IMqttClientStreamProvider.cs
@@ -0,0 +1,50 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Net;
+
+namespace MQTTnet.Channel;
+
+///
+/// Produces a connected byte that carries traffic to an MQTT broker endpoint.
+/// This is the extension point used to plug stream-level proxies (SOCKS5, HTTP CONNECT, etc.)
+/// into the TCP transport without modifying MqttTcpChannel.
+/// Implementations are responsible for the full handshake to the broker (proxy negotiation,
+/// authentication, target connect) but MUST NOT perform any TLS wrapping; TLS is layered on
+/// top of the returned stream by the channel itself, so the broker host can be used as SNI.
+///
+///
+///
+/// The returned is owned by the caller (the MQTT channel). Disposing
+/// it will release any underlying socket. Providers should therefore avoid keeping their
+/// own references to the underlying socket after returning.
+///
+///
+/// Implementations may optionally have the returned stream implement
+/// to expose the local and remote socket
+/// endpoints to MqttTcpChannel for diagnostics.
+///
+///
+/// is invoked on every (re)connect attempt; implementations
+/// must be safe to call concurrently or sequentially across multiple client instances.
+///
+///
+public interface IMqttClientStreamProvider
+{
+ ///
+ /// Establishes a connection to the MQTT broker , performing
+ /// any required proxy handshake, and returns the resulting raw byte stream.
+ ///
+ ///
+ /// The MQTT broker endpoint. Can be a (in which case the
+ /// provider may forward the hostname for remote DNS resolution) or an
+ /// .
+ ///
+ /// Token to observe for cancellation.
+ ///
+ /// A connected whose bytes are transported end-to-end to
+ /// .
+ ///
+ Task ConnectAsync(EndPoint brokerEndPoint, CancellationToken cancellationToken);
+}
diff --git a/Source/MQTTnet/Exceptions/MqttProxyErrorCode.cs b/Source/MQTTnet/Exceptions/MqttProxyErrorCode.cs
new file mode 100644
index 00000000..236e35b6
--- /dev/null
+++ b/Source/MQTTnet/Exceptions/MqttProxyErrorCode.cs
@@ -0,0 +1,52 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+namespace MQTTnet.Exceptions;
+
+///
+/// Categorized error codes for proxy-level failures surfaced via .
+/// The first set of values map 1:1 to SOCKS5 REP codes (RFC 1928 §6); additional values cover
+/// non-protocol conditions (connecting to the proxy itself, authentication, protocol violations).
+///
+public enum MqttProxyErrorCode
+{
+ /// The error category could not be classified.
+ ProxyUnknownError = 0,
+
+ /// The TCP connection to the proxy could not be established (DNS, refused, timeout).
+ ProxyUnreachable,
+
+ /// The proxy returned a response that does not conform to the expected protocol.
+ ProxyProtocolError,
+
+ /// The proxy rejected every authentication method offered by the client (SOCKS5 method = 0xFF).
+ ProxyAuthMethodRejected,
+
+ /// The proxy rejected the supplied credentials.
+ ProxyAuthFailed,
+
+ /// SOCKS5 REP = 0x01: general SOCKS server failure.
+ ProxyGeneralFailure,
+
+ /// SOCKS5 REP = 0x02: connection not allowed by ruleset.
+ ProxyConnectionNotAllowed,
+
+ /// SOCKS5 REP = 0x03: network unreachable.
+ ProxyNetworkUnreachable,
+
+ /// SOCKS5 REP = 0x04: host unreachable.
+ ProxyHostUnreachable,
+
+ /// SOCKS5 REP = 0x05: connection refused by destination.
+ ProxyConnectionRefused,
+
+ /// SOCKS5 REP = 0x06: TTL expired.
+ ProxyTtlExpired,
+
+ /// SOCKS5 REP = 0x07: command not supported by the proxy.
+ ProxyCommandNotSupported,
+
+ /// SOCKS5 REP = 0x08: address type not supported by the proxy.
+ ProxyAddressTypeNotSupported
+}
diff --git a/Source/MQTTnet/Exceptions/MqttProxyException.cs b/Source/MQTTnet/Exceptions/MqttProxyException.cs
new file mode 100644
index 00000000..80ed46d3
--- /dev/null
+++ b/Source/MQTTnet/Exceptions/MqttProxyException.cs
@@ -0,0 +1,39 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+namespace MQTTnet.Exceptions;
+
+///
+/// Thrown when an fails while
+/// establishing the stream to the broker (proxy unreachable, authentication failure,
+/// destination unreachable, protocol violation, ...).
+/// Inherits from so existing connect/reconnect
+/// handling continues to work; callers that care about the specific reason can branch on
+/// .
+///
+public sealed class MqttProxyException : MqttCommunicationException
+{
+ public MqttProxyException(MqttProxyErrorCode errorCode, string message, string proxyAddress = null, Exception innerException = null)
+ : base(BuildMessage(errorCode, message, proxyAddress), innerException)
+ {
+ ErrorCode = errorCode;
+ ProxyAddress = proxyAddress;
+ }
+
+ /// Gets the categorized error reason.
+ public MqttProxyErrorCode ErrorCode { get; }
+
+ /// Gets the proxy address (host:port) associated with the failure, if known.
+ public string ProxyAddress { get; }
+
+ static string BuildMessage(MqttProxyErrorCode errorCode, string message, string proxyAddress)
+ {
+ if (string.IsNullOrEmpty(proxyAddress))
+ {
+ return $"Proxy error '{errorCode}': {message}";
+ }
+
+ return $"Proxy '{proxyAddress}' error '{errorCode}': {message}";
+ }
+}
diff --git a/Source/MQTTnet/Implementations/MqttTcpChannel.cs b/Source/MQTTnet/Implementations/MqttTcpChannel.cs
index f8466837..4fa293ea 100644
--- a/Source/MQTTnet/Implementations/MqttTcpChannel.cs
+++ b/Source/MQTTnet/Implementations/MqttTcpChannel.cs
@@ -55,6 +55,12 @@ public sealed class MqttTcpChannel : IMqttChannel
public async Task ConnectAsync(CancellationToken cancellationToken)
{
+ if (_tcpOptions.StreamProvider != null)
+ {
+ await ConnectViaStreamProviderAsync(cancellationToken).ConfigureAwait(false);
+ return;
+ }
+
CrossPlatformSocket socket = null;
try
{
@@ -103,55 +109,7 @@ public sealed class MqttTcpChannel : IMqttChannel
if (_tcpOptions.TlsOptions?.UseTls == true)
{
- var targetHost = _tcpOptions.TlsOptions.TargetHost;
- if (string.IsNullOrEmpty(targetHost))
- {
- if (_tcpOptions.RemoteEndpoint is DnsEndPoint dns)
- {
- targetHost = dns.Host;
- }
- }
-
- SslStream sslStream;
- if (_tcpOptions.TlsOptions.CertificateSelectionHandler != null)
- {
- sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback, InternalUserCertificateSelectionCallback);
- }
- else
- {
- // Use a different constructor depending on the options for MQTTnet so that we do not have
- // to copy the exact same behavior of the selection handler.
- sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback);
- }
-
- try
- {
- var sslOptions = CreateSslAuthenticationOptions();
-
- sslOptions.TargetHost = targetHost;
-
- if (_tcpOptions.TlsOptions.TrustChain?.Count > 0)
- {
- sslOptions.CertificateChainPolicy = new X509ChainPolicy
- {
- TrustMode = X509ChainTrustMode.CustomRootTrust,
- VerificationFlags = X509VerificationFlags.IgnoreEndRevocationUnknown,
- RevocationMode = _tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.NoCheck : _tcpOptions.TlsOptions.RevocationMode
- };
-
- sslOptions.CertificateChainPolicy.CustomTrustStore.AddRange(_tcpOptions.TlsOptions.TrustChain);
- }
-
- await sslStream.AuthenticateAsClientAsync(sslOptions, cancellationToken).ConfigureAwait(false);
- }
- catch
- {
- await sslStream.DisposeAsync().ConfigureAwait(false);
-
- throw;
- }
-
- _stream = sslStream;
+ _stream = await WrapWithTlsAsync(networkStream, cancellationToken).ConfigureAwait(false);
}
else
{
@@ -168,6 +126,106 @@ public sealed class MqttTcpChannel : IMqttChannel
}
}
+ async Task ConnectViaStreamProviderAsync(CancellationToken cancellationToken)
+ {
+ Stream providerStream = null;
+ try
+ {
+ providerStream = await _tcpOptions.StreamProvider.ConnectAsync(_tcpOptions.RemoteEndpoint, cancellationToken).ConfigureAwait(false);
+ if (providerStream == null)
+ {
+ throw new MqttCommunicationException("The configured stream provider returned a null stream.");
+ }
+
+ cancellationToken.ThrowIfCancellationRequested();
+
+ if (_tcpOptions.TlsOptions?.UseTls == true)
+ {
+ _stream = await WrapWithTlsAsync(providerStream, cancellationToken).ConfigureAwait(false);
+ }
+ else
+ {
+ _stream = providerStream;
+ }
+
+ // Endpoints are diagnostic only. Prefer the provider's reported values if available;
+ // fall back to the broker endpoint as the "remote" so logs/diagnostics still show
+ // something meaningful when the provider does not expose its socket endpoints.
+ if (providerStream is IMqttClientStreamEndPoints connectedStream)
+ {
+ LocalEndPoint = connectedStream.LocalEndPoint;
+ RemoteEndPoint = connectedStream.RemoteEndPoint ?? _tcpOptions.RemoteEndpoint;
+ }
+ else
+ {
+ LocalEndPoint = null;
+ RemoteEndPoint = _tcpOptions.RemoteEndpoint;
+ }
+ }
+ catch
+ {
+ if (_stream == null)
+ {
+ providerStream?.Dispose();
+ }
+
+ throw;
+ }
+ }
+
+ async Task WrapWithTlsAsync(Stream baseStream, CancellationToken cancellationToken)
+ {
+ var targetHost = _tcpOptions.TlsOptions.TargetHost;
+ if (string.IsNullOrEmpty(targetHost))
+ {
+ if (_tcpOptions.RemoteEndpoint is DnsEndPoint dns)
+ {
+ targetHost = dns.Host;
+ }
+ }
+
+ SslStream sslStream;
+ if (_tcpOptions.TlsOptions.CertificateSelectionHandler != null)
+ {
+ sslStream = new SslStream(baseStream, false, InternalUserCertificateValidationCallback, InternalUserCertificateSelectionCallback);
+ }
+ else
+ {
+ // Use a different constructor depending on the options for MQTTnet so that we do not have
+ // to copy the exact same behavior of the selection handler.
+ sslStream = new SslStream(baseStream, false, InternalUserCertificateValidationCallback);
+ }
+
+ try
+ {
+ var sslOptions = CreateSslAuthenticationOptions();
+
+ sslOptions.TargetHost = targetHost;
+
+ if (_tcpOptions.TlsOptions.TrustChain?.Count > 0)
+ {
+ sslOptions.CertificateChainPolicy = new X509ChainPolicy
+ {
+ TrustMode = X509ChainTrustMode.CustomRootTrust,
+ VerificationFlags = X509VerificationFlags.IgnoreEndRevocationUnknown,
+ RevocationMode = _tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.NoCheck : _tcpOptions.TlsOptions.RevocationMode
+ };
+
+ sslOptions.CertificateChainPolicy.CustomTrustStore.AddRange(_tcpOptions.TlsOptions.TrustChain);
+ }
+
+ await sslStream.AuthenticateAsClientAsync(sslOptions, cancellationToken).ConfigureAwait(false);
+ }
+ catch
+ {
+ await sslStream.DisposeAsync().ConfigureAwait(false);
+
+ throw;
+ }
+
+ return sslStream;
+ }
+
public Task DisconnectAsync(CancellationToken cancellationToken)
{
Dispose();
diff --git a/Source/MQTTnet/Options/MqttClientOptionsBuilder.cs b/Source/MQTTnet/Options/MqttClientOptionsBuilder.cs
index 8c7e99bc..b7dd6d93 100644
--- a/Source/MQTTnet/Options/MqttClientOptionsBuilder.cs
+++ b/Source/MQTTnet/Options/MqttClientOptionsBuilder.cs
@@ -6,6 +6,7 @@ using System;
using System.Net;
using System.Net.Sockets;
using System.Text;
+using MQTTnet.Channel;
using MQTTnet.Formatter;
using MQTTnet.Packets;
using MQTTnet.Protocol;
@@ -252,6 +253,31 @@ public sealed class MqttClientOptionsBuilder
return this;
}
+ ///
+ /// Configures an optional on the TCP options.
+ /// The provider is responsible for opening the transport stream to the broker (e.g. through
+ /// a SOCKS5 proxy). When the connection should be TLS-secured, configure TLS via
+ /// WithTlsOptions(...); TLS will be layered on top of the stream returned by the
+ /// provider, using the broker host as SNI.
+ ///
+ ///
+ /// Thrown when no TCP transport has been configured yet (call WithTcpServer or
+ /// WithEndPoint before this method).
+ ///
+ public MqttClientOptionsBuilder WithStreamProvider(IMqttClientStreamProvider provider)
+ {
+ ArgumentNullException.ThrowIfNull(provider);
+
+ if (_tcpOptions == null)
+ {
+ throw new InvalidOperationException(
+ "A TCP transport must be configured (via WithTcpServer or WithEndPoint) before assigning a stream provider.");
+ }
+
+ _tcpOptions.StreamProvider = provider;
+ return this;
+ }
+
public MqttClientOptionsBuilder WithProtocolVersion(MqttProtocolVersion value)
{
if (value == MqttProtocolVersion.Unknown)
diff --git a/Source/MQTTnet/Options/MqttClientTcpOptions.cs b/Source/MQTTnet/Options/MqttClientTcpOptions.cs
index d0856f3b..f757391f 100644
--- a/Source/MQTTnet/Options/MqttClientTcpOptions.cs
+++ b/Source/MQTTnet/Options/MqttClientTcpOptions.cs
@@ -4,6 +4,7 @@
using System.Net;
using System.Net.Sockets;
+using MQTTnet.Channel;
namespace MQTTnet;
@@ -44,6 +45,15 @@ public sealed class MqttClientTcpOptions : IMqttClientChannelOptions
public EndPoint RemoteEndpoint { get; set; }
+ ///
+ /// Gets or sets an optional used to obtain the
+ /// transport stream to the broker. When set, the channel will delegate the TCP-level
+ /// connect (including any proxy negotiation) to the provider and then optionally wrap the
+ /// returned stream with TLS, using the broker host as SNI. When null, the channel
+ /// opens a plain TCP socket directly to .
+ ///
+ public IMqttClientStreamProvider StreamProvider { get; set; }
+
public MqttClientTlsOptions TlsOptions { get; set; } = new();
public override string ToString()