Validate MQTT 5 SUBSCRIBE identifiers (#2262)

Observe packet task failures to prevent unobserved overflow exceptions

Server dispatch and session enqueue operations create packet bus items without
waiting for delivery. When the pending-message queue overflows, Fail faults
the item's completion task, which can later raise UnobservedTaskException
even if the application handles the dropped-message notification.

Observe the completion task's exception inside MqttPacketBusItem.Fail. This
keeps the task faulted and preserves the original exception for existing and
future WaitAsync callers, including DeliverApplicationMessageAsync.

Add a garbage-collection regression test for failures without a waiter and
verify that requesting the task before or after failure preserves the original
overflow exception. Re-enable unobserved-exception tracking in the existing
overflow integration tests for both drop strategies.

Before the fix, the focused run failed the new unobserved-exception test and
both enqueue overflow tests. After the fix, all nine focused cases and all
489 tests in the .NET 10 Release test suite pass.

Fixes https://github.com/dotnet/MQTTnet/issues/2271

.github/workflows/ci.yml

@@ -3,7 +3,7 @@ name: CI
 on: [push, pull_request]
 
 env:
-  VERSION: "5.1.0.${{github.run_number}}"
+  VERSION: "5.2.0.${{github.run_number}}"
   PACKAGE_SUFFIX: ""
 
 jobs:
@@ -17,19 +17,26 @@ jobs:
         with:
           dotnet-version: |
             8.0.x
+            10.0.x
 
       - name: Checkout code
         uses: actions/checkout@v4
 
       - name: Load strong name certificate
+        if: ${{ github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository }}
         run: |
           echo "$SNC_BASE64" | base64 --decode > "${{ github.workspace }}/certificate.snk"
         shell: bash
         env:
           SNC_BASE64: ${{ secrets.SNC_BASE64 }}
 
-      - name: Build package
+      - name: Build package (signed)
-        run: dotnet build MQTTnet.sln --configuration Release /p:FileVersion=${{ env.VERSION }} /p:AssemblyVersion=${{ env.VERSION }} /p:PackageVersion=${{ env.VERSION }}${{ env.PACKAGE_SUFFIX }} /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=${{ github.workspace }}/certificate.snk
+        if: ${{ github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository }}
+        run: dotnet build MQTTnet.slnx --configuration Release /p:FileVersion=${{ env.VERSION }} /p:AssemblyVersion=${{ env.VERSION }} /p:PackageVersion=${{ env.VERSION }}${{ env.PACKAGE_SUFFIX }} /p:SignAssembly=true /p:AssemblyOriginatorKeyFile=${{ github.workspace }}/certificate.snk
+
+      - name: Build package (unsigned)
+        if: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository }}
+        run: dotnet build MQTTnet.slnx --configuration Release /p:FileVersion=${{ env.VERSION }} /p:AssemblyVersion=${{ env.VERSION }} /p:PackageVersion=${{ env.VERSION }}${{ env.PACKAGE_SUFFIX }}
 
       - name: Upload nuget packages
         uses: actions/upload-artifact@v4
@@ -48,24 +55,19 @@ jobs:
         with:
           dotnet-version: |
             8.0.x
+            10.0.x
 
       - name: Checkout code
         uses: actions/checkout@v4
 
       - name: Execute tests
-        run: dotnet test --framework net8.0 --configuration Release Source/MQTTnet.Tests/MQTTnet.Tests.csproj
+        run: dotnet test --framework net10.0 --configuration Release --project Source/MQTTnet.Tests/MQTTnet.Tests.csproj
 
   sign:
     needs: build
     runs-on: windows-latest # Code signing must run on a Windows agent for Authenticode signing (dll/exe)
-    if: github.repository == 'dotnet/MQTTnet'
+    if: github.repository == 'dotnet/MQTTnet' && (github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository)
     steps:
-      - name: Setup .NET SDK
-        uses: actions/setup-dotnet@v4
-        with:
-          dotnet-version: |
-            6.0.x
-
       - name: Download nuget packages
         uses: actions/download-artifact@v4
         with:
@@ -73,7 +75,7 @@ jobs:
           path: nugets
 
       - name: Install sign CLI tool
-        run: dotnet tool install --tool-path . sign --version 0.9.0-beta.23127.3
+        run: dotnet tool install --tool-path . sign --version 0.9.1-beta.25379.1
 
       - name: Sign nugets
         shell: pwsh
@@ -114,7 +116,7 @@ jobs:
         run: dotnet nuget push ${{ github.workspace }}/nugets/**/*.nupkg -k ${{ secrets.MYGET_API_KEY }} -s https://www.myget.org/F/mqttnet/api/v3/index.json --skip-duplicate
 
   publish-nuget:
-    # Tag schema is 'v5.0.0-rc' etc.
+    # Tag schema is 'v5.1.0-rc' etc.
     if: startsWith(github.ref, 'refs/tags/v')
     needs: [sign, test]
     runs-on: ubuntu-latest

Directory.Build.props

@@ -1,6 +1,6 @@
 <Project>
     <PropertyGroup>
-        <TargetFramework>net8.0</TargetFramework>
+        <TargetFrameworks>net8.0;net10.0</TargetFrameworks>
         <Product>MQTTnet</Product>
         <Authors>The contributors of MQTTnet</Authors>
         <Company>The contributors of MQTTnet</Company>
@@ -11,6 +11,7 @@
         <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
         <EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild>
         <ImplicitUsings>enable</ImplicitUsings>
+        <Nullable>disable</Nullable>
 
         <NeutralLanguage>en-US</NeutralLanguage>
         <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
@@ -24,35 +25,10 @@
         <AnalysisLevel>latest</AnalysisLevel>
         <AnalysisMode>recommended</AnalysisMode>
         <AnalysisModeStyle>default</AnalysisModeStyle>
-
-        <TestingPlatformShowTestsFailure>true</TestingPlatformShowTestsFailure>
-        <TestingPlatformDotnetTestSupport>true</TestingPlatformDotnetTestSupport>
-        <EnableMSTestRunner>true</EnableMSTestRunner>
     </PropertyGroup>
 
-    <PropertyGroup Condition="'$(GITHUB_ACTIONS)' == 'true'">
+    <PropertyGroup Condition="'$(GITHUB_ACTIONS)'=='true'">
         <ContinuousIntegrationBuild>true</ContinuousIntegrationBuild>
     </PropertyGroup>
 
-    <PropertyGroup Condition="'$(GeneratePackageOnBuild)'=='true'">
-        <PackageLicenseFile>LICENSE</PackageLicenseFile>
-        <PackageIcon>nuget.png</PackageIcon>
-        <PackageProjectUrl>https://github.com/dotnet/MQTTnet</PackageProjectUrl>
-        <RepositoryUrl>https://github.com/dotnet/MQTTnet.git</RepositoryUrl>
-        <RepositoryType>git</RepositoryType>
-        <IncludeSymbols>true</IncludeSymbols>
-        <SymbolPackageFormat>snupkg</SymbolPackageFormat>
-        <PublishRepositoryUrl>true</PublishRepositoryUrl>
-        <EmbedUntrackedSources>true</EmbedUntrackedSources>
-    </PropertyGroup>
-
-    <ItemGroup Condition="'$(GeneratePackageOnBuild)'=='true'">
-        <None Include="README.md" Pack="true" PackagePath="" Condition="Exists('README.md')" />
-        <None Include="..\..\Images\nuget.png" Pack="true" PackagePath="" />
-        <None Include="..\..\LICENSE" Pack="true" PackagePath="" />
-    </ItemGroup>
-
-    <ItemGroup Condition="'$(IncludeSymbols)'=='true'">
-        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="8.0.0" PrivateAssets="All"/>
-    </ItemGroup>
 </Project>

Directory.Build.targets

@@ -0,0 +1,24 @@
+<Project>
+
+    <PropertyGroup Condition="'$(GeneratePackageOnBuild)'=='true'">
+        <PackageLicenseFile>LICENSE</PackageLicenseFile>
+        <PackageIcon>nuget.png</PackageIcon>
+        <PackageProjectUrl>https://github.com/dotnet/MQTTnet</PackageProjectUrl>
+        <RepositoryUrl>https://github.com/dotnet/MQTTnet.git</RepositoryUrl>
+        <RepositoryType>git</RepositoryType>
+        <IncludeSymbols>true</IncludeSymbols>
+        <SymbolPackageFormat>snupkg</SymbolPackageFormat>
+        <PublishRepositoryUrl>true</PublishRepositoryUrl>
+        <EmbedUntrackedSources>true</EmbedUntrackedSources>
+    </PropertyGroup>
+
+    <ItemGroup Condition="'$(GeneratePackageOnBuild)'=='true'">
+        <None Include="README.md" Pack="true" PackagePath="" Condition="Exists('README.md')" />
+        <None Include="..\..\Images\nuget.png" Pack="true" PackagePath="" />
+        <None Include="..\..\LICENSE" Pack="true" PackagePath="" />
+    </ItemGroup>
+
+    <ItemGroup Condition="'$(IncludeSymbols)'=='true'">
+        <PackageReference Include="Microsoft.SourceLink.GitHub" Version="8.0.0" PrivateAssets="All"/>
+    </ItemGroup>
+</Project>

MQTTnet.sln

