787 lines
30 KiB
C#
787 lines
30 KiB
C#
using System.Buffers;
|
|
using System.Net;
|
|
using System.Net.Sockets;
|
|
using System.Runtime.InteropServices;
|
|
using System.Diagnostics;
|
|
using System.Security.Cryptography;
|
|
using System.Security.Cryptography.X509Certificates;
|
|
using Fleet.Core;
|
|
|
|
namespace MoShiro.Measurement;
|
|
|
|
internal delegate void MsQuicDatagramCallback(MsQuicConnection connection, ReadOnlySpan<byte> datagram);
|
|
|
|
/// <summary>A single MsQuic connection with connection-bound datagram delivery.</summary>
|
|
internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
|
|
{
|
|
private const uint DatagramReceiveEnabled = 0x0500000D;
|
|
private const uint ClientCredentialFlags = 0x00000001 | 0x00000004 | 0x00000010 | 0x00004000;
|
|
private const uint CertificateCredentialNone = 0;
|
|
private const uint CertificateCredentialContext = 3;
|
|
private const uint CertificateCredentialPkcs12 = 6;
|
|
private const uint ConnectionEventConnected = 0;
|
|
private const uint ConnectionEventShutdownByTransport = 1;
|
|
private const uint ConnectionEventShutdownByPeer = 2;
|
|
private const uint ConnectionEventShutdownComplete = 3;
|
|
private const uint ConnectionEventPeerAddressChanged = 5;
|
|
private const uint ConnectionEventDatagramStateChanged = 10;
|
|
private const uint ConnectionEventDatagramReceived = 11;
|
|
private const uint ConnectionEventDatagramSendStateChanged = 12;
|
|
private const uint ConnectionEventPeerCertificateReceived = 15;
|
|
private const int DatagramStateFinal = 3;
|
|
private const int QuicAddressBytes = 28;
|
|
private static readonly MsQuicApi.ConnectionCallback s_callback = OnEvent;
|
|
|
|
private readonly object _sendGate = new();
|
|
private readonly object _nativeGate = new();
|
|
private readonly TaskCompletionSource _connected = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
private readonly TaskCompletionSource _shutdown = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
private readonly GCHandle _self;
|
|
private readonly byte[]? _expectedPeerSpki;
|
|
private readonly Action<MsQuicConnection>? _closed;
|
|
private readonly Action<IPEndPoint>? _peerAddressChanged;
|
|
private readonly ContextNodeLog _log;
|
|
private readonly Dictionary<nint, SendState> _sends = [];
|
|
private nint _registration;
|
|
private nint _configuration;
|
|
private nint _handle;
|
|
private int _maxDatagramBytes;
|
|
private int _sendEnabled;
|
|
private int _disposeStarted;
|
|
private double? _handshakeDurationMs;
|
|
private QuicTransportSnapshot? _transportSnapshot;
|
|
private long _statisticsAt;
|
|
private NativeCounters? _nativeCounters;
|
|
private readonly long _createdAt = Stopwatch.GetTimestamp();
|
|
private int _localPort;
|
|
private int _receivedDatagram;
|
|
private long _sendRejected;
|
|
private long _sendRejectedAt;
|
|
|
|
private sealed record NativeCounters(ulong Sent, ulong Received, long At);
|
|
|
|
private MsQuicConnection(
|
|
nint handle,
|
|
nint registration,
|
|
nint configuration,
|
|
byte[]? expectedPeerSpki,
|
|
MsQuicDatagramCallback? datagramHandler,
|
|
INodeLog log,
|
|
Action<MsQuicConnection>? closed,
|
|
Action<IPEndPoint>? peerAddressChanged)
|
|
{
|
|
_handle = handle;
|
|
_registration = registration;
|
|
_configuration = configuration;
|
|
_expectedPeerSpki = expectedPeerSpki;
|
|
DatagramHandler = datagramHandler;
|
|
_log = new ContextNodeLog(log, ("quic_id", DiagnosticId), ("side", expectedPeerSpki is null ? "listener" : "dialer"));
|
|
_closed = closed;
|
|
_peerAddressChanged = peerAddressChanged;
|
|
_self = GCHandle.Alloc(this);
|
|
}
|
|
|
|
public MsQuicDatagramCallback? DatagramHandler { get; set; }
|
|
|
|
internal string DiagnosticId { get; } = Guid.NewGuid().ToString("N");
|
|
|
|
public IPEndPoint? RemoteEndPoint { get; private set; }
|
|
|
|
public IPEndPoint? LocalEndPoint => Volatile.Read(ref _handle) is nint handle && handle != 0
|
|
? ReadEndpoint(MsQuicApi.GetAddress(handle, 0x05000001))
|
|
: null;
|
|
|
|
public nint Handle => Volatile.Read(ref _handle);
|
|
|
|
public IPEndPoint? RefreshRemoteEndPoint()
|
|
{
|
|
if (RemoteEndPoint is not null || Handle == 0)
|
|
{
|
|
return RemoteEndPoint;
|
|
}
|
|
|
|
try
|
|
{
|
|
RemoteEndPoint = ReadEndpoint(MsQuicApi.GetAddress(Handle, 0x05000002));
|
|
}
|
|
catch (InvalidOperationException)
|
|
{
|
|
// Address isn't available until the handshake finishes.
|
|
}
|
|
|
|
return RemoteEndPoint;
|
|
}
|
|
|
|
public bool IsReady => Volatile.Read(ref _handle) != 0 && _connected.Task.IsCompletedSuccessfully && Volatile.Read(ref _sendEnabled) != 0;
|
|
|
|
public int MaxDatagramBytes => Volatile.Read(ref _maxDatagramBytes);
|
|
|
|
public double? HandshakeDurationMs => _handshakeDurationMs;
|
|
|
|
public QuicTransportSnapshot? TransportSnapshot => Volatile.Read(ref _transportSnapshot);
|
|
|
|
// Called only from this connection's serialized native callbacks. GetParam is
|
|
// inline on the MsQuic worker; calling it from a heartbeat can wait forever
|
|
// for a stalled worker and take the entire Fleet heartbeat down with it.
|
|
private void UpdateTransportSnapshot()
|
|
{
|
|
nint handle = Volatile.Read(ref _handle);
|
|
long now = Stopwatch.GetTimestamp();
|
|
if (handle == 0 || (_statisticsAt != 0 && Stopwatch.GetElapsedTime(_statisticsAt, now) < TimeSpan.FromSeconds(1)))
|
|
{
|
|
return;
|
|
}
|
|
|
|
_statisticsAt = now;
|
|
try
|
|
{
|
|
MsQuicApi.QuicStatisticsV2 stats = MsQuicApi.GetStatisticsV2(handle);
|
|
ulong lost = stats.SendSuspectedLostPackets >= stats.SendSpuriousLostPackets
|
|
? stats.SendSuspectedLostPackets - stats.SendSpuriousLostPackets : 0;
|
|
double? loss = stats.SendTotalPackets > 0 ? lost * 100.0 / stats.SendTotalPackets : null;
|
|
Volatile.Write(ref _transportSnapshot, new QuicTransportSnapshot(loss, stats.SendTotalPackets,
|
|
stats.SendCongestionCount, stats.SendEcnCongestionCount, stats.SendPathMtu));
|
|
Volatile.Write(ref _nativeCounters, new NativeCounters(stats.SendTotalPackets, stats.RecvTotalPackets, now));
|
|
}
|
|
catch (InvalidOperationException)
|
|
{
|
|
// Some event phases do not yet expose statistics. Retain the last snapshot.
|
|
}
|
|
}
|
|
|
|
public static async Task<MsQuicConnection> ConnectAsync(
|
|
IPAddress address,
|
|
int port,
|
|
int localPort,
|
|
byte[] expectedPeerSpki,
|
|
MsQuicDatagramCallback datagramHandler,
|
|
INodeLog log,
|
|
CancellationToken ct,
|
|
QuicSettings? settings = null)
|
|
{
|
|
nint registration = 0;
|
|
nint configuration = 0;
|
|
MsQuicConnection? connection = null;
|
|
long handshakeStarted = 0;
|
|
try
|
|
{
|
|
registration = MsQuicApi.RegistrationOpen("moshiro-dialer");
|
|
configuration = MsQuicApi.ConfigurationOpen(registration, QuicAlpn.Value, settings);
|
|
LoadClientCredentials(configuration);
|
|
var endpoint = new IPEndPoint(address.AddressFamily == AddressFamily.InterNetworkV6 ? IPAddress.IPv6Any : IPAddress.Any, localPort);
|
|
SocketAddress serialized = endpoint.Serialize();
|
|
byte[] localAddress = new byte[QuicAddressBytes];
|
|
for (int i = 0; i < serialized.Size; i++)
|
|
{
|
|
localAddress[i] = serialized[i];
|
|
}
|
|
ushort nativeFamily = ReadAddressFamily(localAddress);
|
|
nint name = Marshal.StringToCoTaskMemUTF8(address.ToString());
|
|
try
|
|
{
|
|
nint handle = MsQuicApi.ConnectionOpen(registration, s_callback, 0);
|
|
connection = new MsQuicConnection(handle, registration, configuration, expectedPeerSpki, datagramHandler, log, null, null);
|
|
connection.RemoteEndPoint = new IPEndPoint(address, port);
|
|
connection._localPort = localPort;
|
|
registration = 0;
|
|
configuration = 0;
|
|
MsQuicApi.SetContext(handle, GCHandle.ToIntPtr(connection._self));
|
|
MsQuicApi.SetParam(handle, DatagramReceiveEnabled, 1);
|
|
MsQuicApi.ConnectionBind(handle, localAddress);
|
|
connection._log.Info("starting MsQuic QUIC handshake", ("phase", "quic_handshake"), ("remote", connection.RemoteEndPoint),
|
|
("local_port", localPort), ("timeout_ms", 10_000));
|
|
handshakeStarted = Stopwatch.GetTimestamp();
|
|
MsQuicApi.ConnectionStart(handle, connection._configuration, nativeFamily, name, checked((ushort)port));
|
|
}
|
|
finally
|
|
{
|
|
Marshal.FreeCoTaskMem(name);
|
|
}
|
|
|
|
await connection._connected.Task.WaitAsync(TimeSpan.FromSeconds(10), ct);
|
|
connection._handshakeDurationMs = Stopwatch.GetElapsedTime(handshakeStarted).TotalMilliseconds;
|
|
return connection;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
if (connection is not null)
|
|
{
|
|
if (!ct.IsCancellationRequested)
|
|
{
|
|
connection.LogState(NodeLogLevel.Warn, "QUIC handshake failed", ("exception", ex.GetType().Name));
|
|
}
|
|
await connection.DisposeAsync();
|
|
}
|
|
|
|
if (configuration != 0)
|
|
{
|
|
MsQuicApi.ConfigurationClose(configuration);
|
|
}
|
|
|
|
if (registration != 0)
|
|
{
|
|
MsQuicApi.RegistrationClose(registration);
|
|
}
|
|
|
|
if (ex is TimeoutException)
|
|
{
|
|
throw new TimeoutException($"QUIC handshake timed out after 10000 ms: remote={address}:{port}, local_udp_port={localPort}", ex);
|
|
}
|
|
throw;
|
|
}
|
|
}
|
|
|
|
public static MsQuicConnection Accept(
|
|
nint handle,
|
|
nint configuration,
|
|
MsQuicDatagramCallback datagramHandler,
|
|
INodeLog log,
|
|
Action<MsQuicConnection> closed,
|
|
Action<IPEndPoint>? peerAddressChanged)
|
|
{
|
|
var connection = new MsQuicConnection(handle, 0, 0, null, datagramHandler, log, closed, peerAddressChanged);
|
|
MsQuicApi.SetContext(handle, GCHandle.ToIntPtr(connection._self));
|
|
MsQuicApi.SetCallbackHandler(handle, s_callback, GCHandle.ToIntPtr(connection._self));
|
|
connection.RefreshRemoteEndPoint();
|
|
connection._log.Info("accepted MsQuic QUIC connection", ("phase", "quic_handshake"), ("remote", connection.RemoteEndPoint));
|
|
MsQuicApi.SetParam(handle, DatagramReceiveEnabled, 1);
|
|
// Transport parameters must be set before SetConfiguration starts TLS.
|
|
MsQuicApi.ConnectionSetConfiguration(handle, configuration);
|
|
return connection;
|
|
}
|
|
|
|
private int _authenticatedDatagram;
|
|
|
|
internal void NoteAuthenticatedDatagram(string? peerId, uint streamId)
|
|
{
|
|
if (Interlocked.Exchange(ref _authenticatedDatagram, 1) == 0)
|
|
{
|
|
_log.Info("first authenticated measurement datagram", ("phase", "measurement"),
|
|
("peer", peerId), ("remote", RemoteEndPoint), ("stream", streamId));
|
|
}
|
|
}
|
|
|
|
public bool Send(ReadOnlySpan<byte> datagram)
|
|
{
|
|
// ConnectionClose must not free the handle or a pending send while
|
|
// its native buffer is still being prepared/submitted.
|
|
lock (_nativeGate)
|
|
{
|
|
return SendCore(datagram);
|
|
}
|
|
}
|
|
|
|
private bool SendCore(ReadOnlySpan<byte> datagram)
|
|
{
|
|
nint handle = Volatile.Read(ref _handle);
|
|
if (handle == 0 || !IsReady || datagram.Length > MaxDatagramBytes)
|
|
{
|
|
LogSendRejected(handle == 0 ? "connection_closed" : !IsReady ? "datagram_not_ready" : "datagram_too_large", datagram.Length, null);
|
|
return false;
|
|
}
|
|
|
|
byte[] buffer = ArrayPool<byte>.Shared.Rent(datagram.Length);
|
|
datagram.CopyTo(buffer);
|
|
GCHandle pinned = GCHandle.Alloc(buffer, GCHandleType.Pinned);
|
|
nint nativeBuffer = Marshal.AllocHGlobal(Marshal.SizeOf<MsQuicApi.QuicBuffer>());
|
|
var state = new SendState(buffer, pinned, nativeBuffer);
|
|
GCHandle stateHandle = GCHandle.Alloc(state);
|
|
nint context = GCHandle.ToIntPtr(stateHandle);
|
|
lock (_sendGate)
|
|
{
|
|
if (_handle == 0)
|
|
{
|
|
stateHandle.Free();
|
|
state.Release();
|
|
return false;
|
|
}
|
|
|
|
_sends.Add(context, state);
|
|
}
|
|
|
|
var nativeBufferValue = new MsQuicApi.QuicBuffer
|
|
{
|
|
Length = (uint)datagram.Length,
|
|
Buffer = pinned.AddrOfPinnedObject(),
|
|
};
|
|
Marshal.StructureToPtr(nativeBufferValue, nativeBuffer, false);
|
|
uint status = MsQuicApi.DatagramSend(handle, nativeBuffer, context);
|
|
if (MsQuicApi.Succeeded(status))
|
|
{
|
|
return true;
|
|
}
|
|
|
|
ReleaseSend(context);
|
|
LogSendRejected("native_send_failed", datagram.Length, status);
|
|
return false;
|
|
}
|
|
|
|
private void LogSendRejected(string reason, int bytes, uint? status)
|
|
{
|
|
long rejected = Interlocked.Increment(ref _sendRejected);
|
|
long now = Stopwatch.GetTimestamp();
|
|
long previous = Interlocked.Read(ref _sendRejectedAt);
|
|
if (previous != 0 && Stopwatch.GetElapsedTime(previous, now) < TimeSpan.FromSeconds(30))
|
|
{
|
|
return;
|
|
}
|
|
if (Interlocked.CompareExchange(ref _sendRejectedAt, now, previous) != previous)
|
|
{
|
|
return;
|
|
}
|
|
|
|
_log.Warn("QUIC datagram submission rejected", ("phase", "quic_data"), ("reason", reason), ("remote", RemoteEndPoint),
|
|
("bytes", bytes), ("max_datagram", MaxDatagramBytes), ("rejected_total", rejected), ("status", status is { } s ? $"0x{s:X8}" : null));
|
|
}
|
|
|
|
// Diagnostics only read managed snapshots. Never query the native worker from
|
|
// timeout, heartbeat, or teardown paths: the worker may itself be blocked.
|
|
private void LogState(NodeLogLevel level, string message, params (string Key, object? Value)[] details)
|
|
{
|
|
NativeCounters? counters = Volatile.Read(ref _nativeCounters);
|
|
_log.Write(level, message, [
|
|
("remote", RemoteEndPoint), ("local_port", _localPort),
|
|
("phase", _connected.Task.IsCompletedSuccessfully ? "quic_data" : "quic_handshake"),
|
|
("elapsed_ms", (long)Stopwatch.GetElapsedTime(_createdAt).TotalMilliseconds),
|
|
("handshake_completed", _connected.Task.IsCompletedSuccessfully),
|
|
("packets_sent", counters?.Sent), ("packets_received", counters?.Received),
|
|
("stats_age_ms", counters is null ? null : (long?)Stopwatch.GetElapsedTime(counters.At).TotalMilliseconds),
|
|
.. details,
|
|
]);
|
|
}
|
|
|
|
internal static string StatusName(uint status) => status == 0 ? "success" : !OperatingSystem.IsLinux() ? "native_error" : status switch
|
|
{
|
|
1 => "invalid_state", 22 => "invalid_parameter", 62 => "connection_idle", 103 => "handshake_failure",
|
|
110 => "connection_timeout", 111 => "connection_refused", 113 => "unreachable", 125 => "aborted", _ => "native_error",
|
|
};
|
|
|
|
public async ValueTask DisposeAsync()
|
|
{
|
|
lock (_nativeGate)
|
|
{
|
|
if (Interlocked.Exchange(ref _disposeStarted, 1) == 0 && Volatile.Read(ref _handle) is nint handle && handle != 0)
|
|
{
|
|
MsQuicApi.ConnectionShutdown(handle);
|
|
}
|
|
}
|
|
|
|
if (Volatile.Read(ref _handle) != 0)
|
|
{
|
|
// 原生层偶尔不回报关闭完成：等一会儿就放手，调用方（重拨循环、有序退出）不能跟着卡住
|
|
try
|
|
{
|
|
await _shutdown.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
}
|
|
catch (TimeoutException)
|
|
{
|
|
_log.Warn("QUIC connection did not report shutdown completion in time; moving on");
|
|
}
|
|
}
|
|
}
|
|
|
|
public Task WaitForShutdownAsync() => _shutdown.Task;
|
|
|
|
private static void LoadClientCredentials(nint configuration)
|
|
{
|
|
var credential = new MsQuicApi.CredentialConfig { Type = CertificateCredentialNone, Flags = ClientCredentialFlags };
|
|
if (OperatingSystem.IsWindows())
|
|
{
|
|
credential.Flags &= ~0x00004000u;
|
|
}
|
|
|
|
MsQuicApi.ConfigurationLoadCredential(configuration, ref credential);
|
|
}
|
|
|
|
internal static void LoadServerCredentials(nint configuration, X509Certificate2 certificate)
|
|
{
|
|
var credential = new MsQuicApi.CredentialConfig();
|
|
GCHandle pfxPin = default;
|
|
nint pkcs12 = 0;
|
|
try
|
|
{
|
|
if (OperatingSystem.IsWindows())
|
|
{
|
|
credential.Type = CertificateCredentialContext;
|
|
credential.Certificate = certificate.Handle;
|
|
}
|
|
else
|
|
{
|
|
byte[] pfxBytes = certificate.Export(X509ContentType.Pkcs12);
|
|
pfxPin = GCHandle.Alloc(pfxBytes, GCHandleType.Pinned);
|
|
var pfxCertificate = new MsQuicApi.Pkcs12Certificate
|
|
{
|
|
Asn1Blob = pfxPin.AddrOfPinnedObject(),
|
|
Asn1BlobLength = (uint)pfxBytes.Length,
|
|
};
|
|
pkcs12 = Marshal.AllocHGlobal(Marshal.SizeOf<MsQuicApi.Pkcs12Certificate>());
|
|
Marshal.StructureToPtr(pfxCertificate, pkcs12, false);
|
|
credential.Type = CertificateCredentialPkcs12;
|
|
credential.Flags = 0x00004000;
|
|
credential.Certificate = pkcs12;
|
|
}
|
|
|
|
MsQuicApi.ConfigurationLoadCredential(configuration, ref credential);
|
|
}
|
|
finally
|
|
{
|
|
if (pkcs12 != 0)
|
|
{
|
|
Marshal.FreeHGlobal(pkcs12);
|
|
}
|
|
|
|
if (pfxPin.IsAllocated)
|
|
{
|
|
pfxPin.Free();
|
|
}
|
|
}
|
|
}
|
|
|
|
private static uint OnEvent(nint handle, nint context, nint eventPointer)
|
|
{
|
|
try
|
|
{
|
|
var connection = (MsQuicConnection)GCHandle.FromIntPtr(context).Target!;
|
|
uint eventType = unchecked((uint)Marshal.ReadInt32(eventPointer));
|
|
switch (eventType)
|
|
{
|
|
case ConnectionEventConnected:
|
|
connection.RefreshRemoteEndPoint();
|
|
try
|
|
{
|
|
connection._localPort = connection.LocalEndPoint?.Port ?? connection._localPort;
|
|
}
|
|
catch (InvalidOperationException)
|
|
{
|
|
// An unavailable diagnostic address must not fail a successful handshake.
|
|
}
|
|
connection._connected.TrySetResult();
|
|
connection.LogState(NodeLogLevel.Info, "MsQuic QUIC handshake completed");
|
|
break;
|
|
case ConnectionEventShutdownByTransport:
|
|
{
|
|
uint status = unchecked((uint)Marshal.ReadInt32(eventPointer, IntPtr.Size));
|
|
ulong errorCode = unchecked((ulong)Marshal.ReadInt64(eventPointer, IntPtr.Size * 2));
|
|
connection.LogState(NodeLogLevel.Error, "MsQuic QUIC transport shutdown", ("status", $"0x{status:X8}"),
|
|
("status_name", StatusName(status)), ("error_code", errorCode));
|
|
connection._connected.TrySetException(new IOException($"QUIC connection closed during handshake (transport status 0x{status:X8}, error code {errorCode})."));
|
|
break;
|
|
}
|
|
case ConnectionEventShutdownByPeer:
|
|
{
|
|
ulong errorCode = unchecked((ulong)Marshal.ReadInt64(eventPointer, IntPtr.Size));
|
|
connection.LogState(NodeLogLevel.Warn, "MsQuic QUIC peer shutdown", ("error_code", errorCode));
|
|
connection._connected.TrySetException(new IOException($"QUIC connection closed during handshake by peer (error code {errorCode})."));
|
|
break;
|
|
}
|
|
case ConnectionEventShutdownComplete:
|
|
connection.OnShutdownComplete(handle, eventPointer);
|
|
break;
|
|
case ConnectionEventPeerAddressChanged:
|
|
{
|
|
nint address = Marshal.ReadIntPtr(eventPointer, IntPtr.Size);
|
|
IPEndPoint? endpoint = ReadEndpoint(address);
|
|
if (endpoint is not null)
|
|
{
|
|
connection._log.Info("QUIC peer address changed", ("previous", connection.RemoteEndPoint), ("remote", endpoint));
|
|
connection.RemoteEndPoint = endpoint;
|
|
connection._peerAddressChanged?.Invoke(endpoint);
|
|
}
|
|
|
|
break;
|
|
}
|
|
case ConnectionEventDatagramStateChanged:
|
|
connection.OnDatagramStateChanged(eventPointer);
|
|
break;
|
|
case ConnectionEventDatagramReceived:
|
|
connection.OnDatagramReceived(eventPointer);
|
|
break;
|
|
case ConnectionEventDatagramSendStateChanged:
|
|
connection.OnDatagramSendStateChanged(eventPointer);
|
|
break;
|
|
case ConnectionEventPeerCertificateReceived:
|
|
return connection.ValidatePeerCertificate(eventPointer);
|
|
}
|
|
|
|
connection.UpdateTransportSnapshot();
|
|
return 0;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
try
|
|
{
|
|
if (context != 0 && GCHandle.FromIntPtr(context).Target is MsQuicConnection connection)
|
|
{
|
|
connection._log.Error("MsQuic connection callback failed", ("err", ex.Message));
|
|
}
|
|
}
|
|
catch
|
|
{
|
|
// Keep exceptions inside the native callback boundary.
|
|
}
|
|
|
|
return OperatingSystem.IsWindows() ? 0x80004005u : 1u;
|
|
}
|
|
}
|
|
|
|
private void OnDatagramStateChanged(nint eventPointer)
|
|
{
|
|
(bool sendEnabled, int maxSendLength) = ReadDatagramState(eventPointer);
|
|
int previousMax = Interlocked.Exchange(ref _maxDatagramBytes, maxSendLength);
|
|
int previousEnabled = Interlocked.Exchange(ref _sendEnabled, sendEnabled ? 1 : 0);
|
|
if (previousMax != maxSendLength || previousEnabled != (sendEnabled ? 1 : 0))
|
|
{
|
|
_log.Info("QUIC datagram negotiation changed", ("phase", "datagram_negotiation"), ("remote", RemoteEndPoint),
|
|
("send_enabled", sendEnabled), ("max_datagram", maxSendLength));
|
|
}
|
|
}
|
|
|
|
internal static (bool SendEnabled, int MaxSendLength) ReadDatagramState(nint eventPointer)
|
|
{
|
|
// BOOLEAN is followed by padding before the naturally aligned uint16_t.
|
|
var state = Marshal.PtrToStructure<DatagramState>(eventPointer + IntPtr.Size);
|
|
return (state.SendEnabled != 0, state.MaxSendLength);
|
|
}
|
|
|
|
[StructLayout(LayoutKind.Sequential)]
|
|
private struct DatagramState
|
|
{
|
|
public byte SendEnabled;
|
|
public ushort MaxSendLength;
|
|
}
|
|
|
|
private void OnDatagramReceived(nint eventPointer)
|
|
{
|
|
nint bufferPointer = Marshal.ReadIntPtr(eventPointer, IntPtr.Size);
|
|
if (bufferPointer == 0)
|
|
{
|
|
return;
|
|
}
|
|
|
|
var buffer = Marshal.PtrToStructure<MsQuicApi.QuicBuffer>(bufferPointer);
|
|
if (buffer.Buffer != 0 && buffer.Length > 0 && buffer.Length <= 65535)
|
|
{
|
|
if (Interlocked.Exchange(ref _receivedDatagram, 1) == 0)
|
|
{
|
|
_log.Info("first QUIC datagram received", ("phase", "quic_data"), ("remote", RemoteEndPoint), ("bytes", buffer.Length));
|
|
}
|
|
unsafe
|
|
{
|
|
DatagramHandler?.Invoke(this, new ReadOnlySpan<byte>((void*)buffer.Buffer, checked((int)buffer.Length)));
|
|
}
|
|
}
|
|
}
|
|
|
|
private void OnDatagramSendStateChanged(nint eventPointer)
|
|
{
|
|
nint context = Marshal.ReadIntPtr(eventPointer, IntPtr.Size);
|
|
int state = Marshal.ReadInt32(eventPointer, IntPtr.Size * 2);
|
|
if (state >= DatagramStateFinal)
|
|
{
|
|
ReleaseSend(context);
|
|
}
|
|
}
|
|
|
|
private uint ValidatePeerCertificate(nint eventPointer)
|
|
{
|
|
nint certificate = Marshal.ReadIntPtr(eventPointer, IntPtr.Size);
|
|
if (certificate == 0 || _expectedPeerSpki is null)
|
|
{
|
|
return 0;
|
|
}
|
|
|
|
byte[] der;
|
|
if (OperatingSystem.IsWindows())
|
|
{
|
|
nint encoded = Marshal.ReadIntPtr(certificate, IntPtr.Size);
|
|
int length = Marshal.ReadInt32(certificate, IntPtr.Size * 2);
|
|
if (encoded == 0 || length <= 0 || length > 64 * 1024)
|
|
{
|
|
return 0x80004005u;
|
|
}
|
|
|
|
der = new byte[length];
|
|
Marshal.Copy(encoded, der, 0, length);
|
|
}
|
|
else
|
|
{
|
|
var nativeBuffer = Marshal.PtrToStructure<MsQuicApi.QuicBuffer>(certificate);
|
|
if (nativeBuffer.Buffer == 0 || nativeBuffer.Length == 0 || nativeBuffer.Length > 64 * 1024)
|
|
{
|
|
return 1;
|
|
}
|
|
|
|
der = new byte[checked((int)nativeBuffer.Length)];
|
|
Marshal.Copy(nativeBuffer.Buffer, der, 0, der.Length);
|
|
}
|
|
|
|
using X509Certificate2 peerCertificate = X509CertificateLoader.LoadCertificate(der);
|
|
byte[] actualSpki = ExportSubjectPublicKeyInfo(peerCertificate);
|
|
bool valid = actualSpki.AsSpan().SequenceEqual(_expectedPeerSpki);
|
|
if (!valid)
|
|
{
|
|
_log.Error("MsQuic peer certificate SPKI mismatch",
|
|
("expected_sha256", Convert.ToHexStringLower(SHA256.HashData(_expectedPeerSpki))),
|
|
("actual_sha256", Convert.ToHexStringLower(SHA256.HashData(actualSpki))));
|
|
}
|
|
return valid ? 0 : OperatingSystem.IsWindows() ? 0x80004005u : 1u;
|
|
}
|
|
|
|
private static byte[] ExportSubjectPublicKeyInfo(X509Certificate2 certificate)
|
|
{
|
|
using ECDsa? ecdsa = certificate.GetECDsaPublicKey();
|
|
if (ecdsa is not null)
|
|
{
|
|
return ecdsa.ExportSubjectPublicKeyInfo();
|
|
}
|
|
|
|
using RSA? rsa = certificate.GetRSAPublicKey();
|
|
if (rsa is not null)
|
|
{
|
|
return rsa.ExportSubjectPublicKeyInfo();
|
|
}
|
|
|
|
using DSA? dsa = certificate.GetDSAPublicKey();
|
|
if (dsa is not null)
|
|
{
|
|
return dsa.ExportSubjectPublicKeyInfo();
|
|
}
|
|
|
|
return [];
|
|
}
|
|
|
|
private void ReleaseSend(nint context)
|
|
{
|
|
SendState? state;
|
|
lock (_sendGate)
|
|
{
|
|
if (!_sends.Remove(context, out state))
|
|
{
|
|
return;
|
|
}
|
|
}
|
|
|
|
if (GCHandle.FromIntPtr(context).IsAllocated)
|
|
{
|
|
GCHandle.FromIntPtr(context).Free();
|
|
}
|
|
|
|
state.Release();
|
|
}
|
|
|
|
private void OnShutdownComplete(nint handle, nint eventPointer)
|
|
{
|
|
UpdateTransportSnapshot();
|
|
LogState(NodeLogLevel.Info, "QUIC connection shutdown completed", ("rejected_sends", Interlocked.Read(ref _sendRejected)));
|
|
if (!_connected.Task.IsCompletedSuccessfully)
|
|
{
|
|
byte flags = Marshal.ReadByte(eventPointer, IntPtr.Size);
|
|
_log.Warn("MsQuic shutdown completed before handshake notification",
|
|
("handshake_completed", (flags & 1) != 0),
|
|
("peer_acknowledged", (flags & 2) != 0));
|
|
}
|
|
|
|
if (Interlocked.Exchange(ref _handle, 0) == 0)
|
|
{
|
|
return;
|
|
}
|
|
|
|
// RegistrationClose waits for native workers/children and is forbidden
|
|
// on a MsQuic callback. Return from the callback before waiting for cleanup.
|
|
ThreadPool.QueueUserWorkItem(_ => CloseHandles(handle));
|
|
}
|
|
|
|
private void CloseHandles(nint handle)
|
|
{
|
|
lock (_nativeGate)
|
|
{
|
|
MsQuicApi.ConnectionClose(handle);
|
|
}
|
|
lock (_sendGate)
|
|
{
|
|
foreach ((nint context, SendState state) in _sends)
|
|
{
|
|
GCHandle.FromIntPtr(context).Free();
|
|
state.Release();
|
|
}
|
|
|
|
_sends.Clear();
|
|
}
|
|
|
|
if (_configuration != 0)
|
|
{
|
|
MsQuicApi.ConfigurationClose(_configuration);
|
|
}
|
|
|
|
if (_registration != 0)
|
|
{
|
|
MsQuicApi.RegistrationClose(_registration);
|
|
}
|
|
|
|
_self.Free();
|
|
_shutdown.TrySetResult();
|
|
_closed?.Invoke(this);
|
|
}
|
|
|
|
private static IPEndPoint? ReadEndpoint(nint address)
|
|
{
|
|
if (address == 0)
|
|
{
|
|
return null;
|
|
}
|
|
|
|
ushort family = unchecked((ushort)Marshal.ReadInt16(address));
|
|
if (family == 2)
|
|
{
|
|
byte[] bytes = new byte[4];
|
|
Marshal.Copy(address + 4, bytes, 0, bytes.Length);
|
|
int port = (Marshal.ReadByte(address, 2) << 8) | Marshal.ReadByte(address, 3);
|
|
return new IPEndPoint(new IPAddress(bytes), port);
|
|
}
|
|
|
|
ushort ipv6Family = OperatingSystem.IsWindows() ? (ushort)23 : OperatingSystem.IsMacOS() ? (ushort)30 : (ushort)10;
|
|
if (family == ipv6Family)
|
|
{
|
|
byte[] bytes = new byte[16];
|
|
Marshal.Copy(address + 8, bytes, 0, bytes.Length);
|
|
int port = (Marshal.ReadByte(address, 2) << 8) | Marshal.ReadByte(address, 3);
|
|
long scope = unchecked((uint)Marshal.ReadInt32(address, 24));
|
|
return new IPEndPoint(new IPAddress(bytes, scope), port);
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
private static IPEndPoint? ReadEndpoint(byte[] address)
|
|
{
|
|
unsafe
|
|
{
|
|
fixed (byte* addressBytes = address)
|
|
{
|
|
return ReadEndpoint((nint)addressBytes);
|
|
}
|
|
}
|
|
}
|
|
|
|
private static ushort ReadAddressFamily(byte[] address) => (ushort)(address[0] | (address[1] << 8));
|
|
|
|
private sealed class SendState(byte[] buffer, GCHandle pin, nint nativeBuffer)
|
|
{
|
|
public void Release()
|
|
{
|
|
if (pin.IsAllocated)
|
|
{
|
|
pin.Free();
|
|
}
|
|
|
|
if (nativeBuffer != 0)
|
|
{
|
|
Marshal.FreeHGlobal(nativeBuffer);
|
|
}
|
|
|
|
ArrayPool<byte>.Shared.Return(buffer);
|
|
}
|
|
}
|
|
|
|
}
|