Compare commits

..
Author SHA1 Message Date
christian f6357c42b6 Prepare recommended fix 2026-06-16 19:10:11 +02:00
christian 16512d5b89 Create Unit Test 2026-06-16 19:09:58 +02:00
39 changed files with 215 additions and 3126 deletions
+3 -3
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:
@@ -32,11 +32,11 @@ jobs:
- 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
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: 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 }}
run: dotnet build MQTTnet.sln --configuration Release /p:FileVersion=${{ env.VERSION }} /p:AssemblyVersion=${{ env.VERSION }} /p:PackageVersion=${{ env.VERSION }}${{ env.PACKAGE_SUFFIX }}
- name: Upload nuget packages
uses: actions/upload-artifact@v4
+90
View File
@@ -0,0 +1,90 @@
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
Directory.Build.targets = Directory.Build.targets
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
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
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>
+3 -3
View File
@@ -12,9 +12,9 @@
# MQTTnet
MQTTnet is a high performance .NET library for MQTT based communication.
It provides a MQTT client and a MQTT server (broker) and supports the MQTT protocol up to version 5.
It is compatible with mostly any supported .NET Framework version and CPU architecture.
MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (
broker) and supports the MQTT protocol up to version 5. It is compatible with mostly any supported .NET Framework
version and CPU architecture.
## Features
-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);
}
}
-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"/>
@@ -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);
}
}
@@ -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));
}
}
@@ -1,97 +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.IO;
using System.Net;
using System.Net.Security;
using System.Net.Sockets;
using System.Security.Cryptography;
using System.Security.Cryptography.X509Certificates;
using System.Threading.Tasks;
using MQTTnet.Certificates;
using MQTTnet.Server;
namespace MQTTnet.Tests.Extensions.Socks5;
internal static class Socks5TestHelper
{
public static async Task<BrokerHandle> StartBrokerAsync()
{
var factory = new MqttServerFactory();
var options = factory.CreateServerOptionsBuilder()
.WithDefaultEndpoint()
.WithDefaultEndpointPort(0)
.Build();
var server = factory.CreateMqttServer(options);
await server.StartAsync();
return new BrokerHandle(server, options);
}
public static int GetBrokerPort(BrokerHandle handle)
{
return handle.Options.DefaultEndpointOptions.Port;
}
public static async Task<TlsBrokerHandle> StartTlsBrokerAsync()
{
var certificate = CreateSelfSignedServerCertificate();
var factory = new MqttServerFactory();
var options = factory.CreateServerOptionsBuilder()
.WithoutDefaultEndpoint()
.WithEncryptedEndpoint()
.WithEncryptedEndpointPort(0)
.WithEncryptionCertificate(new InMemoryCertificateProvider(certificate))
.Build();
var server = factory.CreateMqttServer(options);
await server.StartAsync();
return new TlsBrokerHandle(server, options, certificate);
}
public static int GetTlsBrokerPort(TlsBrokerHandle handle)
{
return handle.Options.TlsEndpointOptions.Port;
}
public static X509Certificate2 CreateSelfSignedServerCertificate()
{
var sanBuilder = new SubjectAlternativeNameBuilder();
sanBuilder.AddIpAddress(IPAddress.Loopback);
sanBuilder.AddIpAddress(IPAddress.IPv6Loopback);
sanBuilder.AddDnsName("localhost");
using var rsa = RSA.Create(2048);
var certRequest = new CertificateRequest("CN=localhost", rsa, HashAlgorithmName.SHA256, RSASignaturePadding.Pkcs1);
certRequest.CertificateExtensions.Add(
new X509KeyUsageExtension(X509KeyUsageFlags.DataEncipherment | X509KeyUsageFlags.KeyEncipherment | X509KeyUsageFlags.DigitalSignature, false));
certRequest.CertificateExtensions.Add(sanBuilder.Build());
using var cert = certRequest.CreateSelfSigned(DateTimeOffset.UtcNow.AddMinutes(-10), DateTimeOffset.UtcNow.AddMinutes(60));
#pragma warning disable SYSLIB0057
return new X509Certificate2(
cert.Export(X509ContentType.Pfx),
(string)null,
X509KeyStorageFlags.MachineKeySet | X509KeyStorageFlags.Exportable);
#pragma warning restore SYSLIB0057
}
internal sealed record BrokerHandle(MqttServer Server, MqttServerOptions Options);
internal sealed record TlsBrokerHandle(MqttServer Server, MqttServerOptions Options, X509Certificate2 Certificate);
sealed class InMemoryCertificateProvider : ICertificateProvider
{
readonly X509Certificate2 _certificate;
public InMemoryCertificateProvider(X509Certificate2 certificate)
{
_certificate = certificate;
}
public X509Certificate2 GetCertificate() => _certificate;
}
}
@@ -5,12 +5,8 @@
using System.Buffers;
using System.Diagnostics;
using System.Globalization;
using System.Net;
using System.Net.Sockets;
using System.Security.Cryptography.X509Certificates;
using System.Text;
using MQTTnet.Adapter;
using MQTTnet.Diagnostics.Logger;
using MQTTnet.Exceptions;
using MQTTnet.Formatter;
using MQTTnet.Internal;
@@ -459,31 +455,64 @@ public sealed class MqttClient_Tests : BaseTestClass
}
[TestMethod]
public async Task KeepAlive_Ping_Send_Timeout_Raises_Disconnected()
public async Task Raise_Disconnected_On_KeepAlive_Timeout()
{
var adapterFactory = new KeepAliveTimeoutAdapterFactory();
var disconnected = new TaskCompletionSource<MqttClientDisconnectedEventArgs>(TaskCreationOptions.RunContinuationsAsynchronously);
using var listener = new TcpListener(System.Net.IPAddress.Loopback, 0);
listener.Start();
using var client = new MqttClientFactory().CreateMqttClient(adapterFactory);
client.DisconnectedAsync += eventArgs =>
var port = ((System.Net.IPEndPoint)listener.LocalEndpoint).Port;
using var testEnvironment = new TestEnvironment(TestContext);
using var client = testEnvironment.CreateClient();
var options = new MqttClientOptionsBuilder()
.WithTcpServer("127.0.0.1", port)
.WithKeepAlivePeriod(TimeSpan.FromSeconds(5))
.WithTimeout(TimeSpan.FromSeconds(10)) // For ping timeout and others
.Build();
var disconnectedTask = new TaskCompletionSource<MqttClientDisconnectedEventArgs>();
client.DisconnectedAsync += e =>
{
disconnected.TrySetResult(eventArgs);
disconnectedTask.TrySetResult(e);
return CompletedTask.Instance;
};
var options = new MqttClientOptionsBuilder()
.WithTcpServer("localhost")
.WithKeepAlivePeriod(TimeSpan.FromSeconds(1))
.Build();
// Fake server implementation
_ = Task.Run(async () =>
{
using var clientSocket = await listener.AcceptTcpClientAsync();
await using var stream = clientSocket.GetStream();
await client.ConnectAsync(options);
// Read CONNECT
var buffer = new byte[1024];
_ = await stream.ReadAsync(buffer.AsMemory(), CancellationToken.None);
await adapterFactory.Adapter.PingRequestReceived.Task.WaitAsync(TimeSpan.FromSeconds(3));
var disconnectedEventArgs = await disconnected.Task.WaitAsync(TimeSpan.FromSeconds(3));
// Send CONNACK
var connack = new byte[] { 0x20, 0x02, 0x00, 0x00 };
await stream.WriteAsync(connack.AsMemory(), CancellationToken.None);
Assert.IsTrue(disconnectedEventArgs.ClientWasConnected);
Assert.IsInstanceOfType<OperationCanceledException>(disconnectedEventArgs.Exception);
Assert.IsFalse(client.IsConnected);
// Wait until client drops connection because of ping timeout
while (true)
{
var readCount = await stream.ReadAsync(buffer.AsMemory(), CancellationToken.None);
if (readCount == 0)
{
break; // connection closed
}
}
});
await client.ConnectAsync(options, CancellationToken.None);
var timeoutTask = Task.Delay(TimeSpan.FromSeconds(10));
var completedTask = await Task.WhenAny(disconnectedTask.Task, timeoutTask);
listener.Stop();
Assert.AreEqual(disconnectedTask.Task, completedTask, "DisconnectedAsync was not raised.");
var result = await disconnectedTask.Task;
Assert.IsNotNull(result);
}
[TestMethod]
@@ -902,189 +931,4 @@ public sealed class MqttClient_Tests : BaseTestClass
Assert.IsTrue(client1.TryPingAsync().GetAwaiter().GetResult());
Assert.IsFalse(disconnectedFired);
}
// Regression test for issue #2079.
// A late PUBACK (one whose matching publish request was already cancelled or timed
// out client-side) must NOT raise a protocol violation and must NOT disconnect the
// client. Before the fix, the client received the late PUBACK, failed to find a
// waiter and threw a MqttProtocolViolationException, tearing down the connection.
[TestMethod]
[DataRow(MqttQualityOfServiceLevel.AtLeastOnce)]
[DataRow(MqttQualityOfServiceLevel.ExactlyOnce)]
public async Task Cancelled_Publish_Does_Not_Disconnect_Client(MqttQualityOfServiceLevel qos)
{
using var testEnvironment = new TestEnvironment(TestContext);
var server = await testEnvironment.StartServer();
// Delay the broker's acknowledgement long enough for the publish caller to give
// up. When the caller's cancellation token fires the awaitable is removed; the
// (slightly later) PUBACK / PUBREC then arrives without a matching waiter.
server.InterceptingPublishAsync += async e =>
{
await Task.Delay(TimeSpan.FromSeconds(2), CancellationToken.None);
};
var client = await testEnvironment.ConnectClient();
var disconnected = false;
client.DisconnectedAsync += _ =>
{
disconnected = true;
return CompletedTask.Instance;
};
var message = new MqttApplicationMessageBuilder().WithTopic("late-ack").WithQualityOfServiceLevel(qos).Build();
using (var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(250)))
{
await Assert.ThrowsExactlyAsync<MqttCommunicationTimedOutException>(() => client.PublishAsync(message, cts.Token));
}
// Wait long enough for the delayed broker acknowledgement to come back.
await Task.Delay(TimeSpan.FromSeconds(3));
Assert.IsFalse(disconnected, "Client must not disconnect when a late acknowledgement arrives.");
Assert.IsTrue(client.IsConnected, "Client must still be connected after a late acknowledgement.");
// The connection must remain usable for further publishes.
var followUp = new MqttApplicationMessageBuilder().WithTopic("late-ack").WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce).Build();
await client.PublishAsync(followUp);
}
// Regression test for issue #2078.
// The MqttClient must support concurrent publishes from multiple threads without
// dropping messages, mismatching acknowledgements or disconnecting the client.
[TestMethod]
[DataRow(MqttQualityOfServiceLevel.AtMostOnce)]
[DataRow(MqttQualityOfServiceLevel.AtLeastOnce)]
[DataRow(MqttQualityOfServiceLevel.ExactlyOnce)]
public async Task Concurrent_Publish_From_Multiple_Threads(MqttQualityOfServiceLevel qos)
{
const int Threads = 8;
const int MessagesPerThread = 250;
using var testEnvironment = new TestEnvironment(TestContext);
await testEnvironment.StartServer();
var publisher = await testEnvironment.ConnectClient();
var subscriber = await testEnvironment.ConnectClient();
var receivedCount = 0;
subscriber.ApplicationMessageReceivedAsync += _ =>
{
Interlocked.Increment(ref receivedCount);
return CompletedTask.Instance;
};
await subscriber.SubscribeAsync("concurrent/#", qos);
var disconnected = false;
publisher.DisconnectedAsync += _ =>
{
disconnected = true;
return CompletedTask.Instance;
};
var tasks = Enumerable.Range(0, Threads).Select(threadIndex => Task.Run(async () =>
{
for (var i = 0; i < MessagesPerThread; i++)
{
var message = new MqttApplicationMessageBuilder()
.WithTopic($"concurrent/{threadIndex}/{i}")
.WithPayload(BitConverter.GetBytes(i))
.WithQualityOfServiceLevel(qos)
.Build();
await publisher.PublishAsync(message);
}
})).ToArray();
await Task.WhenAll(tasks);
// Allow the broker to deliver any in-flight messages to the subscriber.
SpinWait.SpinUntil(() => receivedCount >= Threads * MessagesPerThread, TimeSpan.FromSeconds(30));
Assert.IsFalse(disconnected, "Publisher must not disconnect during concurrent publishes.");
Assert.IsTrue(publisher.IsConnected);
Assert.AreEqual(Threads * MessagesPerThread, receivedCount);
}
sealed class KeepAliveTimeoutAdapterFactory : IMqttClientAdapterFactory
{
public KeepAliveTimeoutAdapter Adapter { get; private set; }
public IMqttChannelAdapter CreateClientAdapter(MqttClientOptions options, MqttPacketInspector packetInspector, IMqttNetLogger logger)
{
Adapter = new KeepAliveTimeoutAdapter(options.ProtocolVersion, options.WriterBufferSize, options.WriterBufferSizeMax);
return Adapter;
}
}
sealed class KeepAliveTimeoutAdapter : IMqttChannelAdapter
{
int _connAckReturned;
public KeepAliveTimeoutAdapter(MqttProtocolVersion protocolVersion, int writerBufferSize, int writerBufferSizeMax)
{
PacketFormatterAdapter = new MqttPacketFormatterAdapter(protocolVersion, new MqttBufferWriter(writerBufferSize, writerBufferSizeMax));
}
public long BytesReceived { get; }
public long BytesSent { get; }
public X509Certificate2 ClientCertificate { get; }
public EndPoint RemoteEndPoint { get; } = new DnsEndPoint("localhost", 1883);
public EndPoint LocalEndPoint { get; } = new DnsEndPoint("localhost", 0);
public bool IsSecureConnection { get; }
public MqttPacketFormatterAdapter PacketFormatterAdapter { get; }
public TaskCompletionSource PingRequestReceived { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
public Task ConnectAsync(CancellationToken cancellationToken)
{
return CompletedTask.Instance;
}
public Task DisconnectAsync(CancellationToken cancellationToken)
{
return CompletedTask.Instance;
}
public void Dispose()
{
}
public async Task<MqttPacket> ReceivePacketAsync(CancellationToken cancellationToken)
{
if (Interlocked.Exchange(ref _connAckReturned, 1) == 0)
{
return new MqttConnAckPacket
{
ReturnCode = MqttConnectReturnCode.ConnectionAccepted
};
}
await Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken);
return null;
}
public void ResetStatistics()
{
}
public Task SendPacketAsync(MqttPacket packet, CancellationToken cancellationToken)
{
if (packet is MqttPingReqPacket)
{
PingRequestReceived.TrySetResult();
return Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken);
}
return CompletedTask.Instance;
}
}
}
}
@@ -1,40 +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.Exceptions;
using MQTTnet.Formatter;
using MQTTnet.Packets;
namespace MQTTnet.Tests.Formatter;
[TestClass]
public sealed class MqttV5PublishPropertyValidation_Tests
{
[TestMethod]
[DataRow("0100")]
[DataRow("0200000000")]
[DataRow("030000")]
[DataRow("08000161")]
[DataRow("090000")]
[DataRow("230001")]
public void Publish_Rejects_Duplicate_Singleton_Property(string propertyHex)
{
var property = Convert.FromHexString(propertyHex);
byte[] packet = [0x30, (byte)(4 + 2 * property.Length), 0, 1, (byte)'a', (byte)(2 * property.Length), .. property, .. property];
Assert.ThrowsExactly<MqttProtocolViolationException>(() => MqttPacketSerializationHelper.DecodePacket(packet, MqttProtocolVersion.V500));
}
[TestMethod]
public void Publish_Allows_Repeated_Subscription_Identifiers_And_User_Properties()
{
byte[] packet = [0x30, 18, 0, 1, (byte)'a', 14, 0x0B, 1, 0x26, 0, 0, 0, 0, 0x0B, 1, 0x26, 0, 0, 0, 0];
var decoded = (MqttPublishPacket)MqttPacketSerializationHelper.DecodePacket(packet, MqttProtocolVersion.V500);
CollectionAssert.AreEqual(new uint[] { 1, 1 }, decoded.SubscriptionIdentifiers);
Assert.HasCount(2, decoded.UserProperties);
}
}
@@ -1,62 +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.Adapter;
using MQTTnet.Exceptions;
using MQTTnet.Formatter;
using MQTTnet.Formatter.V5;
using MQTTnet.Packets;
namespace MQTTnet.Tests.Formatter;
[TestClass]
public sealed class MqttV5SubscribeIdentifier_Tests
{
[TestMethod]
[DataRow("0B00")]
[DataRow("0B010B01")]
[DataRow("0B010B02")]
[DataRow("0B0126000000000B02")]
public void Reject_Invalid_Subscription_Identifier(string propertiesHex)
{
var properties = Convert.FromHexString(propertiesHex);
byte[] packet = [0x82, (byte)(7 + properties.Length), 0, 1, (byte)properties.Length, .. properties, 0, 1, (byte)'a', 0];
Assert.ThrowsExactly<MqttProtocolViolationException>(() => MqttPacketSerializationHelper.DecodePacket(packet, MqttProtocolVersion.V500));
}
[TestMethod]
[DataRow("", 0U)]
[DataRow("0B01", 1U)]
[DataRow("0BFFFFFF7F", 268435455U)]
[DataRow("26000000000B012600000000", 1U)]
public void Accept_Valid_Subscription_Identifier(string propertiesHex, uint expected)
{
var properties = Convert.FromHexString(propertiesHex);
byte[] packet = [0x82, (byte)(7 + properties.Length), 0, 1, (byte)properties.Length, .. properties, 0, 1, (byte)'a', 0];
var decoded = (MqttSubscribePacket)MqttPacketSerializationHelper.DecodePacket(packet, MqttProtocolVersion.V500);
Assert.AreEqual(expected, decoded.SubscriptionIdentifier);
}
[TestMethod]
public void Accept_Same_Identifier_In_Separate_Subscribe_Packets()
{
byte[] packet = [0x82, 9, 0, 1, 2, 0x0B, 1, 0, 1, (byte)'a', 0];
var decoder = new MqttV5PacketDecoder();
var received = new ReceivedMqttPacket(packet[0], new ArraySegment<byte>(packet, 2, packet.Length - 2), packet.Length);
Assert.AreEqual(1U, ((MqttSubscribePacket)decoder.Decode(received)).SubscriptionIdentifier);
Assert.AreEqual(1U, ((MqttSubscribePacket)decoder.Decode(received)).SubscriptionIdentifier);
}
[TestMethod]
public void Accept_Repeated_Subscription_Identifiers_In_Publish()
{
byte[] packet = [0x30, 8, 0, 1, (byte)'a', 4, 0x0B, 1, 0x0B, 1];
var decoded = (MqttPublishPacket)MqttPacketSerializationHelper.DecodePacket(packet, MqttProtocolVersion.V500);
CollectionAssert.AreEqual(new uint[] { 1, 1 }, decoded.SubscriptionIdentifiers);
}
}
@@ -2,11 +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.Runtime.CompilerServices;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Server;
using MQTTnet.Server.Exceptions;
namespace MQTTnet.Tests.Internal;
@@ -14,57 +11,6 @@ namespace MQTTnet.Tests.Internal;
[TestClass]
public sealed class MqttPacketBusItem_Tests
{
[TestMethod]
public void Fail_Without_Waiter_Does_Not_Raise_Unobserved_Exception()
{
var exception = new MqttPendingMessagesOverflowException("test", MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage);
var unobservedExceptions = 0;
void OnUnobservedTaskException(object sender, UnobservedTaskExceptionEventArgs args)
{
if (args.Exception.Flatten().InnerExceptions.Contains(exception))
{
Interlocked.Increment(ref unobservedExceptions);
args.SetObserved();
}
}
TaskScheduler.UnobservedTaskException += OnUnobservedTaskException;
try
{
var taskReference = CreateFailedPacketBusItem(exception);
GC.Collect();
GC.WaitForPendingFinalizers();
GC.Collect();
Assert.IsFalse(taskReference.IsAlive, "The faulted task must be collected to exercise unobserved exception reporting.");
Assert.AreEqual(0, unobservedExceptions);
}
finally
{
TaskScheduler.UnobservedTaskException -= OnUnobservedTaskException;
}
}
[TestMethod]
[DataRow(false)]
[DataRow(true)]
public async Task Wait_Failed_Packet_Bus_Item_Preserves_Exception(bool waitBeforeFailure)
{
var item = new MqttPacketBusItem(new MqttPublishPacket());
var exception = new MqttPendingMessagesOverflowException("test", MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage);
var task = waitBeforeFailure ? item.WaitAsync() : null;
item.Fail(exception);
task ??= item.WaitAsync();
var thrownException = await Assert.ThrowsExactlyAsync<MqttPendingMessagesOverflowException>(async () => await task);
Assert.AreSame(exception, thrownException);
Assert.IsTrue(task.IsFaulted);
}
[TestMethod]
public void Fire_Completed_Event()
{
@@ -94,13 +40,4 @@ public sealed class MqttPacketBusItem_Tests
await item.WaitAsync();
});
}
[MethodImpl(MethodImplOptions.NoInlining)]
static WeakReference CreateFailedPacketBusItem(Exception exception)
{
// Keep strong references out of the collecting method, including in debug builds.
var item = new MqttPacketBusItem(new MqttPublishPacket());
item.Fail(exception);
return new WeakReference(item.WaitAsync());
}
}
+1 -27
View File
@@ -3,10 +3,7 @@
// See the LICENSE file in the project root for more information.
using System.Net;
using System.Net.Security;
using System.Net.Sockets;
using System.Reflection;
using System.Security.Cryptography.X509Certificates;
using MQTTnet.Implementations;
namespace MQTTnet.Tests;
@@ -15,21 +12,6 @@ namespace MQTTnet.Tests;
[TestClass]
public class MqttTcpChannel_Tests
{
[TestMethod]
public void Certificate_Validation_Uses_Default_Handler_When_No_Custom_Handler()
{
var options = new MqttClientOptionsBuilder().WithTcpServer("localhost")
.WithTlsOptions(o => o.WithAllowUntrustedCertificates().WithIgnoreCertificateRevocationErrors())
.Build();
var tcpChannel = new MqttTcpChannel(options);
using var chain = new X509Chain();
var isValid = InvokeCertificateValidationCallback(tcpChannel, chain, SslPolicyErrors.RemoteCertificateChainErrors);
Assert.IsTrue(isValid);
}
[TestMethod]
public async Task Dispose_Channel_While_Used()
{
@@ -95,12 +77,4 @@ public class MqttTcpChannel_Tests
ct.Cancel(false);
}
}
static bool InvokeCertificateValidationCallback(MqttTcpChannel tcpChannel, X509Chain chain, SslPolicyErrors sslPolicyErrors)
{
var method = typeof(MqttTcpChannel).GetMethod("InternalUserCertificateValidationCallback", BindingFlags.Instance | BindingFlags.NonPublic);
Assert.IsNotNull(method);
return (bool)method.Invoke(tcpChannel, [null, null, chain, sslPolicyErrors]);
}
}
}
@@ -48,7 +48,7 @@ public sealed class Injection_Tests : BaseTestClass
[TestMethod]
public async Task Enqueue_Application_Message_At_Session_Level_QueueOverflow_DropNewMessageStrategy()
{
using var testEnvironment = CreateTestEnvironment();
using var testEnvironment = CreateTestEnvironment(trackUnobservedTaskException: false);
var server = await testEnvironment.StartServer(
builder => builder
@@ -126,7 +126,7 @@ public sealed class Injection_Tests : BaseTestClass
[TestMethod]
public async Task Enqueue_Application_Message_At_Session_Level_QueueOverflow_DropOldestQueuedMessageStrategy()
{
using var testEnvironment = CreateTestEnvironment();
using var testEnvironment = CreateTestEnvironment(trackUnobservedTaskException: false);
var server = await testEnvironment.StartServer(
builder => builder
@@ -322,7 +322,7 @@ public sealed class Injection_Tests : BaseTestClass
[TestMethod]
public async Task Deliver_Application_Message_At_Session_Level_QueueOverflow_DropOldestQueuedMessageStrategy()
{
using var testEnvironment = CreateTestEnvironment();
using var testEnvironment = CreateTestEnvironment(trackUnobservedTaskException: false);
var server = await testEnvironment.StartServer(
builder => builder
@@ -1,27 +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;
namespace MQTTnet.Channel;
/// <summary>
/// Optional marker interface that a <see cref="Stream"/> returned by an
/// <see cref="IMqttClientStreamProvider"/> may implement to expose the local and remote
/// socket endpoints to the MQTT TCP channel (used for diagnostics only).
/// </summary>
public interface IMqttClientStreamEndPoints
{
/// <summary>
/// Gets the local endpoint of the socket actually opened by the provider
/// (e.g. the local interface used to reach the proxy). May be <c>null</c> if not known.
/// </summary>
EndPoint LocalEndPoint { get; }
/// <summary>
/// Gets the remote endpoint of the socket actually opened by the provider
/// (e.g. the proxy address). May be <c>null</c> if not known.
/// </summary>
EndPoint RemoteEndPoint { get; }
}
@@ -1,50 +0,0 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Net;
namespace MQTTnet.Channel;
/// <summary>
/// Produces a connected byte <see cref="Stream"/> that carries traffic to an MQTT broker endpoint.
/// This is the extension point used to plug stream-level proxies (SOCKS5, HTTP CONNECT, etc.)
/// into the TCP transport without modifying <c>MqttTcpChannel</c>.
/// Implementations are responsible for the full handshake to the broker (proxy negotiation,
/// authentication, target connect) but MUST NOT perform any TLS wrapping; TLS is layered on
/// top of the returned stream by the channel itself, so the broker host can be used as SNI.
/// </summary>
/// <remarks>
/// <para>
/// The returned <see cref="Stream"/> is owned by the caller (the MQTT channel). Disposing
/// it will release any underlying socket. Providers should therefore avoid keeping their
/// own references to the underlying socket after returning.
/// </para>
/// <para>
/// Implementations may optionally have the returned stream implement
/// <see cref="IMqttClientStreamEndPoints"/> to expose the local and remote socket
/// endpoints to <c>MqttTcpChannel</c> for diagnostics.
/// </para>
/// <para>
/// <see cref="ConnectAsync"/> is invoked on every (re)connect attempt; implementations
/// must be safe to call concurrently or sequentially across multiple client instances.
/// </para>
/// </remarks>
public interface IMqttClientStreamProvider
{
/// <summary>
/// Establishes a connection to the MQTT broker <paramref name="brokerEndPoint"/>, performing
/// any required proxy handshake, and returns the resulting raw byte stream.
/// </summary>
/// <param name="brokerEndPoint">
/// The MQTT broker endpoint. Can be a <see cref="DnsEndPoint"/> (in which case the
/// provider may forward the hostname for remote DNS resolution) or an
/// <see cref="IPEndPoint"/>.
/// </param>
/// <param name="cancellationToken">Token to observe for cancellation.</param>
/// <returns>
/// A connected <see cref="Stream"/> whose bytes are transported end-to-end to
/// <paramref name="brokerEndPoint"/>.
/// </returns>
Task<Stream> ConnectAsync(EndPoint brokerEndPoint, CancellationToken cancellationToken);
}
@@ -1,52 +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.Exceptions;
/// <summary>
/// Categorized error codes for proxy-level failures surfaced via <see cref="MqttProxyException"/>.
/// The first set of values map 1:1 to SOCKS5 REP codes (RFC 1928 §6); additional values cover
/// non-protocol conditions (connecting to the proxy itself, authentication, protocol violations).
/// </summary>
public enum MqttProxyErrorCode
{
/// <summary>The error category could not be classified.</summary>
ProxyUnknownError = 0,
/// <summary>The TCP connection to the proxy could not be established (DNS, refused, timeout).</summary>
ProxyUnreachable,
/// <summary>The proxy returned a response that does not conform to the expected protocol.</summary>
ProxyProtocolError,
/// <summary>The proxy rejected every authentication method offered by the client (SOCKS5 method = 0xFF).</summary>
ProxyAuthMethodRejected,
/// <summary>The proxy rejected the supplied credentials.</summary>
ProxyAuthFailed,
/// <summary>SOCKS5 REP = 0x01: general SOCKS server failure.</summary>
ProxyGeneralFailure,
/// <summary>SOCKS5 REP = 0x02: connection not allowed by ruleset.</summary>
ProxyConnectionNotAllowed,
/// <summary>SOCKS5 REP = 0x03: network unreachable.</summary>
ProxyNetworkUnreachable,
/// <summary>SOCKS5 REP = 0x04: host unreachable.</summary>
ProxyHostUnreachable,
/// <summary>SOCKS5 REP = 0x05: connection refused by destination.</summary>
ProxyConnectionRefused,
/// <summary>SOCKS5 REP = 0x06: TTL expired.</summary>
ProxyTtlExpired,
/// <summary>SOCKS5 REP = 0x07: command not supported by the proxy.</summary>
ProxyCommandNotSupported,
/// <summary>SOCKS5 REP = 0x08: address type not supported by the proxy.</summary>
ProxyAddressTypeNotSupported
}
@@ -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.Exceptions;
/// <summary>
/// Thrown when an <see cref="MQTTnet.Channel.IMqttClientStreamProvider"/> fails while
/// establishing the stream to the broker (proxy unreachable, authentication failure,
/// destination unreachable, protocol violation, ...).
/// Inherits from <see cref="MqttCommunicationException"/> so existing connect/reconnect
/// handling continues to work; callers that care about the specific reason can branch on
/// <see cref="ErrorCode"/>.
/// </summary>
public sealed class MqttProxyException : MqttCommunicationException
{
public MqttProxyException(MqttProxyErrorCode errorCode, string message, string proxyAddress = null, Exception innerException = null)
: base(BuildMessage(errorCode, message, proxyAddress), innerException)
{
ErrorCode = errorCode;
ProxyAddress = proxyAddress;
}
/// <summary>Gets the categorized error reason.</summary>
public MqttProxyErrorCode ErrorCode { get; }
/// <summary>Gets the proxy address (host:port) associated with the failure, if known.</summary>
public string ProxyAddress { get; }
static string BuildMessage(MqttProxyErrorCode errorCode, string message, string proxyAddress)
{
if (string.IsNullOrEmpty(proxyAddress))
{
return $"Proxy error '{errorCode}': {message}";
}
return $"Proxy '{proxyAddress}' error '{errorCode}': {message}";
}
}
@@ -476,28 +476,8 @@ public sealed class MqttV5PacketDecoder
}
var propertiesReader = new MqttV5PropertiesReader(_bufferReader);
var seenSingletonProperties = 0;
while (propertiesReader.MoveNext())
{
// MQTT 5.0 section 3.3.2.3 permits only User Properties and
// Subscription Identifiers to occur more than once in PUBLISH.
var propertyFlag = propertiesReader.CurrentPropertyId switch
{
MqttPropertyId.PayloadFormatIndicator => 1,
MqttPropertyId.MessageExpiryInterval => 2,
MqttPropertyId.ContentType => 4,
MqttPropertyId.ResponseTopic => 8,
MqttPropertyId.CorrelationData => 16,
MqttPropertyId.TopicAlias => 32,
_ => 0
};
if ((seenSingletonProperties & propertyFlag) != 0)
{
throw new MqttProtocolViolationException($"Property '{propertiesReader.CurrentPropertyId}' must not occur more than once in a PUBLISH packet.");
}
seenSingletonProperties |= propertyFlag;
if (propertiesReader.CurrentPropertyId == MqttPropertyId.PayloadFormatIndicator)
{
packet.PayloadFormatIndicator = propertiesReader.ReadPayloadFormatIndicator();
@@ -673,17 +653,7 @@ public sealed class MqttV5PacketDecoder
{
if (propertiesReader.CurrentPropertyId == MqttPropertyId.SubscriptionIdentifier)
{
// MQTT 5.0 section 3.8.2.1.2 forbids zero and repeated identifiers.
if (packet.SubscriptionIdentifier != 0)
{
throw new MqttProtocolViolationException("Subscription Identifier must not occur more than once in a SUBSCRIBE packet.");
}
packet.SubscriptionIdentifier = propertiesReader.ReadSubscriptionIdentifier();
if (packet.SubscriptionIdentifier == 0)
{
throw new MqttProtocolViolationException("Subscription Identifier must not be zero.");
}
}
else
{
-15
View File
@@ -2,21 +2,6 @@ using MQTTnet.Diagnostics.PacketInspection;
namespace MQTTnet;
/// <summary>
/// Represents an MQTT client.
/// <para>
/// Instances of <see cref="IMqttClient" /> are safe for concurrent use: <see cref="PublishAsync" />,
/// <see cref="SubscribeAsync" />, <see cref="UnsubscribeAsync" /> and <see cref="PingAsync" /> may be
/// invoked from multiple threads at the same time. Outgoing packets are serialized internally so
/// that they reach the network in well-formed order, and incoming acknowledgements are routed back
/// to the corresponding caller via the packet identifier.
/// </para>
/// <para>
/// <see cref="ConnectAsync" /> and <see cref="DisconnectAsync" /> must not be called concurrently
/// with each other or with publish/subscribe operations. Coordinate connection lifecycle changes on
/// a single thread.
/// </para>
/// </summary>
public interface IMqttClient : IDisposable
{
event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync;
+57 -110
View File
@@ -55,12 +55,6 @@ public sealed class MqttTcpChannel : IMqttChannel
public async Task ConnectAsync(CancellationToken cancellationToken)
{
if (_tcpOptions.StreamProvider != null)
{
await ConnectViaStreamProviderAsync(cancellationToken).ConfigureAwait(false);
return;
}
CrossPlatformSocket socket = null;
try
{
@@ -109,7 +103,55 @@ public sealed class MqttTcpChannel : IMqttChannel
if (_tcpOptions.TlsOptions?.UseTls == true)
{
_stream = await WrapWithTlsAsync(networkStream, cancellationToken).ConfigureAwait(false);
var targetHost = _tcpOptions.TlsOptions.TargetHost;
if (string.IsNullOrEmpty(targetHost))
{
if (_tcpOptions.RemoteEndpoint is DnsEndPoint dns)
{
targetHost = dns.Host;
}
}
SslStream sslStream;
if (_tcpOptions.TlsOptions.CertificateSelectionHandler != null)
{
sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback, InternalUserCertificateSelectionCallback);
}
else
{
// Use a different constructor depending on the options for MQTTnet so that we do not have
// to copy the exact same behavior of the selection handler.
sslStream = new SslStream(networkStream, false, InternalUserCertificateValidationCallback);
}
try
{
var sslOptions = CreateSslAuthenticationOptions();
sslOptions.TargetHost = targetHost;
if (_tcpOptions.TlsOptions.TrustChain?.Count > 0)
{
sslOptions.CertificateChainPolicy = new X509ChainPolicy
{
TrustMode = X509ChainTrustMode.CustomRootTrust,
VerificationFlags = X509VerificationFlags.IgnoreEndRevocationUnknown,
RevocationMode = _tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.NoCheck : _tcpOptions.TlsOptions.RevocationMode
};
sslOptions.CertificateChainPolicy.CustomTrustStore.AddRange(_tcpOptions.TlsOptions.TrustChain);
}
await sslStream.AuthenticateAsClientAsync(sslOptions, cancellationToken).ConfigureAwait(false);
}
catch
{
await sslStream.DisposeAsync().ConfigureAwait(false);
throw;
}
_stream = sslStream;
}
else
{
@@ -126,106 +168,6 @@ public sealed class MqttTcpChannel : IMqttChannel
}
}
async Task ConnectViaStreamProviderAsync(CancellationToken cancellationToken)
{
Stream providerStream = null;
try
{
providerStream = await _tcpOptions.StreamProvider.ConnectAsync(_tcpOptions.RemoteEndpoint, cancellationToken).ConfigureAwait(false);
if (providerStream == null)
{
throw new MqttCommunicationException("The configured stream provider returned a null stream.");
}
cancellationToken.ThrowIfCancellationRequested();
if (_tcpOptions.TlsOptions?.UseTls == true)
{
_stream = await WrapWithTlsAsync(providerStream, cancellationToken).ConfigureAwait(false);
}
else
{
_stream = providerStream;
}
// Endpoints are diagnostic only. Prefer the provider's reported values if available;
// fall back to the broker endpoint as the "remote" so logs/diagnostics still show
// something meaningful when the provider does not expose its socket endpoints.
if (providerStream is IMqttClientStreamEndPoints connectedStream)
{
LocalEndPoint = connectedStream.LocalEndPoint;
RemoteEndPoint = connectedStream.RemoteEndPoint ?? _tcpOptions.RemoteEndpoint;
}
else
{
LocalEndPoint = null;
RemoteEndPoint = _tcpOptions.RemoteEndpoint;
}
}
catch
{
if (_stream == null)
{
providerStream?.Dispose();
}
throw;
}
}
async Task<SslStream> WrapWithTlsAsync(Stream baseStream, CancellationToken cancellationToken)
{
var targetHost = _tcpOptions.TlsOptions.TargetHost;
if (string.IsNullOrEmpty(targetHost))
{
if (_tcpOptions.RemoteEndpoint is DnsEndPoint dns)
{
targetHost = dns.Host;
}
}
SslStream sslStream;
if (_tcpOptions.TlsOptions.CertificateSelectionHandler != null)
{
sslStream = new SslStream(baseStream, false, InternalUserCertificateValidationCallback, InternalUserCertificateSelectionCallback);
}
else
{
// Use a different constructor depending on the options for MQTTnet so that we do not have
// to copy the exact same behavior of the selection handler.
sslStream = new SslStream(baseStream, false, InternalUserCertificateValidationCallback);
}
try
{
var sslOptions = CreateSslAuthenticationOptions();
sslOptions.TargetHost = targetHost;
if (_tcpOptions.TlsOptions.TrustChain?.Count > 0)
{
sslOptions.CertificateChainPolicy = new X509ChainPolicy
{
TrustMode = X509ChainTrustMode.CustomRootTrust,
VerificationFlags = X509VerificationFlags.IgnoreEndRevocationUnknown,
RevocationMode = _tcpOptions.TlsOptions.IgnoreCertificateRevocationErrors ? X509RevocationMode.NoCheck : _tcpOptions.TlsOptions.RevocationMode
};
sslOptions.CertificateChainPolicy.CustomTrustStore.AddRange(_tcpOptions.TlsOptions.TrustChain);
}
await sslStream.AuthenticateAsClientAsync(sslOptions, cancellationToken).ConfigureAwait(false);
}
catch
{
await sslStream.DisposeAsync().ConfigureAwait(false);
throw;
}
return sslStream;
}
public Task DisconnectAsync(CancellationToken cancellationToken)
{
Dispose();
@@ -362,18 +304,23 @@ public sealed class MqttTcpChannel : IMqttChannel
bool InternalUserCertificateValidationCallback(object sender, X509Certificate x509Certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors)
{
var eventArgs = new MqttClientCertificateValidationEventArgs(x509Certificate, chain, sslPolicyErrors, _tcpOptions);
var certificateValidationHandler = _tcpOptions?.TlsOptions?.CertificateValidationHandler;
if (certificateValidationHandler != null)
{
var eventArgs = new MqttClientCertificateValidationEventArgs(x509Certificate, chain, sslPolicyErrors, _tcpOptions);
return certificateValidationHandler(eventArgs);
}
return MqttClientDefaultCertificateValidationHandler.Handle(eventArgs);
if (_tcpOptions?.TlsOptions?.IgnoreCertificateChainErrors ?? false)
{
sslPolicyErrors &= ~SslPolicyErrors.RemoteCertificateChainErrors;
}
return sslPolicyErrors == SslPolicyErrors.None;
}
X509CertificateCollection LoadCertificates()
{
return _tcpOptions.TlsOptions.ClientCertificatesProvider?.GetCertificates();
}
}
}
@@ -33,10 +33,6 @@ public sealed class MqttPacketBusItem
public void Fail(Exception exception)
{
_promise.TrySetException(exception);
// Queued packets do not always have a waiter (for example, normal server dispatch).
// Observe the failure here while preserving it for callers of WaitAsync.
_ = _promise.Task.Exception;
}
public Task<MqttPacket> WaitAsync()
+7 -21
View File
@@ -953,21 +953,6 @@ public sealed class MqttClient : Disposable, IMqttClient
{
case MqttPublishPacket publishPacket:
EnqueueReceivedPublishPacket(publishPacket);
break;
case MqttPubAckPacket _:
case MqttPubCompPacket _:
case MqttSubAckPacket _:
case MqttUnsubAckPacket _:
// Acknowledgement packets are dispatched to the awaiter that issued the
// matching request. If no awaiter is registered (for example because the
// request was cancelled or timed out client-side before the broker replied)
// the late acknowledgement is harmless and must NOT terminate the connection.
// See issues #2078 and #2079.
if (!_packetDispatcher.TryDispatch(packet))
{
_logger.Warning("Received {0} for an unknown packet identifier. Probably the matching request was cancelled or timed out.", packet.GetType().Name);
}
break;
case MqttPubRecPacket pubRecPacket:
await ProcessReceivedPubRecPacket(pubRecPacket, cancellationToken).ConfigureAwait(false);
@@ -1064,12 +1049,13 @@ public sealed class MqttClient : Disposable, IMqttClient
switch (exception)
{
case OperationCanceledException:
if (cancellationToken.IsCancellationRequested)
{
return;
}
// if (cancellationToken.IsCancellationRequested)
// {
// return;
// }
break;
//break;
return;
case MqttCommunicationException:
_logger.Warning(exception, "Communication error while sending/receiving keep alive packets");
break;
@@ -1085,4 +1071,4 @@ public sealed class MqttClient : Disposable, IMqttClient
_logger.Verbose("Stopped sending keep alive packets");
}
}
}
}
@@ -6,7 +6,6 @@ using System;
using System.Net;
using System.Net.Sockets;
using System.Text;
using MQTTnet.Channel;
using MQTTnet.Formatter;
using MQTTnet.Packets;
using MQTTnet.Protocol;
@@ -253,31 +252,6 @@ public sealed class MqttClientOptionsBuilder
return this;
}
/// <summary>
/// Configures an optional <see cref="IMqttClientStreamProvider"/> on the TCP options.
/// The provider is responsible for opening the transport stream to the broker (e.g. through
/// a SOCKS5 proxy). When the connection should be TLS-secured, configure TLS via
/// <c>WithTlsOptions(...)</c>; TLS will be layered on top of the stream returned by the
/// provider, using the broker host as SNI.
/// </summary>
/// <exception cref="InvalidOperationException">
/// Thrown when no TCP transport has been configured yet (call <c>WithTcpServer</c> or
/// <c>WithEndPoint</c> before this method).
/// </exception>
public MqttClientOptionsBuilder WithStreamProvider(IMqttClientStreamProvider provider)
{
ArgumentNullException.ThrowIfNull(provider);
if (_tcpOptions == null)
{
throw new InvalidOperationException(
"A TCP transport must be configured (via WithTcpServer or WithEndPoint) before assigning a stream provider.");
}
_tcpOptions.StreamProvider = provider;
return this;
}
public MqttClientOptionsBuilder WithProtocolVersion(MqttProtocolVersion value)
{
if (value == MqttProtocolVersion.Unknown)
@@ -4,7 +4,6 @@
using System.Net;
using System.Net.Sockets;
using MQTTnet.Channel;
namespace MQTTnet;
@@ -45,15 +44,6 @@ public sealed class MqttClientTcpOptions : IMqttClientChannelOptions
public EndPoint RemoteEndpoint { get; set; }
/// <summary>
/// Gets or sets an optional <see cref="IMqttClientStreamProvider"/> used to obtain the
/// transport stream to the broker. When set, the channel will delegate the TCP-level
/// connect (including any proxy negotiation) to the provider and then optionally wrap the
/// returned stream with TLS, using the broker host as SNI. When <c>null</c>, the channel
/// opens a plain TCP socket directly to <see cref="RemoteEndpoint"/>.
/// </summary>
public IMqttClientStreamProvider StreamProvider { get; set; }
public MqttClientTlsOptions TlsOptions { get; set; } = new();
public override string ToString()
+1 -4
View File
@@ -1,4 +1 @@
* Client: Late acknowledgement packets (PUBACK, PUBCOMP, SUBACK, UNSUBACK) for canceled or timed-out requests no longer trigger a protocol violation and tear down the connection (#2078, #2079)
* Client: Add support for SOCKS5 proxies (#2251, thanks to @koepalex)
* Client: Fixed keep alive ping send timeout disconnect handling (#2253, thanks to @suhashollakc)
* Server: Fixed exposing Topic Alias to clients (#2250, thanks to @suhashollakc)
- Server: Fixed exposinbg Topic Alias to clients (#2250, thanks to @suhashollakc)