@@ -1,95 +0,0 @@
-
-Microsoft Visual Studio Solution File, Format Version 12.00
-# Visual Studio Version 17
-VisualStudioVersion = 17.0.31919.166
-MinimumVisualStudioVersion = 10.0.40219.1
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet", "Source\MQTTnet\MQTTnet.csproj", "{3587E506-55A2-4EB3-99C7-DC01E42D25D2}"
-EndProject
-Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{B3F60ECB-45BA-4C66-8903-8BB89CA67998}"
-	ProjectSection(SolutionItems) = preProject
-		.github\workflows\ci.yml = .github\workflows\ci.yml
-		CODE-OF-CONDUCT.md = CODE-OF-CONDUCT.md
-		Directory.Build.props = Directory.Build.props
-		LICENSE = LICENSE
-		README.md = README.md
-		Source\ReleaseNotes.md = Source\ReleaseNotes.md
-	EndProjectSection
-EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.AspNetCore", "Source\MQTTnet.AspnetCore\MQTTnet.AspNetCore.csproj", "{F10C4060-F7EE-4A83-919F-FF723E72F94A}"
-EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Extensions.Rpc", "Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj", "{C444E9C8-95FA-430E-9126-274129DE16CD}"
-EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Samples", "Samples\MQTTnet.Samples.csproj", "{71CF35F5-3327-4A91-AAF4-5340F6701771}"
-EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Tests", "Source\MQTTnet.Tests\MQTTnet.Tests.csproj", "{B270F32A-9F3E-42EE-A989-813E35E29ADB}"
-EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Benchmarks", "Source\MQTTnet.Benchmarks\MQTTnet.Benchmarks.csproj", "{2F516E76-AAC4-4219-B7D1-34CDD3CFF381}"
-EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.TestApp", "Source\MQTTnet.TestApp\MQTTnet.TestApp.csproj", "{175D5340-CC5B-4542-939D-4E7D15A0BC8D}"
-EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.AspTestApp", "Source\MQTTnet.AspTestApp\MQTTnet.AspTestApp.csproj", "{72867E4C-4E15-4E8E-8FAB-AE9253286BBC}"
-EndProject
-Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Extensions.TopicTemplate", "Source\MQTTnet.Extensions.TopicTemplate\MQTTnet.Extensions.TopicTemplate.csproj", "{374D5861-F7A6-42A2-8CBE-040846000EF9}"
-EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.Server", "Source\MQTTnet.Server\MQTTnet.Server.csproj", "{C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}"
-EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MQTTnet.PowerShell", "Source\MQTTnet.PowerShell\MQTTnet.PowerShell.csproj", "{37B129C5-0231-4EB5-AF13-77959068911C}"
-EndProject
-Global
-	GlobalSection(SolutionConfigurationPlatforms) = preSolution
-		Debug|Any CPU = Debug|Any CPU
-		Release|Any CPU = Release|Any CPU
-	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}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{3587E506-55A2-4EB3-99C7-DC01E42D25D2}.Release|Any CPU.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}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{F10C4060-F7EE-4A83-919F-FF723E72F94A}.Release|Any CPU.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}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{C444E9C8-95FA-430E-9126-274129DE16CD}.Release|Any CPU.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}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{71CF35F5-3327-4A91-AAF4-5340F6701771}.Release|Any CPU.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}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{B270F32A-9F3E-42EE-A989-813E35E29ADB}.Release|Any CPU.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}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{2F516E76-AAC4-4219-B7D1-34CDD3CFF381}.Release|Any CPU.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}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{175D5340-CC5B-4542-939D-4E7D15A0BC8D}.Release|Any CPU.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}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{72867E4C-4E15-4E8E-8FAB-AE9253286BBC}.Release|Any CPU.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}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{374D5861-F7A6-42A2-8CBE-040846000EF9}.Release|Any CPU.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}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{C876EFFD-C5AD-4E26-BD9F-4C48252C02BC}.Release|Any CPU.Build.0 = Release|Any CPU
-		{37B129C5-0231-4EB5-AF13-77959068911C}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
-		{37B129C5-0231-4EB5-AF13-77959068911C}.Debug|Any CPU.Build.0 = Debug|Any CPU
-		{37B129C5-0231-4EB5-AF13-77959068911C}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{37B129C5-0231-4EB5-AF13-77959068911C}.Release|Any CPU.Build.0 = Release|Any CPU
-	EndGlobalSection
-	GlobalSection(SolutionProperties) = preSolution
-		HideSolutionNode = FALSE
-	EndGlobalSection
-	GlobalSection(ExtensibilityGlobals) = postSolution
-		SolutionGuid = {07536672-5CBC-4BE3-ACE0-708A431A7894}
-	EndGlobalSection
-EndGlobal

MQTTnet.slnx

@@ -0,0 +1,28 @@
+<Solution>
+  <Configurations>
+    <Platform Name="Any CPU" />
+    <Platform Name="x64" />
+    <Platform Name="x86" />
+  </Configurations>
+  <Folder Name="/Solution Items/">
+    <File Path=".github/workflows/ci.yml" />
+    <File Path="CODE-OF-CONDUCT.md" />
+    <File Path="Directory.Build.props" />
+    <File Path="Directory.Build.targets" />
+    <File Path="LICENSE" />
+    <File Path="README.md" />
+    <File Path="Source/ReleaseNotes.md" />
+  </Folder>
+  <Project Path="Samples/MQTTnet.Samples.csproj" />
+  <Project Path="Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj" />
+  <Project Path="Source/MQTTnet.AspTestApp/MQTTnet.AspTestApp.csproj" />
+  <Project Path="Source/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj" />
+  <Project Path="Source/MQTTnet.Extensions.Rpc/MQTTnet.Extensions.Rpc.csproj" />
+  <Project Path="Source/MQTTnet.Extensions.Socks5/MQTTnet.Extensions.Socks5.csproj" />
+  <Project Path="Source/MQTTnet.Extensions.TopicTemplate/MQTTnet.Extensions.TopicTemplate.csproj" />
+  <Project Path="Source/MQTTnet.Server/MQTTnet.Server.csproj" />
+  <Project Path="Source/MQTTnet.TestApp/MQTTnet.TestApp.csproj" />
+  <Project Path="Source/MQTTnet.Tests.Extensions.Socks5/MQTTnet.Tests.Extensions.Socks5.csproj" />
+  <Project Path="Source/MQTTnet.Tests/MQTTnet.Tests.csproj" />
+  <Project Path="Source/MQTTnet/MQTTnet.csproj" />
+</Solution>

README.md

@@ -12,9 +12,9 @@
 
 # MQTTnet
 
-MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (
+MQTTnet is a high performance .NET library for MQTT based communication.
-broker) and supports the MQTT protocol up to version 5. It is compatible with mostly any supported .NET Framework
+It provides a MQTT client and a MQTT server (broker) and supports the MQTT protocol up to version 5.
-version and CPU architecture.
+It is compatible with mostly any supported .NET Framework version and CPU architecture.
 
 ## Features
 

Samples/Client/Client_Publish_Samples.cs

@@ -6,6 +6,9 @@
 // ReSharper disable UnusedMember.Global
 // ReSharper disable InconsistentNaming
 
+using System;
+using System.Text;
+
 namespace MQTTnet.Samples.Client;
 
 public static class Client_Publish_Samples
@@ -84,4 +87,21 @@ public static class Client_Publish_Samples
 
         Console.WriteLine("MQTT application message is published.");
     }
+
+    public static MqttApplicationMessage Create_Message_With_Binary_User_Property()
+    {
+        /*
+         * MQTT v5 user properties are encoded as UTF-8 strings. When the UTF-8 payload is already available
+         * as a byte buffer, the builder APIs can avoid creating intermediate strings.
+         */
+
+        var encodedValue = Encoding.UTF8.GetBytes("sensor-01");
+
+        return new MqttApplicationMessageBuilder()
+            .WithTopic("samples/metadata/binary")
+            .WithUserProperty("client-id", encodedValue.AsMemory())
+            .WithUserProperty("checksum", new ArraySegment<byte>(encodedValue))
+            .WithPayload("metadata")
+            .Build();
+    }
 }

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);
+    }
+}

Samples/Client/Client_Subscribe_Samples.cs

@@ -7,7 +7,7 @@
 // ReSharper disable InconsistentNaming
 // ReSharper disable UnusedMember.Local
 
-using MQTTnet.Extensions.TopicTemplate;
+using System.Text;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
 using MQTTnet.Samples.Helpers;
