Compare commits

..
Author SHA1 Message Date
christian 076039abc7 Add binary files to repo for testing purposes 2025-10-24 17:35:12 +02:00
christian 61cbaa06a8 Expose more properties 2025-10-20 21:41:20 +02:00
christian 1114fa041d Add new PowerShell project 2025-10-20 20:52:31 +02:00
322 changed files with 2578 additions and 3540 deletions
+13 -15
View File
@@ -3,7 +3,7 @@ name: CI
on: [push, pull_request]
env:
VERSION: "5.2.0.${{github.run_number}}"
VERSION: "5.1.0.${{github.run_number}}"
PACKAGE_SUFFIX: ""
jobs:
@@ -17,26 +17,19 @@ 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 (signed)
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: Build package
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
- name: Upload nuget packages
uses: actions/upload-artifact@v4
@@ -55,19 +48,24 @@ jobs:
with:
dotnet-version: |
8.0.x
10.0.x
- name: Checkout code
uses: actions/checkout@v4
- name: Execute tests
run: dotnet test --framework net10.0 --configuration Release --project Source/MQTTnet.Tests/MQTTnet.Tests.csproj
run: dotnet test --framework net8.0 --configuration Release 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' && (github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository)
if: github.repository == 'dotnet/MQTTnet'
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:
@@ -75,7 +73,7 @@ jobs:
path: nugets
- name: Install sign CLI tool
run: dotnet tool install --tool-path . sign --version 0.9.1-beta.25379.1
run: dotnet tool install --tool-path . sign --version 0.9.0-beta.23127.3
- name: Sign nugets
shell: pwsh
@@ -116,7 +114,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.1.0-rc' etc.
# Tag schema is 'v5.0.0-rc' etc.
if: startsWith(github.ref, 'refs/tags/v')
needs: [sign, test]
runs-on: ubuntu-latest
+27 -3
View File
@@ -1,6 +1,6 @@
<Project>
<PropertyGroup>
<TargetFrameworks>net8.0;net10.0</TargetFrameworks>
<TargetFramework>net8.0</TargetFramework>
<Product>MQTTnet</Product>
<Authors>The contributors of MQTTnet</Authors>
<Company>The contributors of MQTTnet</Company>
@@ -11,7 +11,6 @@
<AllowUnsafeBlocks>true</AllowUnsafeBlocks>
<EnforceCodeStyleInBuild>false</EnforceCodeStyleInBuild>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>disable</Nullable>
<NeutralLanguage>en-US</NeutralLanguage>
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
@@ -25,10 +24,35 @@
<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>
-24
View File
@@ -1,24 +0,0 @@
<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>
+95
View File
@@ -0,0 +1,95 @@
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
-28
View File
@@ -1,28 +0,0 @@
<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>
-20
View File
@@ -6,9 +6,6 @@
// ReSharper disable UnusedMember.Global
// ReSharper disable InconsistentNaming
using System;
using System.Text;
namespace MQTTnet.Samples.Client;
public static class Client_Publish_Samples
@@ -87,21 +84,4 @@ 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();
}
}
-86
View File
@@ -1,86 +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.
// 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);
}
}
+11 -12
View File
@@ -7,7 +7,7 @@
// ReSharper disable InconsistentNaming
// ReSharper disable UnusedMember.Local
using System.Text;
using MQTTnet.Extensions.TopicTemplate;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Samples.Helpers;
@@ -16,6 +16,8 @@ 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()
{
/*
@@ -40,7 +42,7 @@ public static class Client_Subscribe_Samples
await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicFilter("mqttnet/samples/topic/2").Build();
var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicTemplate(sampleTemplate.WithParameter("id", "2")).Build();
await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
@@ -53,7 +55,7 @@ public static class Client_Subscribe_Samples
public static async Task Send_Responses()
{
/*
* This sample subscribes to a topic and sends a detailed response to the client. This requires at least QoS level 1 to work!
* This sample subscribes to a topic and sends a response to the broker. This requires at least QoS level 1 to work!
*/
var mqttFactory = new MqttClientFactory();
@@ -68,7 +70,7 @@ public static class Client_Subscribe_Samples
args.ResponseReasonString = "That did not work!";
// User properties require MQTT v5!
args.ResponseUserProperties.Add(new MqttUserProperty("My", Encoding.UTF8.GetBytes("Data")));
args.ResponseUserProperties.Add(new MqttUserProperty("My", "Data"));
// Now the broker will resend the message again.
return Task.CompletedTask;
@@ -78,7 +80,7 @@ public static class Client_Subscribe_Samples
await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicFilter("topic/test").Build();
var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicTemplate(sampleTemplate.WithParameter("id", "1")).Build();
var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
@@ -104,10 +106,9 @@ 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()
.WithTopicFilter(t => t.WithTopic("topic/1").WithAtLeastOnceQoS())
.WithTopicFilter(t => t.WithTopic("topic/2").WithAtMostOnceQoS())
.WithTopicFilter(t => t.WithTopic("topic/3").WithNoLocal())
.WithTopicFilter(t => t.WithRetainHandling(MqttRetainHandling.SendAtSubscribe))
.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);
@@ -131,9 +132,7 @@ public static class Client_Subscribe_Samples
await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
var topicFilter = mqttFactory.CreateTopicFilterBuilder().WithTopic("mqttnet/samples/topic/2").WithAtLeastOnceQoS();
var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicFilter(topicFilter).Build();
var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder().WithTopicTemplate(sampleTemplate.WithParameter("id", "1")).Build();
var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
@@ -1,75 +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 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();
}
}
-1
View File
@@ -12,7 +12,6 @@
<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"/>
-2
View File
@@ -59,12 +59,10 @@ 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;
}
@@ -2,6 +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 Microsoft.AspNetCore.Builder;
using Microsoft.Extensions.DependencyInjection;
using MQTTnet.Server;
@@ -2,6 +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 MQTTnet.Server;
namespace MQTTnet.AspNetCore;
@@ -2,6 +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.Runtime.InteropServices;
namespace MQTTnet.AspNetCore;
@@ -2,6 +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 Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Routing;
@@ -2,6 +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 MQTTnet.Adapter;
using MQTTnet.Diagnostics.Logger;
using MQTTnet.Formatter;
@@ -2,10 +2,13 @@
// 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;
@@ -2,6 +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.Threading.Tasks;
using Microsoft.AspNetCore.Connections;
using Microsoft.AspNetCore.Connections.Features;
using MQTTnet.Adapter;
@@ -2,6 +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.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Hosting;
using MQTTnet.Diagnostics.Logger;
using MQTTnet.Server;
@@ -2,6 +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.Linq;
using Microsoft.AspNetCore.Http;
namespace MQTTnet.AspNetCore;
@@ -2,8 +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.Net;
using System.Net.WebSockets;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Http;
using MQTTnet.Adapter;
using MQTTnet.Diagnostics.Logger;
@@ -2,6 +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.Runtime.InteropServices;
using MQTTnet.Adapter;
@@ -2,6 +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 Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection.Extensions;
using Microsoft.Extensions.Hosting;
@@ -2,10 +2,13 @@
// 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;
@@ -2,9 +2,13 @@
// 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;
@@ -2,6 +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.IO.Pipelines;
using System.Net.Sockets;
@@ -2,7 +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.Buffers;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO.Pipelines;
using System.Net.Sockets;
@@ -1 +1,3 @@
using Microsoft.VisualStudio.TestTools.UnitTesting;
[assembly: DoNotParallelize]
@@ -2,6 +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.Threading;
using System.Threading.Tasks;
using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs;
using MQTTnet.Internal;
@@ -2,6 +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.IO;
using System.Threading;
using BenchmarkDotNet.Attributes;
using MQTTnet.Adapter;
using MQTTnet.Diagnostics.Logger;
@@ -1,5 +1,6 @@
using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs;
using System;
namespace MQTTnet.Benchmarks;
@@ -2,6 +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.Text;
using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs;
@@ -2,6 +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 BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs;
using MQTTnet.Formatter;
+3
View File
@@ -2,6 +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.Linq;
using BenchmarkDotNet.Running;
namespace MQTTnet.Benchmarks;
@@ -4,7 +4,9 @@ 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;
@@ -1,6 +1,7 @@
using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs;
using MQTTnet.Tests.Mockups;
using MQTTnet.Tests.Server;
namespace MQTTnet.Benchmarks;
@@ -4,7 +4,10 @@
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;
@@ -5,6 +5,7 @@
using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs;
using MQTTnet.Tests.Mockups;
using MQTTnet.Tests.Server;
namespace MQTTnet.Benchmarks;
@@ -2,6 +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.Collections.Generic;
using System.Linq;
using BenchmarkDotNet.Attributes;
using MQTTnet.Server;
@@ -5,6 +5,8 @@
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;
@@ -1,3 +1,6 @@
using System;
using System.Collections.Generic;
namespace MQTTnet.Benchmarks;
public class TopicGenerator
@@ -2,6 +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.Collections.Generic;
using System.Linq;
using BenchmarkDotNet.Attributes;
using MQTTnet.Server;
@@ -2,6 +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;
namespace MQTTnet.Extensions.Rpc;
public sealed class DefaultMqttRpcClientTopicGenerationStrategy : IMqttRpcClientTopicGenerationStrategy
@@ -2,6 +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.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Protocol;
namespace MQTTnet.Extensions.Rpc;
@@ -2,6 +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;
namespace MQTTnet.Extensions.Rpc;
public static class MqttFactoryExtensions
@@ -2,8 +2,12 @@
// 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;
@@ -2,7 +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.Collections.Generic;
using System.Text;
using System.Threading.Tasks;
using MQTTnet.Protocol;
namespace MQTTnet.Extensions.Rpc;
@@ -2,6 +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;
namespace MQTTnet.Extensions.Rpc;
public sealed class MqttRpcClientOptionsBuilder
@@ -2,6 +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.Collections.Generic;
using MQTTnet.Protocol;
namespace MQTTnet.Extensions.Rpc;
@@ -1,21 +0,0 @@
<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>
@@ -1,39 +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.
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());
}
}
@@ -1,109 +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.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);
}
}
@@ -1,39 +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.
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;
}
@@ -1,65 +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.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}";
}
}
@@ -1,94 +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.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;
}
}
@@ -1,507 +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.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);
}
}
@@ -2,7 +2,11 @@
// 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;
@@ -2,6 +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.Linq;
using MQTTnet.Packets;
using MQTTnet.Protocol;
@@ -0,0 +1,47 @@
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);
}
}
@@ -0,0 +1,31 @@
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.");
}
}
@@ -0,0 +1,13 @@
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());
}
}
@@ -0,0 +1,52 @@
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);
}
}
@@ -0,0 +1,74 @@
// 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();
}
}
}
@@ -0,0 +1,26 @@
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.");
}
}
@@ -0,0 +1,16 @@
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.");
}
}
@@ -0,0 +1,55 @@
// 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);
}
}
@@ -0,0 +1,23 @@
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);
}
}
@@ -0,0 +1,28 @@
<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>
@@ -0,0 +1,30 @@
// 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; }
}
@@ -0,0 +1,50 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.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;
}
}
+15
View File
@@ -0,0 +1,15 @@
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
@@ -0,0 +1,55 @@
{
"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": ""
}
}
}
File diff suppressed because it is too large Load Diff
@@ -40,7 +40,6 @@ 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)
@@ -51,26 +50,4 @@ 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;
}
}
@@ -51,7 +51,6 @@ 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);
@@ -65,32 +64,4 @@ 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;
}
}
@@ -202,14 +202,9 @@ 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
}
}
@@ -218,8 +218,6 @@ 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);
@@ -581,4 +579,4 @@ public sealed class MqttConnectedClient : IDisposable
_logger.Warning(exception, "Client '{0}': Error while sending DISCONNECT packet (ReasonCode = {1})", Id, options.ReasonCode);
}
}
}
}
+3
View File
@@ -2,6 +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.Threading;
using System.Threading.Tasks;
using MQTTnet.Exceptions;
namespace MQTTnet.TestApp;
+2
View File
@@ -2,6 +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.Threading.Tasks;
using MQTTnet.Diagnostics.Logger;
namespace MQTTnet.TestApp;
+2
View File
@@ -2,8 +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.Net;
using System.Text;
using System.Threading.Tasks;
using MQTTnet.Diagnostics.Logger;
using MQTTnet.Internal;
using MQTTnet.Protocol;
@@ -2,6 +2,7 @@
<PropertyGroup>
<OutputType>Exe</OutputType>
<DebugType>Full</DebugType>
<IsPackable>false</IsPackable>
</PropertyGroup>
@@ -1,5 +1,10 @@
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;
@@ -2,6 +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.Text;
using MQTTnet.Diagnostics.Logger;
@@ -2,9 +2,13 @@
// 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;
+4
View File
@@ -2,6 +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.Threading;
using System.Threading.Tasks;
namespace MQTTnet.TestApp;
public static class Program
@@ -2,7 +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.Security.Authentication;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Formatter;
using MQTTnet.Internal;
using MQTTnet.Protocol;
@@ -2,6 +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.Threading;
using System.Threading.Tasks;
using MQTTnet.Diagnostics.Logger;
using MQTTnet.Server;
+4
View File
@@ -2,7 +2,11 @@
// 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;
+2
View File
@@ -1,3 +1,5 @@
using System;
using System.Collections.Generic;
using MQTTnet.Extensions.TopicTemplate;
namespace MQTTnet.TestApp;
@@ -1,5 +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.
[assembly: DoNotParallelize]
@@ -1,529 +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;
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; }
}
@@ -1,5 +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.
global using Microsoft.VisualStudio.TestTools.UnitTesting;
@@ -1,28 +0,0 @@
<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>
@@ -1,566 +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;
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();
}
}
}
@@ -1,112 +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;
using MQTTnet.Extensions.Socks5;
namespace MQTTnet.Tests.Extensions.Socks5;
[TestClass]
public sealed class Socks5ProxyOptionsValidation_Tests
{
[TestMethod]
public void Builder_Build_Without_Host_Throws()
{
var builder = new Socks5ProxyOptionsBuilder();
Assert.ThrowsExactly<InvalidOperationException>(() => builder.Build());
}
[TestMethod]
public void Builder_Build_With_Invalid_Port_Throws()
{
var builder = new Socks5ProxyOptionsBuilder()
.WithHost("127.0.0.1")
.WithPort(0);
Assert.ThrowsExactly<InvalidOperationException>(() => builder.Build());
}
[TestMethod]
public void Builder_Build_With_NonPositive_HandshakeTimeout_Throws()
{
var builder = new Socks5ProxyOptionsBuilder()
.WithHost("127.0.0.1")
.WithHandshakeTimeout(TimeSpan.Zero);
Assert.ThrowsExactly<InvalidOperationException>(() => builder.Build());
}
[TestMethod]
public void Builder_Build_With_Password_But_Without_Username_Throws()
{
var builder = new Socks5ProxyOptionsBuilder()
.WithHost("127.0.0.1")
.WithCredentials(null, new byte[] { 0x01, 0x02 });
Assert.ThrowsExactly<InvalidOperationException>(() => builder.Build());
}
[TestMethod]
public void Builder_Build_With_Username_But_Without_Password_Throws()
{
var builder = new Socks5ProxyOptionsBuilder()
.WithHost("127.0.0.1")
.WithCredentials("alice", Array.Empty<byte>());
Assert.ThrowsExactly<InvalidOperationException>(() => builder.Build());
}
[TestMethod]
public void StreamProvider_Ctor_With_Invalid_Port_Throws()
{
var options = new Socks5ProxyOptions
{
Host = "127.0.0.1",
Port = 70000
};
Assert.ThrowsExactly<ArgumentOutOfRangeException>(() => _ = new Socks5StreamProvider(options));
}
[TestMethod]
public void StreamProvider_Ctor_With_NonPositive_HandshakeTimeout_Throws()
{
var options = new Socks5ProxyOptions
{
Host = "127.0.0.1",
Port = 1080,
HandshakeTimeout = TimeSpan.Zero
};
Assert.ThrowsExactly<ArgumentOutOfRangeException>(() => _ = new Socks5StreamProvider(options));
}
[TestMethod]
public void StreamProvider_Ctor_With_Password_But_Without_Username_Throws()
{
var options = new Socks5ProxyOptions
{
Host = "127.0.0.1",
Port = 1080,
Password = new byte[] { 0x01 }
};
Assert.ThrowsExactly<ArgumentException>(() => _ = new Socks5StreamProvider(options));
}
[TestMethod]
public void StreamProvider_Ctor_With_Username_But_Without_Password_Throws()
{
var options = new Socks5ProxyOptions
{
Host = "127.0.0.1",
Port = 1080,
Username = "alice",
Password = Array.Empty<byte>()
};
Assert.ThrowsExactly<ArgumentException>(() => _ = new Socks5StreamProvider(options));
}
}

Some files were not shown because too many files have changed in this diff Show More