Skip to content

Implement E2EE (DAVE) #142

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Quahu merged 3 commits into from
Feb 24, 2026
Merged

Implement E2EE (DAVE) #142

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
357d3ce
Implement E2EE
Quahu Feb 20, 2026
cdbf57e
Fix MemoryStream cast
Quahu Feb 20, 2026
6dd178f
Copilot suggestions
Quahu Feb 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/Disqord.Core/Library/Library.Constants.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,6 @@ public static class Constants

public const int RestApiVersion = 10;

public const int VoiceVersion = 4;
public const int VoiceVersion = 8;
}
}
56 changes: 44 additions & 12 deletions src/Disqord.Voice.Api/Default/DefaultVoiceGateway.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,12 +1,12 @@
using System;
using System.Buffers.Binary;
using System.IO;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Disqord.Serialization.Json;
using Disqord.Voice.Api.Models;
using Disqord.WebSocket;
using Disqord.WebSocket.Default.Discord;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Qommon.Binding;
Expand All @@ -27,7 +27,7 @@ public class DefaultVoiceGateway : IVoiceGateway

public IWebSocketClientFactory WebSocketClientFactory { get; }

private DiscordWebSocket _ws = null!;
private VoiceWebSocket _ws = null!;
private readonly Binder<IVoiceGatewayClient> _binder;