@@ -16,8 +16,6 @@ namespace MQTTnet.Samples.Client;
 
 public static class Client_Subscribe_Samples
 {
-    static readonly MqttTopicTemplate sampleTemplate = new("mqttnet/samples/topic/{id}");
-
     public static async Task Handle_Received_Application_Message()
     {
         /*
@@ -42,7 +40,7 @@ public static class Client_Subscribe_Samples
 
         await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
 
-        var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicTemplate(sampleTemplate.WithParameter("id", "2")).Build();
+        var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicFilter("mqttnet/samples/topic/2").Build();
 
         await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
 
@@ -55,7 +53,7 @@ public static class Client_Subscribe_Samples
     public static async Task Send_Responses()
     {
         /*
-         * This sample subscribes to a topic and sends a response to the broker. This requires at least QoS level 1 to work!
+         * This sample subscribes to a topic and sends a detailed response to the client. This requires at least QoS level 1 to work!
          */
 
         var mqttFactory = new MqttClientFactory();
@@ -70,7 +68,7 @@ public static class Client_Subscribe_Samples
             args.ResponseReasonString = "That did not work!";
 
             // User properties require MQTT v5!
-            args.ResponseUserProperties.Add(new MqttUserProperty("My", "Data"));
+            args.ResponseUserProperties.Add(new MqttUserProperty("My", Encoding.UTF8.GetBytes("Data")));
 
             // Now the broker will resend the message again.
             return Task.CompletedTask;
@@ -80,7 +78,7 @@ public static class Client_Subscribe_Samples
 
         await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
 
-        var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicTemplate(sampleTemplate.WithParameter("id", "1")).Build();
+        var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicFilter("topic/test").Build();
 
         var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
 
@@ -106,9 +104,10 @@ public static class Client_Subscribe_Samples
         // Create the subscribe options including several topics with different options.
         // It is also possible to all of these topics using a dedicated call of _SubscribeAsync_ per topic.
         var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
-            .WithTopicTemplate(sampleTemplate.WithParameter("id", "1"))
+            .WithTopicFilter(t => t.WithTopic("topic/1").WithAtLeastOnceQoS())
-            .WithTopicTemplate(sampleTemplate.WithParameter("id", "2"), noLocal: true)
+            .WithTopicFilter(t => t.WithTopic("topic/2").WithAtMostOnceQoS())
-            .WithTopicTemplate(sampleTemplate.WithParameter("id", "3"), retainHandling: MqttRetainHandling.SendAtSubscribe)
+            .WithTopicFilter(t => t.WithTopic("topic/3").WithNoLocal())
+            .WithTopicFilter(t => t.WithRetainHandling(MqttRetainHandling.SendAtSubscribe))
             .Build();
 
         var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
@@ -132,7 +131,9 @@ public static class Client_Subscribe_Samples
 
         await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
 
-        var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicTemplate(sampleTemplate.WithParameter("id", "1")).Build();
+        var topicFilter = mqttFactory.CreateTopicFilterBuilder().WithTopic("mqttnet/samples/topic/2").WithAtLeastOnceQoS();
+
+        var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicFilter(topicFilter).Build();
 
         var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
 

Samples/Client/Client_Subscribe_TopicTemplate_Samples.cs

@@ -0,0 +1,75 @@
+// 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 MQTTnet.Extensions.TopicTemplate;
+using MQTTnet.Protocol;
+using MQTTnet.Samples.Helpers;
+
+// ReSharper disable UnusedType.Global
+// ReSharper disable UnusedMember.Global
+// ReSharper disable InconsistentNaming
+// ReSharper disable UnusedMember.Local
+
+namespace MQTTnet.Samples.Client;
+
+public static class Client_Subscribe_TopicTemplate_Samples
+{
+    static readonly MqttTopicTemplate SampleTemplate = new("mqttnet/samples/topic/{id}");
+
+    public static async Task Subscribe_Multiple_Topics_With_Template()
+    {
+        /*
+         * This sample subscribes to several topics in a single request.
+         *
+         * This sample requires the nuget 'MQTTnet.Extensions.TopicTemplate'.
+         */
+
+        var mqttFactory = new MqttClientFactory();
+
+        using var mqttClient = mqttFactory.CreateMqttClient();
+        var mqttClientOptions = new MqttClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build();
+
+        await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
+
+        // Create the subscribe options including several topics with different options.
+        // It is also possible to all of these topics using a dedicated call of _SubscribeAsync_ per topic.
+        var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
+            .WithTopicTemplate(SampleTemplate.WithParameter("id", "1"))
+            .WithTopicTemplate(SampleTemplate.WithParameter("id", "2"), noLocal: true)
+            .WithTopicTemplate(SampleTemplate.WithParameter("id", "3"), retainHandling: MqttRetainHandling.SendAtSubscribe)
+            .Build();
+
+        var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
+
+        Console.WriteLine("MQTT client subscribed to topics.");
+
+        // The response contains additional data sent by the server after subscribing.
+        response.DumpToConsole();
+    }
+
+    public static async Task Subscribe_Topic_With_Template()
+    {
+        /*
+         * This sample subscribes to a topic.
+         *
+         * This sample requires the nuget 'MQTTnet.Extensions.TopicTemplate'.
+         */
+
+        var mqttFactory = new MqttClientFactory();
+
+        using var mqttClient = mqttFactory.CreateMqttClient();
+        var mqttClientOptions = new MqttClientOptionsBuilder().WithTcpServer("broker.hivemq.com").Build();
+
+        await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
+
+        var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicTemplate(SampleTemplate.WithParameter("id", "1")).Build();
+
+        var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
+
+        Console.WriteLine("MQTT client subscribed to topic.");
+
+        // The response contains additional data sent by the server after subscribing.
+        response.DumpToConsole();
+    }
+}

Samples/MQTTnet.Samples.csproj

@@ -12,6 +12,7 @@
 
         <ProjectReference Include="..\Source\MQTTnet.AspnetCore\MQTTnet.AspNetCore.csproj"/>
         <ProjectReference Include="..\Source\MQTTnet.Extensions.Rpc\MQTTnet.Extensions.Rpc.csproj"/>
+        <ProjectReference Include="..\Source\MQTTnet.Extensions.Socks5\MQTTnet.Extensions.Socks5.csproj" />
         <ProjectReference Include="..\Source\MQTTnet.Extensions.TopicTemplate\MQTTnet.Extensions.TopicTemplate.csproj" />
         <ProjectReference Include="..\Source\MQTTnet.Server\MQTTnet.Server.csproj"/>
         <ProjectReference Include="..\Source\MQTTnet\MQTTnet.csproj"/>

Samples/Server/Server_TLS_Samples.cs

@@ -59,10 +59,12 @@ public static class Server_TLS_Samples
         certRequest.CertificateExtensions.Add(sanBuilder.Build());
 
         using var certificate = certRequest.CreateSelfSigned(DateTimeOffset.Now.AddMinutes(-10), DateTimeOffset.Now.AddMinutes(10));
+#pragma warning disable SYSLIB0057
         var pfxCertificate = new X509Certificate2(
             certificate.Export(X509ContentType.Pfx),
             (string)null!,
             X509KeyStorageFlags.MachineKeySet | X509KeyStorageFlags.Exportable);
+#pragma warning restore SYSLIB0057
 
         return pfxCertificate;
     }

Source/MQTTnet.AspnetCore/ApplicationBuilderExtensions.cs

@@ -2,7 +2,6 @@
 // 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 Microsoft.AspNetCore.Builder;
 using Microsoft.Extensions.DependencyInjection;
 using MQTTnet.Server;

Source/MQTTnet.AspnetCore/AspNetMqttServerOptionsBuilder.cs

@@ -2,7 +2,6 @@
 // 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.Server;
 
 namespace MQTTnet.AspNetCore;

Source/MQTTnet.AspnetCore/BufferExtensions.cs

@@ -2,7 +2,6 @@
 // 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.Runtime.InteropServices;
 
 namespace MQTTnet.AspNetCore;

Source/MQTTnet.AspnetCore/EndpointRouterExtensions.cs

@@ -2,7 +2,6 @@
 // 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 Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Routing;
 

Source/MQTTnet.AspnetCore/MqttClientConnectionContextFactory.cs

@@ -2,7 +2,6 @@
 // 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.Adapter;
 using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Formatter;

Source/MQTTnet.AspnetCore/MqttConnectionContext.cs

@@ -2,13 +2,10 @@
 // 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.IO.Pipelines;
 using System.Net;
 using System.Security.Cryptography.X509Certificates;
-using System.Threading;
-using System.Threading.Tasks;
 using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Http.Connections.Features;
 using Microsoft.AspNetCore.Http.Features;

Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs

@@ -2,8 +2,6 @@
 // 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.Threading.Tasks;
 using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Connections.Features;
 using MQTTnet.Adapter;

Source/MQTTnet.AspnetCore/MqttHostedServer.cs

@@ -2,10 +2,6 @@
 // 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.Collections.Generic;
-using System.Threading;
-using System.Threading.Tasks;
 using Microsoft.Extensions.Hosting;
 using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Server;

Source/MQTTnet.AspnetCore/MqttSubProtocolSelector.cs

@@ -2,9 +2,6 @@
 // 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.Collections.Generic;
-using System.Linq;
 using Microsoft.AspNetCore.Http;
 
 namespace MQTTnet.AspNetCore;

Source/MQTTnet.AspnetCore/MqttWebSocketServerAdapter.cs

@@ -2,10 +2,8 @@
 // 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.Net;
 using System.Net.WebSockets;
-using System.Threading.Tasks;
 using Microsoft.AspNetCore.Http;
 using MQTTnet.Adapter;
 using MQTTnet.Diagnostics.Logger;

Source/MQTTnet.AspnetCore/ReaderExtensions.cs

@@ -2,7 +2,6 @@
 // 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.Runtime.InteropServices;
 using MQTTnet.Adapter;

Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs

@@ -2,7 +2,6 @@
 // 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 Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection.Extensions;
 using Microsoft.Extensions.Hosting;

Source/MQTTnet.AspnetCore/SocketAwaitable.cs

@@ -2,13 +2,10 @@
 // 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.Diagnostics;
 using System.IO.Pipelines;
 using System.Net.Sockets;
 using System.Runtime.CompilerServices;
-using System.Threading;
-using System.Threading.Tasks;
 
 namespace MQTTnet.AspNetCore;
 

Source/MQTTnet.AspnetCore/SocketConnection.cs

@@ -2,13 +2,9 @@
 // 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.Collections.Generic;
-using System.IO;
 using System.IO.Pipelines;
 using System.Net;
 using System.Net.Sockets;
-using System.Threading.Tasks;
 using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Http.Features;
 using MQTTnet.Exceptions;

Source/MQTTnet.AspnetCore/SocketReceiver.cs

@@ -2,7 +2,6 @@
 // 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.Pipelines;
 using System.Net.Sockets;
 

Source/MQTTnet.AspnetCore/SocketSender.cs

@@ -2,9 +2,7 @@
 // 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.Collections.Generic;
 using System.Diagnostics;
 using System.IO.Pipelines;
 using System.Net.Sockets;

Source/MQTTnet.Benchmarks/AssemblyInfo.cs

@@ -1,3 +1 @@
-using Microsoft.VisualStudio.TestTools.UnitTesting;
-
 [assembly: DoNotParallelize]

Source/MQTTnet.Benchmarks/AsyncLockBenchmark.cs

@@ -2,9 +2,6 @@
 // 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.Threading;
-using System.Threading.Tasks;
 using BenchmarkDotNet.Attributes;
 using BenchmarkDotNet.Jobs;
 using MQTTnet.Internal;

Source/MQTTnet.Benchmarks/ChannelAdapterBenchmark.cs

@@ -2,9 +2,6 @@
 // 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.Threading;
 using BenchmarkDotNet.Attributes;
 using MQTTnet.Adapter;
 using MQTTnet.Diagnostics.Logger;

Source/MQTTnet.Benchmarks/MemoryCopyBenchmark.cs

@@ -1,6 +1,5 @@
 using BenchmarkDotNet.Attributes;
 using BenchmarkDotNet.Jobs;
-using System;
 
 
 namespace MQTTnet.Benchmarks;

Source/MQTTnet.Benchmarks/MqttBufferReaderBenchmark.cs

@@ -2,7 +2,6 @@
 // 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.Text;
 using BenchmarkDotNet.Attributes;
 using BenchmarkDotNet.Jobs;

Source/MQTTnet.Benchmarks/MqttPacketReaderWriterBenchmark.cs

@@ -2,7 +2,6 @@
 // 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 BenchmarkDotNet.Attributes;
 using BenchmarkDotNet.Jobs;
 using MQTTnet.Formatter;

Source/MQTTnet.Benchmarks/Program.cs

@@ -2,9 +2,6 @@
 // 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.Collections.Generic;
-using System.Linq;
 using BenchmarkDotNet.Running;
 
 namespace MQTTnet.Benchmarks;

Source/MQTTnet.Benchmarks/ReaderExtensionsBenchmark.cs

@@ -4,9 +4,7 @@ using MQTTnet.AspNetCore;
 using MQTTnet.Exceptions;
 using MQTTnet.Formatter;
 using System.Buffers;
-using System.IO;
 using System.IO.Pipelines;
-using System.Threading.Tasks;
 
 namespace MQTTnet.Benchmarks;
 

Source/MQTTnet.Benchmarks/RoundtripProcessingBenchmark.cs

@@ -1,7 +1,6 @@
 using BenchmarkDotNet.Attributes;
 using BenchmarkDotNet.Jobs;
 using MQTTnet.Tests.Mockups;
-using MQTTnet.Tests.Server;
 
 namespace MQTTnet.Benchmarks;
 

Source/MQTTnet.Benchmarks/SerializerBenchmark.cs

@@ -4,10 +4,7 @@
 
 using BenchmarkDotNet.Attributes;
 using MQTTnet.Packets;
-using System;
 using System.Security.Cryptography.X509Certificates;
-using System.Threading;
-using System.Threading.Tasks;
 using MQTTnet.Adapter;
 using MQTTnet.Channel;
 using MQTTnet.Formatter;

Source/MQTTnet.Benchmarks/ServerProcessingBenchmark.cs

@@ -5,7 +5,6 @@
 using BenchmarkDotNet.Attributes;
 using BenchmarkDotNet.Jobs;
 using MQTTnet.Tests.Mockups;
-using MQTTnet.Tests.Server;
 
 namespace MQTTnet.Benchmarks;
 

Source/MQTTnet.Benchmarks/SubscribeBenchmark.cs

@@ -2,8 +2,6 @@
 // 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.Collections.Generic;
-using System.Linq;
 using BenchmarkDotNet.Attributes;
 using MQTTnet.Server;
 

Source/MQTTnet.Benchmarks/TcpPipesBenchmark.cs

@@ -5,8 +5,6 @@
 using System.IO.Pipelines;
 using System.Net;
 using System.Net.Sockets;
-using System.Threading;
-using System.Threading.Tasks;
 using BenchmarkDotNet.Attributes;
 using BenchmarkDotNet.Jobs;
 using MQTTnet.AspNetCore;

Source/MQTTnet.Benchmarks/TopicGenerator.cs

@@ -1,6 +1,3 @@
-using System;
-using System.Collections.Generic;
-
 namespace MQTTnet.Benchmarks;
 
 public class TopicGenerator

Source/MQTTnet.Benchmarks/UnsubscribeBenchmark.cs

@@ -2,8 +2,6 @@
 // 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.Collections.Generic;
-using System.Linq;
 using BenchmarkDotNet.Attributes;
 using MQTTnet.Server;
 

Source/MQTTnet.Extensions.Rpc/DefaultMqttRpcClientTopicGenerationStrategy.cs

@@ -2,8 +2,6 @@
 // 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;
-
 namespace MQTTnet.Extensions.Rpc;
 
 public sealed class DefaultMqttRpcClientTopicGenerationStrategy : IMqttRpcClientTopicGenerationStrategy

Source/MQTTnet.Extensions.Rpc/IMqttRpcClient.cs

@@ -2,10 +2,6 @@
 // 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.Collections.Generic;
-using System.Threading;
-using System.Threading.Tasks;
 using MQTTnet.Protocol;
 
 namespace MQTTnet.Extensions.Rpc;

Source/MQTTnet.Extensions.Rpc/MqttFactoryExtensions.cs

@@ -2,8 +2,6 @@
 // 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;
-
 namespace MQTTnet.Extensions.Rpc;
 
 public static class MqttFactoryExtensions

Source/MQTTnet.Extensions.Rpc/MqttRpcClient.cs

@@ -2,12 +2,8 @@
 // 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.Collections.Concurrent;
-using System.Collections.Generic;
-using System.Threading;
-using System.Threading.Tasks;
 using MQTTnet.Exceptions;
 using MQTTnet.Formatter;
 using MQTTnet.Internal;

Source/MQTTnet.Extensions.Rpc/MqttRpcClientExtensions.cs

@@ -2,10 +2,7 @@
 // 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.Collections.Generic;
 using System.Text;
-using System.Threading.Tasks;
 using MQTTnet.Protocol;
 
 namespace MQTTnet.Extensions.Rpc;

Source/MQTTnet.Extensions.Rpc/MqttRpcClientOptionsBuilder.cs

@@ -2,8 +2,6 @@
 // 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;
-
 namespace MQTTnet.Extensions.Rpc;
 
 public sealed class MqttRpcClientOptionsBuilder

Source/MQTTnet.Extensions.Rpc/TopicGenerationContext.cs

@@ -2,8 +2,6 @@
 // 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.Collections.Generic;
 using MQTTnet.Protocol;
 
 namespace MQTTnet.Extensions.Rpc;

Source/MQTTnet.Extensions.Socks5/MQTTnet.Extensions.Socks5.csproj

@@ -0,0 +1,21 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <AssemblyName>MQTTnet.Extensions.Socks5</AssemblyName>
+        <RootNamespace>MQTTnet.Extensions.Socks5</RootNamespace>
+        <IsAotCompatible>true</IsAotCompatible>
+
+        <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
+        <Description>This is an extension library which adds SOCKS5 proxy support (RFC 1928 / RFC 1929) to the MQTTnet TCP transport.</Description>
+        <PackageId>MQTTnet.Extensions.Socks5</PackageId>
+        <PackageTags>MQTT Message Queue Telemetry Transport MQTTClient SOCKS SOCKS5 Proxy Tunnel Networking IoT InternetOfThings</PackageTags>
+        <EmbedUntrackedSources>true</EmbedUntrackedSources>
+        <PackageReleaseNotes>For release notes please go to MQTTnet release notes (https://www.nuget.org/packages/MQTTnet/).</PackageReleaseNotes>
+        <GenerateDocumentationFile>true</GenerateDocumentationFile>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <ProjectReference Include="..\MQTTnet\MQTTnet.csproj"/>
+    </ItemGroup>
+
+</Project>

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;
+
+/// <summary>
+///     Sugar extensions for configuring a SOCKS5 proxy on the MQTT client options builder.
+/// </summary>
+public static class MqttClientOptionsBuilderExtensions
+{
+    /// <summary>
+    ///     Configures the client to tunnel its TCP connection to the broker through a SOCKS5
+    ///     proxy. A TCP transport (via <c>WithTcpServer(...)</c> or <c>WithEndPoint(...)</c>)
+    ///     must be configured first.
+    /// </summary>
+    public static MqttClientOptionsBuilder WithSocks5Proxy(this MqttClientOptionsBuilder builder, Socks5ProxyOptions options)
+    {
+        ArgumentNullException.ThrowIfNull(builder);
+        ArgumentNullException.ThrowIfNull(options);
+
+        return builder.WithStreamProvider(new Socks5StreamProvider(options));
+    }
+
+    /// <summary>
+    ///     Configures the client to tunnel its TCP connection to the broker through a SOCKS5
+    ///     proxy using the fluent <see cref="Socks5ProxyOptionsBuilder"/>.
+    /// </summary>
+    public static MqttClientOptionsBuilder WithSocks5Proxy(this MqttClientOptionsBuilder builder, Action<Socks5ProxyOptionsBuilder> configure)
+    {
+        ArgumentNullException.ThrowIfNull(builder);
+        ArgumentNullException.ThrowIfNull(configure);
+
+        var optionsBuilder = new Socks5ProxyOptionsBuilder();
+        configure(optionsBuilder);
+
+        return WithSocks5Proxy(builder, optionsBuilder.Build());
+    }
+}

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;
+
+/// <summary>
+///     Wraps a <see cref="NetworkStream"/> opened by the SOCKS5 stream provider so
+///     <c>MqttTcpChannel</c> can read the original socket's endpoints via
+///     <see cref="IMqttClientStreamEndPoints"/> while the rest of the I/O delegates to the
+///     underlying stream.
+/// </summary>
+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<byte> buffer) => _inner.Read(buffer);
+
+    public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        => _inner.ReadAsync(buffer, offset, count, cancellationToken);
+
+    public override ValueTask<int> ReadAsync(Memory<byte> 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<byte> 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<byte> 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);
+    }
+}

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;
+
+/// <summary>
+///     Protocol constants for SOCKS5 (RFC 1928) and the username/password sub-negotiation
+///     (RFC 1929). Internal use only.
+/// </summary>
+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;
+}

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;
+
+/// <summary>
+///     Options for the SOCKS5 stream provider (RFC 1928 / RFC 1929).
+///     A new provider can be constructed from these options via
+///     <see cref="MqttClientOptionsBuilderExtensions.WithSocks5Proxy(MqttClientOptionsBuilder, Socks5ProxyOptions)"/>.
+/// </summary>
+public sealed class Socks5ProxyOptions
+{
+    /// <summary>
+    ///     The SOCKS5 proxy host name or IP literal. Must be set.
+    /// </summary>
+    public string Host { get; set; }
+
+    /// <summary>
+    ///     The SOCKS5 proxy port. Defaults to 1080.
+    /// </summary>
+    public int Port { get; set; } = 1080;
+
+    /// <summary>
+    ///     The user name used for the username/password authentication sub-negotiation (RFC 1929).
+    ///     When <c>null</c> or empty, the client will only offer the "no authentication" method.
+    /// </summary>
+    public string Username { get; set; }
+
+    /// <summary>
+    ///     The password (raw bytes per RFC 1929) used for the username/password authentication
+    ///     sub-negotiation. Must be 1..255 bytes when <see cref="Username"/> is set.
+    /// </summary>
+    public byte[] Password { get; set; }
+
+    /// <summary>
+    ///     When <c>true</c> (the default) the broker hostname is forwarded to the proxy as a
+    ///     SOCKS5 <c>DOMAINNAME</c> address type, so DNS resolution happens at the proxy
+    ///     (typical when the broker is behind the proxy in a private network).
+    ///     When <c>false</c>, the hostname is resolved locally before sending the CONNECT request
+    ///     and an IPv4 / IPv6 address type is used.
+    /// </summary>
+    public bool ResolveDnsRemotely { get; set; } = true;
+
+    /// <summary>
+    ///     The maximum time allowed for the full SOCKS5 handshake (TCP connect to proxy + greeting
+    ///     + optional auth + CONNECT). The handshake will be cancelled and a
+    ///     <see cref="MQTTnet.Exceptions.MqttProxyException"/> will be thrown when this elapses.
+    ///     Defaults to 30 seconds.
+    /// </summary>
+    public TimeSpan HandshakeTimeout { get; set; } = TimeSpan.FromSeconds(30);
+
+    /// <summary>
+    ///     Address family preference for the socket connecting to the proxy and for local DNS
+    ///     resolution when <see cref="ResolveDnsRemotely"/> is <c>false</c>.
+    /// </summary>
+    public AddressFamily AddressFamily { get; set; } = AddressFamily.Unspecified;
+
+    public override string ToString()
+    {
+        return $"{Host}:{Port}";
+    }
+}

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;
+
+/// <summary>
+///     Fluent builder for <see cref="Socks5ProxyOptions"/>.
+/// </summary>
+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;
+    }
+
+    /// <summary>
+    ///     Controls whether the broker hostname is forwarded to the proxy (remote DNS, default) or
+    ///     resolved locally before issuing the CONNECT request.
+    /// </summary>
+    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;
+    }
+}

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;
+
+/// <summary>
+///     <see cref="IMqttClientStreamProvider"/> implementation that tunnels the connection through
+///     a SOCKS5 proxy (RFC 1928) with optional username/password authentication (RFC 1929).
+/// </summary>
+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<Stream> 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);
+    }
+}

Source/MQTTnet.Extensions.TopicTemplate/MqttTopicTemplate.cs

@@ -2,11 +2,7 @@
 // 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.Collections.Generic;
-using System.Linq;
 using System.Text.RegularExpressions;
-using System.Threading;
 using MQTTnet.Protocol;
 
 namespace MQTTnet.Extensions.TopicTemplate;

Source/MQTTnet.Extensions.TopicTemplate/TopicTemplateExtensions.cs

@@ -2,8 +2,6 @@
 // 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.Linq;
 using MQTTnet.Packets;
 using MQTTnet.Protocol;
 

Source/MQTTnet.PowerShell/Cmdlets/ConnectMqttSessionCmdlet.cs

@@ -1,47 +0,0 @@
-using System.Management.Automation;
-
-namespace MQTTnet.PowerShell.Cmdlets;
-
-[Cmdlet(VerbsCommunications.Connect, "MqttSession")]
-[OutputType(typeof(MqttClientConnectResult))]
-public class ConnectMqttSessionCmdlet : PSCmdlet
-{
-    [Parameter]
-    public SwitchParameter CleanSession { get; set; } = true;
-
-    [Parameter]
-    public string? ClientId { get; set; } = Guid.NewGuid().ToString();
-
-    [Parameter(Mandatory = true)]
-    public required new string Host { get; set; }
-
-    [Parameter]
-    public string? Password { get; set; }
-
-    [Parameter]
-    public int Port { get; set; } = 1883;
-
-    [Parameter(Mandatory = true, ValueFromPipeline = true)]
-    public required PsMqttSession Session { get; set; }
-
-    [Parameter]
-    public string? Username { get; set; }
-
-    [Parameter]
-    public SwitchParameter UseTls { get; set; }
-
-    protected override void ProcessRecord()
-    {
-        var clientOptionsBuilder = new MqttClientOptionsBuilder();
-        clientOptionsBuilder.WithTcpServer(Host, Port);
-
-        if (Username != null)
-        {
-            clientOptionsBuilder.WithCredentials(Username, Password);
-        }
-
-        var response = Session.GetClient().ConnectAsync(clientOptionsBuilder.Build()).GetAwaiter().GetResult();
-
-        WriteObject(response);
-    }
-}