public DefaultVoiceGateway(
Expand All @@ -48,7 +48,7 @@ public void Bind(IVoiceGatewayClient value)
{
_binder.Bind(value);

_ws = new DiscordWebSocket(Logger, WebSocketClientFactory, false);
_ws = new VoiceWebSocket(Logger, WebSocketClientFactory);
}

public ValueTask ConnectAsync(Uri uri, CancellationToken cancellationToken = default)
Expand All @@ -73,22 +73,54 @@ public ValueTask SendAsync(VoiceGatewayPayloadJsonModel payload, CancellationTok
if (json.Length > 4096)
throw new ArgumentException("Voice cannot send payloads longer than 4096 bytes.", nameof(payload));

return _ws.SendAsync(json, cancellationToken);
return _ws.SendAsync(json, WebSocketMessageType.Text, cancellationToken);
}

public async ValueTask<VoiceGatewayPayloadJsonModel> ReceiveAsync(CancellationToken cancellationToken = default)
public ValueTask SendBinaryAsync(ReadOnlyMemory<byte> data, CancellationToken cancellationToken = default)
{
var jsonStream = await _ws.ReceiveAsync(cancellationToken).ConfigureAwait(false);
if (LogsPayloads && data.Length > 0)
Logger.LogTrace("Voice sending binary payload: op {0}, {1} bytes", data.Span[0], data.Length);

return _ws.SendAsync(data, WebSocketMessageType.Binary, cancellationToken);
}

public async ValueTask<VoiceGatewayMessage> ReceiveAsync(CancellationToken cancellationToken = default)
{
var (stream, isBinary) = await _ws.ReceiveAsync(cancellationToken).ConfigureAwait(false);
if (isBinary)
{
return ParseBinaryMessage(stream);
}

if (LogsPayloads)
{
var stream = new MemoryStream();
await jsonStream.CopyToAsync(stream, cancellationToken).ConfigureAwait(false);
Logger.LogTrace("Voice received payload: {0}", Encoding.UTF8.GetString(stream.GetBuffer()));
stream.Position = 0;
jsonStream = stream;
var memoryStream = new MemoryStream();
await stream.CopyToAsync(memoryStream, cancellationToken).ConfigureAwait(false);
Logger.LogTrace("Voice received payload: {0}", Encoding.UTF8.GetString(memoryStream.GetBuffer()));
memoryStream.Position = 0;
stream = memoryStream;
}

return Serializer.Deserialize<VoiceGatewayPayloadJsonModel>(jsonStream)!;
var payload = Serializer.Deserialize<VoiceGatewayPayloadJsonModel>(stream)!;
return VoiceGatewayMessage.FromJson(payload);
}

private static VoiceGatewayMessage ParseBinaryMessage(MemoryStream stream)
{
// Binary format (server → client): [2-byte big-endian seq] [1-byte opcode] [payload]
stream.TryGetBuffer(out var buffer);
var span = buffer.AsSpan();

if (span.Length < 3)
{
throw new InvalidDataException("Binary voice gateway message is too short.");
}

var sequenceNumber = (int) BinaryPrimitives.ReadUInt16BigEndian(span);
var op = (VoiceGatewayPayloadOperation) span[2];
var payload = buffer.AsMemory(3);

return VoiceGatewayMessage.FromBinary(op, sequenceNumber, payload);
}

public ValueTask DisposeAsync()
Expand Down
159 changes: 145 additions & 14 deletions src/Disqord.Voice.Api/Default/DefaultVoiceGatewayClient.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Disqord.Serialization.Json;
Expand Down Expand Up @@ -32,23 +33,39 @@ public class DefaultVoiceGatewayClient : IVoiceGatewayClient

public CancellationToken StoppingToken { get; private set; }

/// <inheritdoc/>
public IReadOnlySet<Snowflake> ConnectedUserIds => _connectedUserIds;

/// <inheritdoc/>
public int LastSequenceNumber { get; private set; }

/// <inheritdoc/>
public int MaxDaveProtocolVersion { get; }

private Func<VoiceGatewayMessage, CancellationToken, Task>? _daveMessageHandler;

private readonly object _stateLock = new();
private Tcs<ReadyJsonModel> _readyTcs;
private Tcs<SessionDescriptionJsonModel> _sessionDescriptionTcs;
private readonly Tcs<ReadyJsonModel> _readyTcs;
private readonly Tcs<SessionDescriptionJsonModel> _sessionDescriptionTcs;
private Tcs? _postSessionDescriptionTcs;

private readonly HashSet<Snowflake> _connectedUserIds = [];

public DefaultVoiceGatewayClient(
Snowflake guildId,
Snowflake currentMemberId,
string sessionId,
string token,
string endpoint,
int maxDaveProtocolVersion,
ILogger logger,
IVoiceGatewayHeartbeater heartbeater,
IVoiceGateway gateway,
IJsonSerializer serializer)
{
GuildId = guildId;
CurrentMemberId = currentMemberId;
MaxDaveProtocolVersion = maxDaveProtocolVersion;
Logger = logger;
Heartbeater = heartbeater;
Heartbeater.Bind(this);
Expand Down Expand Up @@ -93,7 +110,7 @@ public async Task SendAsync(VoiceGatewayPayloadJsonModel payload, CancellationTo
{
Guard.IsNotNull(payload);

var sent = false;
bool sent;
do
{
try
Expand All @@ -105,20 +122,55 @@ public async Task SendAsync(VoiceGatewayPayloadJsonModel payload, CancellationTo
catch (Exception ex)
{
if (ex is not OperationCanceledException)
{
Logger.LogError(ex, "An exception occurred while sending voice payload: {0}.", payload.Op);
}

throw;
}
}
while (!sent);
}

/// <inheritdoc/>
public async Task SendBinaryAsync(ReadOnlyMemory<byte> data, CancellationToken cancellationToken = default)
{
try
{
if (data.Length > 0)
Logger.LogTrace("Sending binary voice payload: op {0}.", (VoiceGatewayPayloadOperation) data.Span[0]);

await Gateway.SendBinaryAsync(data, cancellationToken).ConfigureAwait(false);
}
catch (Exception ex)
{
if (ex is not OperationCanceledException)
Logger.LogError(ex, "An exception occurred while sending binary voice payload.");

throw;
}
}

public Task RunAsync(CancellationToken stoppingToken)
{
StoppingToken = stoppingToken;
return InternalRunAsync(stoppingToken);
}

/// <inheritdoc/>
public void SuspendAfterSessionDescription()
{
_postSessionDescriptionTcs = new Tcs();
}

/// <inheritdoc/>
public void ResumeAfterSessionDescription(Func<VoiceGatewayMessage, CancellationToken, Task>? daveMessageHandler)
{
_daveMessageHandler = daveMessageHandler;
_postSessionDescriptionTcs?.Complete();
_postSessionDescriptionTcs = null;
}

private async Task InternalConnectAsync(CancellationToken stoppingToken)
{
var attempt = 0;
Expand Down Expand Up @@ -168,15 +220,20 @@ private async Task InternalRunAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
var payload = await Gateway.ReceiveAsync(stoppingToken).ConfigureAwait(false);
Logger.LogTrace("Received voice payload: {0}.", payload.Op);
var message = await Gateway.ReceiveAsync(stoppingToken).ConfigureAwait(false);
Logger.LogTrace("Received voice payload: {0}.", message.Op);

if (message.SequenceNumber != null)
{
LastSequenceNumber = message.SequenceNumber.Value;
}

switch (payload.Op)
switch (message.Op)
{
case VoiceGatewayPayloadOperation.Ready:
{
Logger.LogDebug("Successfully identified. The voice gateway is ready.");
var model = payload.D!.ToType<ReadyJsonModel>()!;
var model = message.JsonPayload!.D!.ToType<ReadyJsonModel>()!;

Logger.LogTrace("SSRC: {Ssrc}, IP: {Ip}:{Port}, Modes: {Modes}.", model.Ssrc, model.Ip, model.Port, model.Modes);

Expand All @@ -195,9 +252,16 @@ private async Task InternalRunAsync(CancellationToken stoppingToken)
}
case VoiceGatewayPayloadOperation.SessionDescription:
{
var model = payload.D!.ToType<SessionDescriptionJsonModel>()!;
Logger.LogDebug("The voice gateway sent a session description with mode {0}.", model.Mode);
var model = message.JsonPayload!.D!.ToType<SessionDescriptionJsonModel>()!;
Logger.LogDebug("The voice gateway sent a session description with mode {0}, DAVE protocol version {1}.", model.Mode, model.DaveProtocolVersion);
_sessionDescriptionTcs.Complete(model);

// Wait for the connection to finish setting up (e.g., DAVE handler) before processing further messages.
if (_postSessionDescriptionTcs != null)
{
await _postSessionDescriptionTcs.Task.ConfigureAwait(false);
}

break;
}
case VoiceGatewayPayloadOperation.Speaking:
Expand All @@ -223,7 +287,7 @@ private async Task InternalRunAsync(CancellationToken stoppingToken)
Logger.LogDebug("The voice gateway said hello.");
try
{
var model = payload.D!.ToType<HelloJsonModel>()!;
var model = message.JsonPayload!.D!.ToType<HelloJsonModel>()!;
var interval = TimeSpan.FromMilliseconds(model.HeartbeatInterval);
await Heartbeater.StartAsync(interval).ConfigureAwait(false);
}
Expand Down Expand Up @@ -253,17 +317,67 @@ private async Task InternalRunAsync(CancellationToken stoppingToken)
}
case VoiceGatewayPayloadOperation.ClientConnect:
{
// TODO: client connect
var model = message.JsonPayload!.D!.ToType<ClientConnectJsonModel>()!;
foreach (var userId in model.UserIds)
{
_connectedUserIds.Add(userId);
}

Logger.LogTrace("Client connect: {0} user(s).", model.UserIds.Length);

if (_daveMessageHandler != null)
{
await _daveMessageHandler(message, stoppingToken).ConfigureAwait(false);
}

break;
}
case VoiceGatewayPayloadOperation.ClientDisconnect:
{
// TODO: client disconnect
var model = message.JsonPayload!.D!.ToType<ClientDisconnectJsonModel>()!;
_connectedUserIds.Remove(model.UserId);
Logger.LogTrace("Client disconnect: {0}.", model.UserId);

if (_daveMessageHandler != null)
await _daveMessageHandler(message, stoppingToken).ConfigureAwait(false);

break;
}
case VoiceGatewayPayloadOperation.DaveProtocolPrepareTransition:
case VoiceGatewayPayloadOperation.DaveProtocolExecuteTransition:
case VoiceGatewayPayloadOperation.DaveProtocolPrepareEpoch:
{
Logger.LogTrace("Received DAVE opcode {0}.", message.Op);
if (_daveMessageHandler != null)
{
await _daveMessageHandler(message, stoppingToken).ConfigureAwait(false);
}

break;
}
case VoiceGatewayPayloadOperation.DaveMlsExternalSenderPackage:
case VoiceGatewayPayloadOperation.DaveMlsProposals:
case VoiceGatewayPayloadOperation.DaveMlsAnnounceCommitTransition:
case VoiceGatewayPayloadOperation.DaveMlsWelcome:
{
Logger.LogTrace("Received DAVE binary opcode {0} ({1} bytes).", message.Op, message.BinaryPayload.Length);
if (_daveMessageHandler != null)
{
await _daveMessageHandler(message, stoppingToken).ConfigureAwait(false);
}

break;
}
case VoiceGatewayPayloadOperation.MediaSinkWants:
case VoiceGatewayPayloadOperation.ClientFlags:
case VoiceGatewayPayloadOperation.ChannelOptionsUpdate:
{
// Ignored known opcodes
break;
}
default:
{
Logger.LogWarning("Unknown voice gateway operation: {0}.", payload.Op);
Logger.LogWarning("Unknown voice gateway operation: {0}.", message.Op);
break;
}
}
Expand All @@ -280,6 +394,14 @@ private async Task InternalRunAsync(CancellationToken stoppingToken)
}
else
{
if (stoppingToken.IsCancellationRequested)
{
// The connection is already being stopped (e.g. due to a VOICE_SERVER_UPDATE).
// The non-recoverable close code is expected — treat it as a cancellation.
Logger.LogDebug("The voice gateway was closed with code {0} while already stopping.", closeCode);
throw new OperationCanceledException(stoppingToken);
}

var level = closeCode == VoiceGatewayCloseCode.ForciblyDisconnected
? LogLevel.Information
: LogLevel.Warning;
Expand All @@ -303,6 +425,10 @@ private async Task InternalRunAsync(CancellationToken stoppingToken)
{
await Gateway.CloseAsync(1000, null, default).ConfigureAwait(false);
}
catch (ObjectDisposedException)
{
// Expected during shutdown — the WebSocket may already be disposed.
}
catch (Exception exc)
{
Logger.LogError(exc, "An exception occurred while closing the voice gateway connection after cancellation.");
Expand All @@ -319,6 +445,10 @@ private async Task InternalRunAsync(CancellationToken stoppingToken)
{
await Gateway.CloseAsync(1000, null, default).ConfigureAwait(false);
}
catch (ObjectDisposedException)
{
// Expected during shutdown — the WebSocket may already be disposed.
}
catch (Exception exc)
{
Logger.LogError(exc, "An exception occurred while closing the voice gateway connection after cancellation.");
Expand Down Expand Up @@ -353,7 +483,8 @@ private Task IdentifyAsync(CancellationToken cancellationToken)
ServerId = GuildId,
UserId = CurrentMemberId,
SessionId = SessionId,
Token = Token
Token = Token,
MaxDaveProtocolVersion = MaxDaveProtocolVersion
}
}, cancellationToken);
}
Expand Down
Loading
Loading