Source/MQTTnet.PowerShell/Cmdlets/DisconnectMqttSessionCmdlet.cs

@@ -1,31 +0,0 @@
-using System.Management.Automation;
-
-namespace MQTTnet.PowerShell.Cmdlets;
-
-[Cmdlet(VerbsCommunications.Disconnect, "MqttSession")]
-public class DisconnectMqttSessionCmdlet : PSCmdlet
-{
-    [Parameter]
-    public MqttClientDisconnectOptionsReason Reason { get; set; } = MqttClientDisconnectOptionsReason.NormalDisconnection;
-
-    [Parameter]
-    public string? ReasonString { get; set; }
-
-    [Parameter(Mandatory = true, ValueFromPipeline = true)]
-    public required PsMqttSession Session { get; set; }
-
-    [Parameter]
-    public uint SessionExpiryInterval { get; set; }
-
-    protected override void ProcessRecord()
-    {
-        if (Session.GetClient().IsConnected)
-        {
-            var options = new MqttClientDisconnectOptionsBuilder().WithSessionExpiryInterval(SessionExpiryInterval).WithReason(Reason).WithReasonString(ReasonString).Build();
-
-            Session.GetClient().DisconnectAsync(options).GetAwaiter().GetResult();
-        }
-
-        WriteObject("Disconnected.");
-    }
-}

Source/MQTTnet.PowerShell/Cmdlets/NewMqttSessionCmdlet.cs

@@ -1,13 +0,0 @@
-using System.Management.Automation;
-
-namespace MQTTnet.PowerShell.Cmdlets;
-
-[Cmdlet(VerbsCommon.New, "MqttSession")]
-[OutputType(typeof(PsMqttSession))]
-public class NewMqttSessionCmdlet : PSCmdlet
-{
-    protected override void ProcessRecord()
-    {
-        WriteObject(new PsMqttSession());
-    }
-}

Source/MQTTnet.PowerShell/Cmdlets/PublishMqttMessageCmdlet.cs

@@ -1,52 +0,0 @@
-using System.Management.Automation;
-using System.Text;
-using MQTTnet.Protocol;
-
-namespace MQTTnet.PowerShell.Cmdlets;
-
-[Cmdlet(VerbsData.Publish, "MqttMessage")]
-public class PublishMqttMessageCmdlet : PSCmdlet
-{
-    [Parameter(Mandatory = true)]
-    public string? Payload { get; set; }
-
-    [Parameter]
-    public int QoS { get; set; } = 0;
-
-    [Parameter]
-    public SwitchParameter Retain { get; set; }
-
-    [Parameter(Mandatory = true, ValueFromPipeline = true)]
-    public required PsMqttSession Session { get; set; }
-
-    [Parameter(Mandatory = true)]
-    public required string Topic { get; set; }
-
-    public string? ContentType { get; set; }
-
-    public string? ResponseTopic { get; set; }
-
-    public ushort TopicAlias { get; set; }
-
-    public uint MessageExpiryInterval { get; set; }
-
-    protected override void ProcessRecord()
-    {
-        // if (Session == null || !Session.IsConnected)
-        //     throw new InvalidOperationException("Session not connected.");
-
-        var msg = new MqttApplicationMessageBuilder().WithTopic(Topic)
-            .WithPayload(Encoding.UTF8.GetBytes(Payload ?? string.Empty))
-            .WithQualityOfServiceLevel((MqttQualityOfServiceLevel)QoS)
-            .WithRetainFlag(Retain)
-            .WithContentType(ContentType)
-            .WithResponseTopic(ResponseTopic)
-            .WithTopicAlias(TopicAlias)
-            .WithMessageExpiryInterval(MessageExpiryInterval)
-            .Build();
-
-        var response = Session.GetClient().PublishAsync(msg).GetAwaiter().GetResult();
-
-        WriteObject(response);
-    }
-}

Source/MQTTnet.PowerShell/Cmdlets/ReceiveMqttMessageCmdlet.cs

@@ -1,74 +0,0 @@
-// 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.Management.Automation;
-
-namespace MQTTnet.PowerShell.Cmdlets;
-
-[Cmdlet(VerbsCommunications.Receive, "MqttMessage")]
-[OutputType(typeof(PsMqttMessage))]
-public class ReceiveMqttMessageCmdlet : PSCmdlet
-{
-    [Parameter(Mandatory = true, ValueFromPipeline = true)]
-    public required PsMqttSession Session { get; set; }
-
-    [Parameter]
-    [ValidateRange(0, int.MaxValue)]
-    public int TimeoutSeconds { get; set; } = 0;
-
-    protected override void ProcessRecord()
-    {
-        var tcs = new TaskCompletionSource<PsMqttMessage>();
-        EventHandler<PsMqttMessage>? handler = null;
-        CancellationTokenSource? cts = null;
-
-        try
-        {
-            handler = (s, e) =>
-            {
-                if (!tcs.Task.IsCompleted)
-                {
-                    tcs.TrySetResult(e);
-                }
-            };
-
-            Session.MessageReceived += handler;
-
-            // Optionaler Timeout
-            if (TimeoutSeconds > 0)
-            {
-                cts = new CancellationTokenSource();
-                cts.CancelAfter(TimeSpan.FromSeconds(TimeoutSeconds));
-                cts.Token.Register(() =>
-                {
-                    if (!tcs.Task.IsCompleted)
-                    {
-                        tcs.TrySetCanceled();
-                    }
-                });
-            }
-
-            WriteVerbose("Waiting for next MQTT message...");
-
-            try
-            {
-                var message = tcs.Task.GetAwaiter().GetResult();
-                WriteObject(message);
-            }
-            catch (TaskCanceledException)
-            {
-                WriteWarning($"Timeout after {TimeoutSeconds} seconds — no message received.");
-            }
-        }
-        finally
-        {
-            if (handler != null)
-            {
-                Session.MessageReceived -= handler;
-            }
-
-            cts?.Dispose();
-        }
-    }
-}

Source/MQTTnet.PowerShell/Cmdlets/RegisterMqttMessageHandlerCmdlet.cs

@@ -1,26 +0,0 @@
-using System.Management.Automation;
-using System.Management.Automation.Runspaces;
-
-namespace MQTTnet.PowerShell.Cmdlets;
-
-[Cmdlet(VerbsLifecycle.Register, "MqttMessageHandler")]
-public class RegisterMqttMessageHandlerCmdlet : PSCmdlet
-{
-    [Parameter(Mandatory = true, ValueFromPipeline = true)]
-    public required PsMqttSession Session { get; set; }
-
-    [Parameter(Mandatory = true)]
-    public ScriptBlock? Action { get; set; }
-
-    protected override void ProcessRecord()
-    {
-        EventHandler<PsMqttMessage> handler = (s, e) =>
-        {
-            //throw new NotImplementedException();
-            //InvokeCommand.InvokeScript(Action, false, PipelineResultTypes.Output, null, new object[] { e.Topic, e.Payload });
-        };
-
-        Session.MessageReceived += handler;
-        WriteObject($"Handler registered for MQTT messages on session.");
-    }
-}

Source/MQTTnet.PowerShell/Cmdlets/RemoveMqttSessionCmdlet.cs

@@ -1,16 +0,0 @@
-using System.Management.Automation;
-
-namespace MQTTnet.PowerShell.Cmdlets;
-
-[Cmdlet(VerbsCommon.Remove, "MqttSession")]
-public class RemoveMqttSessionCmdlet : PSCmdlet
-{
-    [Parameter(Mandatory = true, ValueFromPipeline = true)]
-    public required PsMqttSession Session { get; set; }
-
-    protected override void ProcessRecord()
-    {
-        Session.Dispose();
-        WriteObject("Session removed.");
-    }
-}

Source/MQTTnet.PowerShell/Cmdlets/SubscribeMqttTopicCmdlet.cs

@@ -1,55 +0,0 @@
-// 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.Management.Automation;
-using MQTTnet.Protocol;
-
-namespace MQTTnet.PowerShell.Cmdlets;
-
-[Cmdlet("Subscribe", "MqttTopic")]
-[OutputType(typeof(MqttClientSubscribeResult))]
-public class SubscribeMqttTopicCmdlet : PSCmdlet
-{
-    [Parameter]
-    public bool NoLocal { get; set; }
-
-    [Parameter]
-    public int QoS { get; set; } = 0;
-
-    [Parameter]
-    public bool RetainAsPublished { get; set; }
-
-    [Parameter]
-    public MqttRetainHandling RetainHandling { get; set; } = MqttRetainHandling.SendAtSubscribe;
-
-    [Parameter(Mandatory = true, ValueFromPipeline = true)]
-    public required PsMqttSession Session { get; set; }
-
-    [Parameter]
-    public uint SubscriptionIdentifier { get; set; }
-
-    [Parameter(Mandatory = true)]
-    public required string Topic { get; set; }
-
-    protected override void ProcessRecord()
-    {
-        var topicFilter = new MqttTopicFilterBuilder().WithTopic(Topic)
-            .WithQualityOfServiceLevel((MqttQualityOfServiceLevel)QoS)
-            .WithNoLocal(NoLocal)
-            .WithRetainAsPublished(RetainAsPublished)
-            .WithRetainHandling(RetainHandling)
-            .Build();
-
-        var options = new MqttClientSubscribeOptionsBuilder().WithTopicFilter(topicFilter);
-
-        if (SubscriptionIdentifier > 0)
-        {
-            options.WithSubscriptionIdentifier(SubscriptionIdentifier);
-        }
-
-        var response = Session.GetClient().SubscribeAsync(options.Build()).GetAwaiter().GetResult();
-
-        WriteObject(response);
-    }
-}

Source/MQTTnet.PowerShell/Cmdlets/UnsubscribeMqttTopicCmdlet.cs

@@ -1,23 +0,0 @@
-using System.Management.Automation;
-
-namespace MQTTnet.PowerShell.Cmdlets;
-
-[Cmdlet("Unsubscribe", "MqttTopic")]
-[OutputType(typeof(MqttClientUnsubscribeResult))]
-public class UnsubscribeMqttTopicCmdlet : PSCmdlet
-{
-    [Parameter(Mandatory = true, ValueFromPipeline = true)]
-    public required PsMqttSession Session { get; set; }
-
-    [Parameter(Mandatory = true)]
-    public required string Topic { get; set; }
-
-    protected override void ProcessRecord()
-    {
-        var options = new MqttClientUnsubscribeOptionsBuilder().WithTopicFilter(Topic).Build();
-
-        var response = Session.GetClient().UnsubscribeAsync(options).GetAwaiter().GetResult();
-
-        WriteObject(response);
-    }
-}

Source/MQTTnet.PowerShell/MQTTnet.PowerShell.csproj

@@ -1,28 +0,0 @@
-<Project Sdk="Microsoft.NET.Sdk">
-
-  <PropertyGroup>
-    <AssemblyName>MQTTnet.PowerShell</AssemblyName>
-    <LangVersion>default</LangVersion>
-    <Nullable>enable</Nullable>
-  </PropertyGroup>
-
-  <ItemGroup>
-    <PackageReference Include="PowerShellStandard.Library" Version="5.1.1">
-      <PrivateAssets>All</PrivateAssets>
-    </PackageReference>
-  </ItemGroup>
-
-  <ItemGroup>
-    <ProjectReference Include="..\MQTTnet\MQTTnet.csproj" />
-  </ItemGroup>
-
-  <ItemGroup>
-    <Content Include="bin\Debug\net8.0\MQTTnet.dll" />
-    <Content Include="bin\Debug\net8.0\MQTTnet.pdb" />
-    <Content Include="bin\Debug\net8.0\MQTTnet.PowerShell.deps.json" />
-    <Content Include="bin\Debug\net8.0\MQTTnet.PowerShell.dll" />
-    <Content Include="bin\Debug\net8.0\MQTTnet.PowerShell.pdb" />
-    <Content Include="bin\Debug\net8.0\MQTTnet.xml" />
-  </ItemGroup>
-
-</Project>

Source/MQTTnet.PowerShell/PsMqttMessage.cs

@@ -1,30 +0,0 @@
-// 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.Management.Automation;
-using MQTTnet.Packets;
-
-namespace MQTTnet.PowerShell;
-
-public class PsMqttPublishResult
-{
-
-}
-
-public class PsMqttConnectResult
-{
-
-}
-
-public class PsMqttMessage
-{
-    public required string? ContentType { get; init; }
-    public required string Payload { get; init; }
-    public int QoS { get; init; }
-    public required byte[] RawPayload { get; init; }
-    public string? ResponseTopic { get; init; }
-    public bool Retain { get; init; }
-    public required string Topic { get; init; }
-    public required List<MqttUserProperty>? UserProperties { get; init; }
-}

Source/MQTTnet.PowerShell/PsMqttSession.cs

@@ -1,50 +0,0 @@
-// 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;
-using System.Text;
-
-namespace MQTTnet.PowerShell;
-
-public sealed class PsMqttSession : IDisposable
-{
-    readonly IMqttClient _client;
-
-    public PsMqttSession()
-    {
-        var factory = new MqttClientFactory();
-        _client = factory.CreateMqttClient();
-
-        _client.ApplicationMessageReceivedAsync += e =>
-        {
-            MessageReceived?.Invoke(
-                this,
-                new PsMqttMessage
-                {
-                    Topic = e.ApplicationMessage.Topic,
-                    Payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload),
-                    RawPayload = e.ApplicationMessage.Payload.ToArray(),
-                    UserProperties = e.ApplicationMessage.UserProperties,
-                    QoS = (int)e.ApplicationMessage.QualityOfServiceLevel,
-                    ContentType = e.ApplicationMessage.ContentType,
-                    Retain = e.ApplicationMessage.Retain,
-                    ResponseTopic = e.ApplicationMessage.ResponseTopic
-                });
-
-            return Task.CompletedTask;
-        };
-    }
-
-    public event EventHandler<PsMqttMessage>? MessageReceived;
-
-    public void Dispose()
-    {
-        _client.Dispose();
-    }
-
-    public IMqttClient GetClient()
-    {
-        return _client;
-    }
-}

Source/MQTTnet.PowerShell/Test.ps1

@@ -1,15 +0,0 @@
-Import-Module ./bin/Debug/net8.0/MQTTnet.PowerShell.dll
-
-$session = New-MqttSession
-
-Connect-MqttSession -Session $session -Host "192.168.1.16"
-
-Publish-MqttMessage -Session $session  -Topic "Test" -Payload "Hello from PowerShell"
-
-Subscribe-MqttTopic -Session $session -Topic "test/demo"
-
-Receive-MqttMessage -Session $session -TimeoutSeconds 200
-
-Disconnect-MqttSession -Session $session
-Remove-MqttSession -Session $session
-

Source/MQTTnet.PowerShell/bin/Debug/net8.0/MQTTnet.PowerShell.deps.json

@@ -1,55 +0,0 @@
-{
-  "runtimeTarget": {
-    "name": ".NETCoreApp,Version=v8.0",
-    "signature": ""
-  },
-  "compilationOptions": {},
-  "targets": {
-    ".NETCoreApp,Version=v8.0": {
-      "MQTTnet.PowerShell/1.0.0": {
-        "dependencies": {
-          "MQTTnet": "1.0.0",
-          "PowerShellStandard.Library": "5.1.1"
-        },
-        "runtime": {
-          "MQTTnet.PowerShell.dll": {}
-        }
-      },
-      "PowerShellStandard.Library/5.1.1": {
-        "runtime": {
-          "lib/netstandard2.0/System.Management.Automation.dll": {
-            "assemblyVersion": "3.0.0.0",
-            "fileVersion": "5.1.1.0"
-          }
-        }
-      },
-      "MQTTnet/1.0.0": {
-        "runtime": {
-          "MQTTnet.dll": {
-            "assemblyVersion": "1.0.0.0",
-            "fileVersion": "1.0.0.0"
-          }
-        }
-      }
-    }
-  },
-  "libraries": {
-    "MQTTnet.PowerShell/1.0.0": {
-      "type": "project",
-      "serviceable": false,
-      "sha512": ""
-    },
-    "PowerShellStandard.Library/5.1.1": {
-      "type": "package",
-      "serviceable": true,
-      "sha512": "sha512-e31xJjG+Kjbv6YF3Yq6D4Dl3or8v7LrNF41k3CXrWozW6hR1zcOe5KYuZJaGSiAgLnwP8wcW+I3+IWEzMPZKXQ==",
-      "path": "powershellstandard.library/5.1.1",
-      "hashPath": "powershellstandard.library.5.1.1.nupkg.sha512"
-    },
-    "MQTTnet/1.0.0": {
-      "type": "project",
-      "serviceable": false,
-      "sha512": ""
-    }
-  }
-}

Source/MQTTnet.Server/Disconnecting/MqttServerClientDisconnectOptionsBuilder.cs

@@ -40,6 +40,7 @@ public sealed class MqttServerClientDisconnectOptionsBuilder
         return this;
     }
 
+    [Obsolete("Please use more performance `WithUserProperty` with ArraySegment<byte> or ReadOnlyMemory<byte> for the value.")]
     public MqttServerClientDisconnectOptionsBuilder WithUserProperty(string name, string value)
     {
         if (_options.UserProperties == null)
@@ -50,4 +51,26 @@ public sealed class MqttServerClientDisconnectOptionsBuilder
         _options.UserProperties.Add(new MqttUserProperty(name, value));
         return this;
     }
+
+    public MqttServerClientDisconnectOptionsBuilder WithUserProperty(string name, ReadOnlyMemory<byte> value)
+    {
+        if (_options.UserProperties == null)
+        {
+            _options.UserProperties = new List<MqttUserProperty>();
+        }
+
+        _options.UserProperties.Add(new MqttUserProperty(name, value));
+        return this;
+    }
+
+    public MqttServerClientDisconnectOptionsBuilder WithUserProperty(string name, ArraySegment<byte> value)
+    {
+        if (_options.UserProperties == null)
+        {
+            _options.UserProperties = new List<MqttUserProperty>();
+        }
+
+        _options.UserProperties.Add(new MqttUserProperty(name, value));
+        return this;
+    }
 }

Source/MQTTnet.Server/EnhancedAuthentication/ExchangeEnhancedAuthenticationOptionsFactory.cs

@@ -51,6 +51,7 @@ public sealed class ExchangeEnhancedAuthenticationOptionsFactory
         return this;
     }
 
+    [Obsolete("Please use more performance `WithUserProperty` with ArraySegment<byte> or ReadOnlyMemory<byte> for the value.")]
     public ExchangeEnhancedAuthenticationOptionsFactory WithUserProperty(string name, string value)
     {
         ArgumentNullException.ThrowIfNull(name);
@@ -64,4 +65,32 @@ public sealed class ExchangeEnhancedAuthenticationOptionsFactory
 
         return this;
     }
+
+    public ExchangeEnhancedAuthenticationOptionsFactory WithUserProperty(string name, ReadOnlyMemory<byte> value)
+    {
+        ArgumentNullException.ThrowIfNull(name);
+
+        if (_options.UserProperties == null)
+        {
+            _options.UserProperties = new List<MqttUserProperty>();
+        }
+
+        _options.UserProperties.Add(new MqttUserProperty(name, value));
+
+        return this;
+    }
+
+    public ExchangeEnhancedAuthenticationOptionsFactory WithUserProperty(string name, ArraySegment<byte> value)
+    {
+        ArgumentNullException.ThrowIfNull(name);
+
+        if (_options.UserProperties == null)
+        {
+            _options.UserProperties = new List<MqttUserProperty>();
+        }
+
+        _options.UserProperties.Add(new MqttUserProperty(name, value));
+
+        return this;
+    }
 }

Source/MQTTnet.Server/Internal/Adapter/MqttTcpServerListener.cs

@@ -202,9 +202,14 @@ public sealed class MqttTcpServerListener : IDisposable
 
                 clientCertificate = sslStream.RemoteCertificate as X509Certificate2;
 
+                // TODO: Check why this export is needed. Is there something else in the RemoteCertificate as a X509Certificate2???
                 if (clientCertificate == null && sslStream.RemoteCertificate != null)
                 {
+                    #if NET10_0_OR_GREATER
+                    clientCertificate = X509CertificateLoader.LoadCertificate(sslStream.RemoteCertificate.Export(X509ContentType.Cert));
+                    #else
                     clientCertificate = new X509Certificate2(sslStream.RemoteCertificate.Export(X509ContentType.Cert));
+                    #endif
                 }
             }
 

Source/MQTTnet.Server/Internal/MqttConnectedClient.cs

@@ -218,6 +218,8 @@ public sealed class MqttConnectedClient : IDisposable
         HandleTopicAlias(publishPacket);
 
         var applicationMessage = MqttApplicationMessageFactory.Create(publishPacket);
+        // Topic aliases are scoped to this network connection and must not be forwarded to other clients.
+        applicationMessage.TopicAlias = 0;
 
         var dispatchApplicationMessageResult =
             await _sessionsManager.DispatchApplicationMessage(Id, UserName, Session.Items, applicationMessage, cancellationToken).ConfigureAwait(false);

Source/MQTTnet.TestApp/AsyncLockTest.cs

@@ -2,9 +2,6 @@
 // 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.Threading;
-using System.Threading.Tasks;
 using MQTTnet.Exceptions;
 
 namespace MQTTnet.TestApp;

Source/MQTTnet.TestApp/ClientFlowTest.cs

@@ -2,8 +2,6 @@
 // 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.Threading.Tasks;
 using MQTTnet.Diagnostics.Logger;
 
 namespace MQTTnet.TestApp;

Source/MQTTnet.TestApp/ClientTest.cs

@@ -2,10 +2,8 @@
 // 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.Net;
 using System.Text;
-using System.Threading.Tasks;
 using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Internal;
 using MQTTnet.Protocol;

Source/MQTTnet.TestApp/MQTTnet.TestApp.csproj

@@ -2,7 +2,6 @@
 
     <PropertyGroup>
         <OutputType>Exe</OutputType>
-        <DebugType>Full</DebugType>
         <IsPackable>false</IsPackable>
     </PropertyGroup>
 

Source/MQTTnet.TestApp/MessageThroughputTest.cs

@@ -1,10 +1,5 @@
 using MQTTnet.Server;
-using System.Collections.Generic;
-using System;
 using System.Diagnostics;
-using System.Threading;
-using System.Threading.Tasks;
-using System.Linq;
 using MQTTnet.Internal;
 
 namespace MQTTnet.TestApp;

Source/MQTTnet.TestApp/MqttNetConsoleLogger.cs

@@ -2,7 +2,6 @@
 // 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.Text;
 using MQTTnet.Diagnostics.Logger;
 

Source/MQTTnet.TestApp/PerformanceTest.cs

@@ -2,13 +2,9 @@
 // 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.Diagnostics;
-using System.Linq;
 using System.Net;
 using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
 using MQTTnet.Protocol;
 using MQTTnet.Server;
 

Source/MQTTnet.TestApp/Program.cs

@@ -2,10 +2,6 @@
 // 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.Threading;
-using System.Threading.Tasks;
-
 namespace MQTTnet.TestApp;
 
 public static class Program

Source/MQTTnet.TestApp/PublicBrokerTest.cs

@@ -2,10 +2,7 @@
 // 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.Security.Authentication;
-using System.Threading;
-using System.Threading.Tasks;
 using MQTTnet.Formatter;
 using MQTTnet.Internal;
 using MQTTnet.Protocol;

Source/MQTTnet.TestApp/ServerAndClientTest.cs

@@ -2,8 +2,6 @@
 // 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.Threading;
-using System.Threading.Tasks;
 using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Server;
 

Source/MQTTnet.TestApp/ServerTest.cs

@@ -2,11 +2,7 @@
 // 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.Collections.Generic;
-using System.IO;
 using System.Text;
-using System.Threading.Tasks;
 using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Internal;
 using MQTTnet.Protocol;

Source/MQTTnet.TestApp/TopicGenerator.cs

@@ -1,5 +1,3 @@
-using System;
-using System.Collections.Generic;
 using MQTTnet.Extensions.TopicTemplate;
 
 namespace MQTTnet.TestApp;

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]

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;
+
+/// <summary>
+///     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.
+/// </summary>
+internal sealed class FakeSocks5Server : IAsyncDisposable
+{
+    readonly TcpListener _listener;
+    readonly CancellationTokenSource _cts = new();
+    readonly Task _acceptLoop;
+    readonly List<Task> _connectionTasks = new();
+    readonly object _connectionsLock = new();
+    readonly List<RecordedConnect> _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<RecordedConnect> 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<bool> 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
+{
+    /// <summary>
+    ///     The SOCKS5 auth methods the server will accept, in preference order matched against the
+    ///     client's offered methods. Defaults to no-auth only.
+    /// </summary>
+    public byte[] AllowedMethods { get; set; } = [0x00];
+
+    public string ExpectedUsername { get; set; }
+
+    public string ExpectedPassword { get; set; }
+
+    /// <summary>
+    ///     Forces the username/password sub-negotiation to return STATUS=0x01 regardless of the
+    ///     supplied credentials. Useful for tests that exercise the auth-failure path.
+    /// </summary>
+    public bool ForceAuthFailure { get; set; }
+
+    /// <summary>
+    ///     SOCKS5 reply code for the CONNECT response. <c>0x00</c> means "succeeded" and the
+    ///     server will additionally bridge bytes to the destination.
+    /// </summary>
+    public byte ConnectReplyCode { get; set; }
+
+    /// <summary>
+    ///     Reserved byte included in the CONNECT reply. Defaults to <c>0x00</c> per RFC 1928.
+    /// </summary>
+    public byte ReplyReservedByte { get; set; }
+
+    /// <summary>
+    ///     When set, the server sends this byte sequence instead of the regular greeting response.
+    ///     Used to simulate malformed responses (e.g. wrong version).
+    /// </summary>
+    public byte[] GreetingResponseOverride { get; set; }
+
+    /// <summary>
+    ///     If <c>true</c>, closes the connection right after writing the greeting response
+    ///     override. Used together with <see cref="GreetingResponseOverride"/> for protocol-error
+    ///     tests.
+    /// </summary>
+    public bool DropAfterGreetingResponse { get; set; }
+
+    /// <summary>
+    ///     If <c>true</c>, the server closes the connection immediately after accepting (before
+    ///     reading the greeting). Used to simulate proxy mid-handshake disconnects.
+    /// </summary>
+    public bool DropAfterAccept { get; set; }
+
+    /// <summary>
+    ///     Address type used when constructing CONNECT replies. Defaults to <c>0x01</c> (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
+    ///     <c>0x03</c> or <c>0x04</c> to exercise the client's reply parser.
+    /// </summary>
+    public byte ReplyAddressType { get; set; } = 0x01;
+
+    /// <summary>
+    ///     Bound host included in the CONNECT reply (BND.ADDR). Format depends on
+    ///     <see cref="ReplyAddressType"/>: IPv4/IPv6 textual representation for ATYP 0x01/0x04,
+    ///     a domain name for ATYP 0x03. When <c>null</c> or empty the server writes a zeroed
+    ///     address (0.0.0.0, ::, or "0.0.0.0" respectively).
+    /// </summary>
+    public string ReplyBindHost { get; set; }
+
+    /// <summary>
+    ///     Bound port included in the CONNECT reply (BND.PORT). Defaults to <c>0</c>, which
+    ///     is what real SOCKS5 servers return when no meaningful port is available.
+    /// </summary>
+    public ushort ReplyBindPort { get; set; }
+}

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;

Source/MQTTnet.Tests.Extensions.Socks5/MQTTnet.Tests.Extensions.Socks5.csproj

@@ -0,0 +1,28 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <OutputType>Exe</OutputType>
+        <IsPackable>false</IsPackable>
+        <LangVersion>default</LangVersion>
+        <Nullable>disable</Nullable>
+        <NoWarn>$(NoWarn);CA1707;CA1859</NoWarn>
+        <EnableMSTestRunner>true</EnableMSTestRunner>
+        <RootNamespace>MQTTnet.Tests.Extensions.Socks5</RootNamespace>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <PackageReference Include="MSTest" Version="4.0.2" />
+    </ItemGroup>
+
+    <ItemGroup>
+        <ProjectReference Include="..\MQTTnet\MQTTnet.csproj" />
+        <ProjectReference Include="..\MQTTnet.Server\MQTTnet.Server.csproj" />
+        <ProjectReference Include="..\MQTTnet.Extensions.Socks5\MQTTnet.Extensions.Socks5.csproj" />
+    </ItemGroup>
+
+    <ItemGroup>
+        <None Include=".\..\..\global.json"
+                Link="global.json"
+                CopyToOutputDirectory="PreserveNewest"/>
+    </ItemGroup>
+</Project>

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<string>(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<MqttProxyException>(() => 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<MqttProxyException>(() => 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<MqttProxyException>(() => 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<MqttProxyException>(() => 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<MqttProxyException>(() => 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<MqttProxyException>(() => 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<byte>(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<byte>(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<byte>(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<OperationCanceledException>(() => 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<MqttProxyException>(() => 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<MqttProxyException>(() => 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<string>(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<byte>(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<MqttProxyException>(() => 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();
+        }
+    }
+}