From 05eacb3092848e08bdbb530366226eefdbbf70fa Mon Sep 17 00:00:00 2001 From: Talon Date: Tue, 15 Sep 2026 22:53:54 +0200 Subject: [PATCH] Add encrypted managed UDP relay and native voice conformance --- .github/workflows/dotnet.yml | 2 +- CLAUDE.md | 7 +- PROGRESS.md | 44 ++- docs/api-dotnet.md | 34 +- docs/porting-to-dotnet.md | 23 +- docs/roadmap.md | 9 +- docs/security.md | 22 +- dotnet/README.md | 26 +- dotnet/oracle/CMakeLists.txt | 4 + dotnet/oracle/voice.cpp | 127 ++++++++ .../VoiceCat.Server/Transport/MediaFanout.cs | 50 +++ .../VoiceCat.Server/Transport/MediaRelay.cs | 150 +++++++++ .../Transport/MediaSessionCrypto.cs | 10 + .../Transport/TlsControlConnection.cs | 20 +- dotnet/src/VoiceCat.Server/VoiceServer.cs | 97 +++++- .../tests/VoiceCat.Tests/MediaFanoutTests.cs | 110 +++++++ .../tests/VoiceCat.Tests/MediaRelayTests.cs | 303 ++++++++++++++++++ dotnet/tests/VoiceCat.Tests/ServerTests.cs | 7 +- tools/vccli/src/main.cpp | 69 +++- 19 files changed, 1068 insertions(+), 46 deletions(-) create mode 100644 dotnet/oracle/voice.cpp create mode 100644 dotnet/src/VoiceCat.Server/Transport/MediaFanout.cs create mode 100644 dotnet/src/VoiceCat.Server/Transport/MediaRelay.cs create mode 100644 dotnet/src/VoiceCat.Server/Transport/MediaSessionCrypto.cs create mode 100644 dotnet/tests/VoiceCat.Tests/MediaFanoutTests.cs create mode 100644 dotnet/tests/VoiceCat.Tests/MediaRelayTests.cs diff --git a/.github/workflows/dotnet.yml b/.github/workflows/dotnet.yml index 8d962bd..bdb4ff6 100644 --- a/.github/workflows/dotnet.yml +++ b/.github/workflows/dotnet.yml @@ -61,4 +61,4 @@ jobs: pwsh -File dotnet/compare-dsp-fixtures.ps1 dotnet/tests/VoiceCat.Tests/Fixtures/cpp-noise.json build/dev/cpp-noise.json pwsh -File dotnet/build-native.ps1 dotnet restore dotnet/VoiceCat.slnx --locked-mode - VOICECAT_TLS_ORACLE="$PWD/build/dev/bin/voicecat-dotnet-tls-oracle" VOICECAT_DATABASE_ORACLE="$PWD/build/dev/bin/voicecat-dotnet-database-oracle" VOICECAT_VCCLI="$PWD/build/dev/bin/vccli" dotnet test dotnet/VoiceCat.slnx -c Release --no-restore + VOICECAT_TLS_ORACLE="$PWD/build/dev/bin/voicecat-dotnet-tls-oracle" VOICECAT_DATABASE_ORACLE="$PWD/build/dev/bin/voicecat-dotnet-database-oracle" VOICECAT_VOICE_ORACLE="$PWD/build/dev/bin/voicecat-dotnet-voice-oracle" VOICECAT_VCCLI="$PWD/build/dev/bin/vccli" dotnet test dotnet/VoiceCat.slnx -c Release --no-restore diff --git a/CLAUDE.md b/CLAUDE.md index 77f7d11..6c08448 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -25,7 +25,7 @@ native clients (Swift on macOS/iOS, C# on Windows) and the server. ## Build & test commands -The .NET rewrite lives under `dotnet/`. Build and test its wire/crypto, TLS, codec/DSP, and initial control server slices +The .NET rewrite lives under `dotnet/`. Build and test its wire/crypto, TLS, codec/DSP, and managed control/UDP server slices alongside the existing C++ tree: ```powershell @@ -36,8 +36,9 @@ dotnet test dotnet/VoiceCat.slnx -c Release --no-build ./dotnet/check-licenses.ps1 ``` -See `dotnet/README.md` for C# conventions and C++ fixture regeneration, and -`docs/api-dotnet.md` for managed interfaces. Subsequent port phases remain planned. +See `dotnet/README.md` for C# conventions and required native voice/CLI conformance, +and `docs/api-dotnet.md` for managed interfaces. Phase 4 remains in progress; +server administration/reaping and audio/client/UI phases are still pending. The default development preset is **`dev`** — it builds everything (server + tools + tests) with real vcpkg deps. The `skeleton` preset (no deps, stubs only) is a fast smoke check; see diff --git a/PROGRESS.md b/PROGRESS.md index 1233dc7..7d18f56 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -10,6 +10,31 @@ up instantly. Newest status at the top. ## ▶ Where we left off / next action +- **Done (2026-09-15): Phase 4 encrypted voice checkpoint.** Existing pending codec/DSP + and initial server work committed as `4067bab`. Managed server now binds UDP on the + TCP port number, issues 16-byte session tokens, supports subscription and multi-stream + signaling, and authenticates/reseals encoded Opus to eligible channel subscribers. + Crypto and endpoints have one UDP-loop owner; control handlers publish immutable + routing snapshots. Rejects invalid tokens, malformed/forged/replayed media and SSRCs + not owned by the sender. Stop, unsubscribe, channel movement and disconnect update + routing; retired keys are cleared without requiring subsequent UDP traffic. + First endpoint binding is fixed for the session (reconnect to change it), unlike + the C++ oracle's permissive rebinding policy. Packet formats/protocol v2 are unchanged. + Two actual C++ `vccli` processes authenticate, join, chat and exchange mono/stereo + voice through the managed server. Native voice oracle additionally verifies three + concurrent streams with bidirectional decoded PCM energy/metadata, without hardware. + Added finite `vccli --test-tone-ms` and fixed normal `--voice` to subscribe first. + **Verified:** 154/154 managed tests, no skips with TLS/database/voice/CLI variables; + native dev build and 29/29 CTest tests; independent native media staging; warning-free + managed Release build; identical regenerated wire/password fixtures; C++ DSP within + one PCM unit; 22 permissive package licenses; `git diff --check` passes. Fan-out core + allocates zero managed bytes for 50 subscribers; transport scheduling and crypto + fallback are excluded. Transport test delivers all 2,500 packets at a paced 50 pps. + **Next:** media-aware keepalive/reaper, then protected channel joins, channel CRUD, + permissions/moderation/account administration and production configuration. Phase 4 + remains in progress. Audio, managed client/CLI, Windows cutover and C# AppKit/UIKit + follow; keep Swift ReplayKit extension and freeze its ring contract before iOS. + - **In progress (2026-09-15): Phase 4 managed server control plane.** Added TLS socket orchestration, bounded framing/queues, guest and password authentication, persisted channels, state snapshots, channel joins, text routing, ping and disconnect events. @@ -26,7 +51,7 @@ up instantly. Newest status at the top. UDP voice, streams, administration, protected channel joins and production configuration remain pending; this is the first control-plane checkpoint, not Phase 4 completion. -### .NET handoff / discoveries (2026-09-15) +### .NET control-plane checkpoint handoff / discoveries (2026-09-15) - **Working tree:** stay on `dotnet/foundations`, tracking `origin/dotnet/foundations`. Foundation `b76181d` and TLS checkpoint `2df79cd` were committed and pushed. @@ -83,7 +108,7 @@ up instantly. Newest status at the top. SourceGear's native package lacks a NuGet license expression; the audit has an exact-version/repository-identity exception for its public-domain SQLite build. NativeAOT publishing/trimming has not been verified for this solution. -- **Server checkpoint limits:** CLI binds loopback; positional arguments are data +- **Previous control-plane checkpoint limits:** CLI binds loopback; positional arguments are data directory and TCP port. Guests are enabled there, and the hosting API can disable them. Accounts can be provisioned through `AccountStore` or native administration; automatic bootstrap/admin CLI is pending. Authentication enters unprotected Lobby @@ -106,25 +131,22 @@ dotnet restore dotnet/VoiceCat.slnx --locked-mode $env:VOICECAT_TLS_ORACLE = (Resolve-Path build/dev/bin/voicecat-dotnet-tls-oracle.exe).Path $env:VOICECAT_DATABASE_ORACLE = (Resolve-Path build/dev/bin/voicecat-dotnet-database-oracle.exe).Path $env:VOICECAT_VCCLI = (Resolve-Path build/dev/bin/vccli.exe).Path +$env:VOICECAT_VOICE_ORACLE = (Resolve-Path build/dev/bin/voicecat-dotnet-voice-oracle.exe).Path dotnet test dotnet/VoiceCat.slnx -c Release --no-restore ./dotnet/check-licenses.ps1 ``` -Last results: **142/142 managed tests, no skips with those variables set; 29/29 native +Last results: **154/154 managed tests, no skips with those variables set; 29/29 native CTest tests; warning-free Release build; 22 package licenses approved; locked restore and `git diff --check` passed.** Without the variables, native interoperability tests skip; that is not equivalent verification. Desktop CI stages codec/DSP bindings on Windows/Linux/macOS. Its Linux C++ job requires TLS, CLI and database interoperability and regenerates wire/password/DSP fixtures. Only Windows was run locally this session. -**Next behavior to implement:** read `docs/protocol.md`, `docs/security.md` and the -existing UDP binding/relay/session handlers. Write a real-client test for authenticated -UDP binding and encrypted relay between two clients, preserving SSRC/timestamp/flags -and encoded Opus while resealing with each recipient's directional key/counter. -Then implement binding tokens, stream announce/stop/subscription and channel routing. -Reject unauthenticated/replayed media and verify channel/subscription isolation. -Never decode audio on the SFU. Phase 4 exits only when two existing C++ clients can -exchange voice through the managed server as well as authenticate, join and chat. +**Voice behavior now verified:** real clients bind UDP and exchange encrypted Opus, +preserving SSRC/timestamp/flags while resealing with recipient-specific counters. +Never decode audio on the SFU. The two-C++-client voice/text criterion passes; +the remaining Phase 4 server behaviors still need implementation and conformance tests. - **Done (2026-09-15): Codec/DSP desktop port.** TLS checkpoint `2df79cd` committed and pushed to `origin/dotnet/foundations`. Added span-based Opus wrappers, safe native diff --git a/docs/api-dotnet.md b/docs/api-dotnet.md index 1e97f72..8904f20 100644 --- a/docs/api-dotnet.md +++ b/docs/api-dotnet.md @@ -188,8 +188,38 @@ Success returns permissions, then a cloned snapshot; peers receive joined/update events. Server-authoritative text replaces supplied sender ids/timestamps, limits bodies to 4096 UTF-8 bytes, and acknowledges valid or rejected routing. Channel text requires membership; private text echoes to sender and recipient. Protected channel -joins and all admin/moderation handlers are pending. No UDP port or media features -are advertised; voice subscription explicitly fails until the SFU is implemented. +joins and all admin/moderation handlers are pending. + +`VoiceServer.MediaEndPoint` exposes the bound UDP endpoint; UDP uses the same address +and port number as TCP, and `ServerHello.udp_port` advertises it. Successful authentication +issues a 16-byte binding token. TLS confirmation echoes an acknowledgement; a protocol-v2 +bootstrap packet binds the first UDP endpoint. Tokens cannot replace an established +endpoint; reconnect to change endpoints. Invalid tokens and malformed packets are ignored. + +Voice subscription, unsubscribe, stream announce/stop and stream-state signaling are +implemented. Announces require subscription and support microphone, screen audio and +auxiliary device streams, with at most 16 streams per user and labels up to 128 characters. +Stream ids are monotonically assigned per user; SSRCs are assigned server-wide. +Channel audio settings are authoritative; requested bitrate may lower the channel ceiling +(nonzero requests below 500 bps fail). User updates include the actor. Stream-state updates +use the authenticated sender id and ignore unknown stream ids. + +Channel movement clears active streams; joining the current channel preserves them. +Unsubscribe clears streams. Disconnect removes routing and retires media resources, +even if no UDP traffic follows. Senders must own the SSRC and be subscribed; recipients +must be subscribed, bound, in the same channel and not deafened. Server-muted senders +cannot relay. Every voice packet is authenticated with the sender's directional key; +the SFU reseals encoded bytes for each recipient without decoding, replacing only the +sequence and ciphertext/tag. Replay rejection precedes authentication; successful +authentication advances the replay window. + +The UDP loop exclusively owns media crypto, endpoint mutation and packet buffers. +Control handlers publish immutable routing snapshots. Crypto is created within the +TLS owner loop and transferred once. A coalesced notification wakes retired-key cleanup. +The synchronous fan-out core allocates zero managed bytes with platform ChaCha20-Poly1305; +socket scheduling and the allocating BouncyCastle fallback are outside that guarantee. +UDP keepalives are echoed for bound endpoints. The full media-aware reaper remains pending; +the existing 60-second TLS receive-idle timeout still applies. `AccountStore(path)` retains the C++ schema version 2, accepts version 1 migration, and rejects unknown revisions. Opening an existing channel table does not reseed it. diff --git a/docs/porting-to-dotnet.md b/docs/porting-to-dotnet.md index 615a5d3..04e83ae 100644 --- a/docs/porting-to-dotnet.md +++ b/docs/porting-to-dotnet.md @@ -3,7 +3,7 @@ **Status:** wire/media crypto and TLS/exporter foundations implemented under `dotnet/`, including C++ interoperability, persisted TOFU, compatible server credentials, codec/DSP wrappers, desktop native staging, and the initial managed TLS control server. -Phase 4 remains in progress; UDP relay, complete session administration, device audio, +Phase 4 remains in progress; complete session administration, device audio, managed client state, and UI phases remain planned. See `dotnet/README.md`, `docs/api-dotnet.md`, and `PROGRESS.md` for verification and next steps. **Target runtime:** .NET 10 LTS (in-service to Nov 2028), with .NET 11 as the follow-on. @@ -729,9 +729,24 @@ the existing BouncyCastle dependency with a strict libsodium PHC parser, not a n Konscious dependency. Native database tests prove password compatibility in both directions without resets. SQLite uses `Microsoft.Data.Sqlite.Core` 10.0.5, SQLitePCLRaw bundle 3.0.2 and explicitly pinned SourceGear SQLite 3.50.4.2. -See `docs/api-dotnet.md` for limits. UDP voice, streams, protected joins, moderation, -admin handlers and production configuration remain pending. The exit criterion -below has not yet been met. +See `docs/api-dotnet.md` for limits. This first checkpoint did not include UDP voice, +streams, protected joins, moderation, admin handlers or production configuration. +The subsequent voice checkpoint is described below. + +**Voice checkpoint:** the managed server now advertises UDP, issues session-bound +tokens, implements voice subscription and multi-stream signaling, and relays encrypted +Opus with recipient-specific counters. The first UDP endpoint is fixed for the session; +reconnect for endpoint changes. Immutable routing snapshots separate control handlers +from the UDP crypto owner. Real-socket tests cover replay/forgery/SSRC rejection, +channel/subscription isolation, stream stop and disconnect. A native client oracle +exercises bidirectional microphone and screen audio in mono and stereo. The fan-out +core has a 50-subscriber allocation regression test; transport scheduling and the +BouncyCastle crypto fallback are excluded from its zero-allocation guarantee. +Two real C++ `vccli` processes also pass join/text/bidirectional voice tests using +finite `--test-tone-ms` external capture/playback. The transport load test delivers +all 2,500 recipient packets from a sender paced at 50 pps to 50 subscribers. +Administration, protected joins, production configuration and media-aware reaping +still remain before Phase 4 completion. 1. `VoiceCat.Server`: accept loop, `ConnSession` protocol handling, session registry. 2. `Db` on `Microsoft.Data.Sqlite` — same schema. **Resolve the Argon2id hash-compat diff --git a/docs/roadmap.md b/docs/roadmap.md index 6cec4fc..45e8672 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -12,8 +12,13 @@ packaging and TLS/server/client migration remain later checkpoints. - Preserve the existing protobuf and 20-byte media wire formats; keep C++ as the oracle. - **Exit:** managed framing, headers, and ciphertext match fixtures generated by C++; managed tests and the existing C++ behavior suite pass. -- **Next:** prove TLS 1.3/exporter interoperability with C++, then port the server before - client state/audio/UI migration. Native audio packaging follows with codec/audio work. +- **Subsequent checkpoints:** TLS/exporter and credential interoperability, codec/DSP + desktop packaging, and managed control/UDP server slices are implemented. Two C++ + `vccli` processes authenticate, join, chat and exchange mono/stereo voice through + the managed server. The 50-subscriber fan-out core has an allocation regression test. +- **Next:** finish managed server administration, protected joins, production configuration + and media-aware reaping, then audio/client core, Windows cutover, C# AppKit and UIKit. + Keep the Swift ReplayKit extension and its shared ring; defer C++ removal until parity. - See `docs/porting-to-dotnet.md` and `dotnet/README.md`. Each milestone is shippable/testable on its own. The headless C++ test client (`vccli`) diff --git a/docs/security.md b/docs/security.md index 6d6bd9c..0b9287c 100644 --- a/docs/security.md +++ b/docs/security.md @@ -82,7 +82,8 @@ mandatory from the first build. This was chosen over DTLS after weighing two fin `0x01` for server→client. Each export yields a 32-byte directional media key. No second handshake, no certificates on the UDP path — the UDP channel inherits the authenticated, MITM-resistant TLS session's trust. -2. Each UDP voice frame is sealed with **ChaCha20-Poly1305** (libsodium, ISC license). +2. Each UDP voice frame is sealed with **ChaCha20-Poly1305** (libsodium in C++; + platform cryptography with a BouncyCastle fallback in .NET). 3. The full 20-byte header is AEAD **associated data**. The server authenticates/decrypts inbound media and reseals for each recipient, replacing the sequence with that recipient's next send counter. It forwards the encoded Opus bytes without decoding audio. @@ -115,14 +116,21 @@ the design depends on that. UDP packets are not individually authenticated to a *user* beyond the transport session. Binding works as: -1. `AuthResult.udp_token` (issued over TLS) is a short-lived, single-use, random token tied - to `session_id`. -2. Client's first UDP message is `UdpBinding{udp_token}`, sent as the first AEAD media frame - using the keys exported from the TLS session. -3. Server validates the token, binds the **5-tuple → session_id**, and discards the token. +1. `AuthResult.udp_token` (issued over TLS) is a random 16-byte token tied to the + authenticated session. The client confirms it with `UdpBinding` over TLS. +2. Protocol v2 bootstraps UDP with a **plaintext** `UDP_BINDING` packet: the 20-byte + binary header followed by the token. This is not a protobuf or an AEAD voice frame. +3. Server validates the token and binds the **5-tuple → session_id**. The managed server + accepts the first endpoint only; further bootstrap packets cannot replace it. + Endpoint changes require a new authenticated session. The token remains available + for TLS confirmation but cannot establish a second binding. Session removal retires + its endpoint, token and directional keys. The C++ oracle currently permits rebinding + with the same token; this differs in policy, not in the packet format. 4. Thereafter, frames are accepted only on that bound tuple; ssrcs are checked against the streams the session announced. Source-address spoofing can't hijack a session because the - attacker lacks the media key and the token. + attacker lacks the media key. The bootstrap token is visible on UDP, so it is not + a substitute for AEAD authentication and SSRC ownership checks. Header-only keepalives + are echoed only for bound endpoints; they provide liveness, not authenticated content. ## 4. Authentication & accounts (settled: guests + local accounts) diff --git a/dotnet/README.md b/dotnet/README.md index 35a4afd..27e7f86 100644 --- a/dotnet/README.md +++ b/dotnet/README.md @@ -137,5 +137,27 @@ managed code imports and authenticates it, then C++ authenticates a managed-crea account. CI also regenerates the libsodium password fixture. Native checks require the optional `VOICECAT_BUILD_DOTNET_ORACLE=ON` configure flag and a real-deps build. -Phase 4 remains in progress: UDP/SFU relay, streams, protected channel joins, -administration, moderation, and production configuration are the next server work. +The server also advertises UDP on the TCP port number, supports voice subscription +and stream signaling, and reseals encoded audio for subscribers in the same channel. +UDP binding fixes the first endpoint for the session; reconnect after endpoint changes. +Protected joins, administration, moderation, production configuration and full +media-aware reaping remain before Phase 4 completion. + +Enable deterministic native voice interoperability (no audio hardware required): + +```powershell +cmake --build --preset dev --target voicecat-dotnet-voice-oracle +$env:VOICECAT_VOICE_ORACLE = (Resolve-Path build/dev/bin/voicecat-dotnet-voice-oracle.exe).Path +dotnet test dotnet/VoiceCat.slnx -c Release --no-restore +``` + +Two existing C++ clients authenticate, join Lobby or Music Room, publish three +concurrent streams, feed PCM, and verify decoded energy and metadata in both directions. +The native clients use external capture/playback to avoid device dependencies in CI. +`MediaFanoutTests` separately verifies 50-subscriber routing/resealing without managed +allocations after warm-up and reports throughput; socket scheduling is excluded. +The transport load test delivers all 2,500 recipient packets from a paced 50 pps sender. +Native `vccli --test-tone-ms 4000` runs finite external capture/playback, feeds a tone, +and fails without decoded remote audio. Tests start two CLI processes in mono/stereo +channels and also verify channel text. Normal `--voice` now explicitly subscribes before +announcing its microphone stream. No C ABI or wire changes were needed. diff --git a/dotnet/oracle/CMakeLists.txt b/dotnet/oracle/CMakeLists.txt index 6050f93..609af65 100644 --- a/dotnet/oracle/CMakeLists.txt +++ b/dotnet/oracle/CMakeLists.txt @@ -8,6 +8,10 @@ target_link_libraries(voicecat-dotnet-tls-oracle PRIVATE voicecat::voicecat) target_include_directories(voicecat-dotnet-tls-oracle PRIVATE ${CMAKE_SOURCE_DIR}/core/src) target_compile_features(voicecat-dotnet-tls-oracle PRIVATE cxx_std_20) +add_executable(voicecat-dotnet-voice-oracle voice.cpp) +target_link_libraries(voicecat-dotnet-voice-oracle PRIVATE voicecat::voicecat) +target_compile_features(voicecat-dotnet-voice-oracle PRIVATE cxx_std_20) + add_executable(voicecat-dotnet-dsp-oracle dsp.cpp) target_link_libraries(voicecat-dotnet-dsp-oracle PRIVATE voicecat::voicecat) target_include_directories(voicecat-dotnet-dsp-oracle PRIVATE ${CMAKE_SOURCE_DIR}/core/src) diff --git a/dotnet/oracle/voice.cpp b/dotnet/oracle/voice.cpp new file mode 100644 index 0000000..3a31821 --- /dev/null +++ b/dotnet/oracle/voice.cpp @@ -0,0 +1,127 @@ +#include "voicecat.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +struct ClientState { + vc_client* client = nullptr; + std::mutex gate; + std::condition_variable changed; + bool authenticated = false; + bool subscribed = false; + bool joined = false; + uint32_t user = 0; + std::vector> streams; + std::array received{}; + long long energy = 0; + uint32_t channels = 0; +}; + +static void event(void* context, const vc_event* value) { + auto& state = *static_cast(context); + if (value->type == VC_EVENT_SERVER_IDENTITY) { + vc_confirm_server_identity(state.client, 1); + return; + } + std::lock_guard lock(state.gate); + switch (value->type) { + case VC_EVENT_AUTH_RESULT: + state.authenticated = value->result == VC_OK; + state.user = value->user_id; + break; + case VC_EVENT_VOICE_STATE: state.subscribed = value->u32a == 1; break; + case VC_EVENT_JOIN_RESULT: state.joined = value->result == VC_OK; break; + case VC_EVENT_STREAM_STARTED: state.streams.emplace_back(value->user_id, value->stream_id); break; + default: break; + } + state.changed.notify_all(); +} + +static void sink(void* context, uint32_t, uint32_t stream, const int16_t* pcm, + size_t samples, uint32_t channels, uint32_t rate) { + auto& state = *static_cast(context); + if (rate != 48000 || stream >= state.received.size()) return; + std::lock_guard lock(state.gate); + ++state.received[stream]; + state.channels = channels; + for (size_t index = 0; index < samples * channels; ++index) state.energy += std::abs(static_cast(pcm[index])); + state.changed.notify_all(); +} + +template +static bool wait(ClientState& state, Predicate predicate) { + std::unique_lock lock(state.gate); + return state.changed.wait_for(lock, std::chrono::seconds(8), predicate); +} + +struct Destroy { + void operator()(vc_client* client) const { vc_disconnect(client); vc_client_destroy(client); } +}; +using Client = std::unique_ptr; + +static Client connect(ClientState& state, uint16_t port, uint32_t channel, const char* nickname) { + vc_config config{"dotnet-voice-oracle", "1", VC_LOG_OFF}; + Client client(vc_client_create(&config, {event, nullptr, &state})); + state.client = client.get(); + if (!client || vc_set_external_playback(client.get(), 1) != VC_OK || + vc_connect(client.get(), "127.0.0.1", port) != VC_OK || + vc_authenticate_guest(client.get(), nickname) != VC_OK || + !wait(state, [&] { return state.authenticated; }) || + vc_join_channel(client.get(), channel, nullptr) != VC_OK || + !wait(state, [&] { return state.joined; }) || + vc_join_voice(client.get()) != VC_OK || + !wait(state, [&] { return state.subscribed; }) || + vc_set_pcm_sink(client.get(), sink, &state) != VC_OK) return {}; + return client; +} + +int main(int argc, char** argv) { + if (argc != 3) return 1; + uint16_t port = static_cast(std::strtoul(argv[1], nullptr, 10)); + uint32_t channel = static_cast(std::strtoul(argv[2], nullptr, 10)); + ClientState alice, bob; + Client a = connect(alice, port, channel, "Native Alice"); + Client b = connect(bob, port, channel, "Native Bob"); + if (!a || !b) { std::fprintf(stderr, "native authentication/join/subscription failed\n"); return 1; } + std::array ids{}; + vc_stream_desc mic{}; + mic.kind = VC_STREAM_MIC; + mic.external_feed = 1; + vc_stream_desc screen = mic; + screen.kind = VC_STREAM_SCREEN_AUDIO; + if (vc_stream_start(a.get(), &mic, &ids[0]) != VC_OK || + vc_stream_start(a.get(), &screen, &ids[1]) != VC_OK || + vc_stream_start(b.get(), &mic, &ids[2]) != VC_OK || + !wait(alice, [&] { return alice.streams.size() >= 3; }) || + !wait(bob, [&] { return bob.streams.size() >= 3; })) { + std::fprintf(stderr, "native stream signaling failed\n"); return 1; + } + uint32_t channels = channel == 2 ? 2 : 1; + std::vector pcm(960 * channels); + for (size_t sample = 0; sample < 960; ++sample) + for (uint32_t side = 0; side < channels; ++side) + pcm[sample * channels + side] = static_cast(12000 * std::sin(sample * (side == 0 ? 0.058 : 0.083))); + for (int frame = 0; frame < 100; ++frame) { + if (vc_stream_feed_pcm(a.get(), ids[0], pcm.data(), 960, channels) != VC_OK || + vc_stream_feed_pcm(a.get(), ids[1], pcm.data(), 960, channels) != VC_OK || + vc_stream_feed_pcm(b.get(), ids[2], pcm.data(), 960, channels) != VC_OK) return 1; + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + } + bool received = wait(alice, [&] { return alice.received[ids[2]] >= 5 && alice.energy > 0; }) && + wait(bob, [&] { return bob.received[ids[0]] >= 5 && bob.received[ids[1]] >= 5 && bob.energy > 0; }); + { + std::scoped_lock lock(alice.gate, bob.gate); + std::printf("channel=%u channels=%u alice=%d bob-mic=%d bob-screen=%d energy=%lld/%lld\n", + channel, channels, alice.received[ids[2]], bob.received[ids[0]], bob.received[ids[1]], alice.energy, bob.energy); + received = received && alice.channels == channels && bob.channels == channels; + } + return received ? 0 : 1; +} diff --git a/dotnet/src/VoiceCat.Server/Transport/MediaFanout.cs b/dotnet/src/VoiceCat.Server/Transport/MediaFanout.cs new file mode 100644 index 0000000..f43a8a3 --- /dev/null +++ b/dotnet/src/VoiceCat.Server/Transport/MediaFanout.cs @@ -0,0 +1,50 @@ +using System.Net; +using System.Security.Cryptography; +using VoiceCat.Protocol; + +namespace VoiceCat.Server.Transport; + +// One packet at a time. Each returned buffer must be sent before preparing the next recipient. +internal sealed class MediaFanout : IDisposable +{ + private readonly byte[] plaintext = new byte[65535]; + private readonly byte[] output = new byte[65535]; + private MediaRoute[] routes = []; + private MediaRoute? source; + private VoiceFrameHeader header; + private int length; + private int index; + + public bool TryStart(ReadOnlySpan packet, MediaRoute sender, MediaRoute[] recipients) + { + source = null; + if (!VoiceFrameHeader.TryRead(packet, out var candidate) || candidate.Type != MediaFrameType.Voice || candidate.Codec != 0 || + !sender.Subscribed || sender.Muted || !sender.Sources.Contains(candidate.Ssrc) || + packet.Length <= VoiceFrameHeader.Size + 16 || packet.Length > output.Length) return false; + if (!sender.Peer.Crypto.Decryptor.TryDecrypt(packet, plaintext, out header, out length)) return false; + source = sender; + routes = recipients; + index = 0; + return true; + } + + public bool TryNext(out ReadOnlyMemory packet, out SocketAddress? endpoint) + { + packet = default; + endpoint = null; + if (source is null) return false; + while (index < routes.Length) + { + MediaRoute recipient = routes[index++]; + if (ReferenceEquals(recipient.Peer, source.Peer) || recipient.ChannelId != source.ChannelId || + !recipient.Subscribed || recipient.Deafened || recipient.Peer.Endpoint is null) continue; + int size = recipient.Peer.Crypto.Encryptor.Encrypt(header, plaintext.AsSpan(0, length), output); + packet = output.AsMemory(0, size); + endpoint = recipient.Peer.Endpoint; + return true; + } + return false; + } + + public void Dispose() => CryptographicOperations.ZeroMemory(plaintext); +} diff --git a/dotnet/src/VoiceCat.Server/Transport/MediaRelay.cs b/dotnet/src/VoiceCat.Server/Transport/MediaRelay.cs new file mode 100644 index 0000000..e6faf8a --- /dev/null +++ b/dotnet/src/VoiceCat.Server/Transport/MediaRelay.cs @@ -0,0 +1,150 @@ +using System.Collections.Concurrent; +using System.Net; +using System.Net.Sockets; +using System.Security.Cryptography; +using System.Threading.Channels; +using VoiceCat.Protocol; + +namespace VoiceCat.Server.Transport; + +internal sealed class MediaPeer(byte[] token, MediaSessionCrypto crypto) +{ + public byte[] Token { get; } = token; + public MediaSessionCrypto Crypto { get; } = crypto; + // Only the UDP loop reads or changes the endpoint and binding state. + public SocketAddress? Endpoint { get; set; } + public void Dispose() { Crypto.Dispose(); CryptographicOperations.ZeroMemory(Token); } +} + +internal sealed record MediaRoute(MediaPeer Peer, uint ChannelId, bool Subscribed, bool Muted, bool Deafened, uint[] Sources); + +internal sealed class MediaRelay : IAsyncDisposable +{ + private readonly Socket socket; + private readonly CancellationTokenSource shutdown = new(); + private readonly ConcurrentQueue retired = new(); + private readonly Channel changed = Channel.CreateBounded(1); + private MediaRoute[] routes = []; + private readonly byte[] input = new byte[65535]; + private readonly MediaFanout fanout = new(); + private readonly Task receiving; + public IPEndPoint EndPoint { get; } + public event Action? Failed; + + public MediaRelay(IPEndPoint endpoint) + { + socket = new(endpoint.AddressFamily, SocketType.Dgram, ProtocolType.Udp); + try { socket.Bind(endpoint); EndPoint = (IPEndPoint)socket.LocalEndPoint!; } + catch { socket.Dispose(); shutdown.Dispose(); throw; } + receiving = ReceiveAsync(); + } + + // Publications are serialized by the server's session gate. Crypto ownership transfers here. + public void Publish(MediaRoute[] next) + { + MediaRoute[] previous = Volatile.Read(ref routes); + Volatile.Write(ref routes, next); + foreach (MediaRoute route in previous) + if (!next.Any(candidate => ReferenceEquals(candidate.Peer, route.Peer))) retired.Enqueue(route.Peer); + changed.Writer.TryWrite(0); + } + + private void DrainRetired() + { + while (retired.TryDequeue(out MediaPeer? peer)) peer.Dispose(); + } + + private async Task ReceiveAsync() + { + var sender = new SocketAddress(socket.AddressFamily); + Task? receive = null; + Task? update = null; + try + { + while (true) + { + receive ??= socket.ReceiveFromAsync(input, SocketFlags.None, sender, shutdown.Token).AsTask(); + update ??= changed.Reader.WaitToReadAsync(shutdown.Token).AsTask(); + await Task.WhenAny(receive, update).ConfigureAwait(false); + if (update.IsCompleted) + { + await update.ConfigureAwait(false); + while (changed.Reader.TryRead(out _)) { } + update = null; + DrainRetired(); + } + if (!receive.IsCompleted) continue; + int length; + try { length = await receive.ConfigureAwait(false); } + catch (SocketException exception) when (exception.SocketErrorCode is SocketError.MessageSize or SocketError.ConnectionReset) { continue; } + finally { receive = null; } + DrainRetired(); + MediaRoute[] current = Volatile.Read(ref routes); + if (!VoiceFrameHeader.TryRead(input.AsSpan(0, length), out var header)) continue; + MediaRoute? source = null; + foreach (MediaRoute route in current) + if (route.Peer.Endpoint?.Equals(sender) == true) { source = route; break; } + + if (header.Type == MediaFrameType.UdpBinding) + { + if (length != VoiceFrameHeader.Size + 16 || source is not null) continue; + foreach (MediaRoute route in current) + { + if (route.Peer.Endpoint is not null || !CryptographicOperations.FixedTimeEquals(route.Peer.Token, input.AsSpan(VoiceFrameHeader.Size, 16))) continue; + var bound = new SocketAddress(sender.Family, sender.Size); + for (int index = 0; index < sender.Size; index++) bound[index] = sender[index]; + route.Peer.Endpoint = bound; + break; + } + continue; + } + if (source is null) continue; + if (header.Type == MediaFrameType.Keepalive) + { + if (length == VoiceFrameHeader.Size) await SendAsync(input.AsMemory(0, length), sender).ConfigureAwait(false); + continue; + } + if (!fanout.TryStart(input.AsSpan(0, length), source, current)) continue; + while (fanout.TryNext(out ReadOnlyMemory packet, out SocketAddress? endpoint)) + await SendAsync(packet, endpoint!).ConfigureAwait(false); + } + } + catch (Exception exception) when (shutdown.IsCancellationRequested && exception is OperationCanceledException or SocketException or ObjectDisposedException) { } + catch (Exception exception) { Failed?.Invoke(exception); throw; } + finally + { + shutdown.Cancel(); + socket.Dispose(); + fanout.Dispose(); + if (receive is not null) + { + try { await receive.ConfigureAwait(false); } + catch (Exception exception) when (exception is OperationCanceledException or SocketException or ObjectDisposedException) { } + } + if (update is not null) + { + try { await update.ConfigureAwait(false); } + catch (OperationCanceledException) { } + } + } + } + + private async ValueTask SendAsync(ReadOnlyMemory packet, SocketAddress endpoint) + { + try { await socket.SendToAsync(packet, SocketFlags.None, endpoint, shutdown.Token).ConfigureAwait(false); } + catch (SocketException exception) when (exception.SocketErrorCode is SocketError.ConnectionReset or SocketError.HostUnreachable or SocketError.NetworkUnreachable) { } + } + + public async ValueTask DisposeAsync() + { + shutdown.Cancel(); + socket.Dispose(); + try { await receiving.ConfigureAwait(false); } + finally + { + DrainRetired(); + foreach (MediaRoute route in Volatile.Read(ref routes)) route.Peer.Dispose(); + shutdown.Dispose(); + } + } +} diff --git a/dotnet/src/VoiceCat.Server/Transport/MediaSessionCrypto.cs b/dotnet/src/VoiceCat.Server/Transport/MediaSessionCrypto.cs new file mode 100644 index 0000000..87193ba --- /dev/null +++ b/dotnet/src/VoiceCat.Server/Transport/MediaSessionCrypto.cs @@ -0,0 +1,10 @@ +using VoiceCat.Crypto; + +namespace VoiceCat.Server.Transport; + +internal sealed class MediaSessionCrypto(MediaEncryptor encryptor, MediaDecryptor decryptor) : IDisposable +{ + public MediaEncryptor Encryptor { get; } = encryptor; + public MediaDecryptor Decryptor { get; } = decryptor; + public void Dispose() { Encryptor.Dispose(); Decryptor.Dispose(); } +} diff --git a/dotnet/src/VoiceCat.Server/Transport/TlsControlConnection.cs b/dotnet/src/VoiceCat.Server/Transport/TlsControlConnection.cs index 64804ca..d4ad006 100644 --- a/dotnet/src/VoiceCat.Server/Transport/TlsControlConnection.cs +++ b/dotnet/src/VoiceCat.Server/Transport/TlsControlConnection.cs @@ -21,6 +21,8 @@ internal sealed class TlsControlConnection : IAsyncDisposable private int prefixBytes; private byte[]? payload; private int payloadBytes; + private readonly TaskCompletionSource mediaReady = new(TaskCreationOptions.RunContinuationsAsynchronously); + private MediaSessionCrypto? mediaCrypto; public Task Completion { get; } public CancellationToken CancellationToken => lifetime.Token; @@ -48,6 +50,12 @@ internal sealed class TlsControlConnection : IAsyncDisposable public void CompleteWrites() => outgoing.Writer.TryComplete(); + internal async Task TakeMediaCryptoAsync(CancellationToken cancellationToken) + { + await mediaReady.Task.WaitAsync(cancellationToken).ConfigureAwait(false); + return Interlocked.Exchange(ref mediaCrypto, null) ?? throw new InvalidOperationException("Media crypto already has an owner."); + } + private async Task RunAsync() { byte[] ciphertext = new byte[16384]; @@ -80,6 +88,13 @@ internal sealed class TlsControlConnection : IAsyncDisposable break; } tls.ReceiveCiphertext(ciphertext.AsSpan(0, count)); + if (tls.IsReady && !mediaReady.Task.IsCompleted) + { + var encryptor = tls.CreateMediaEncryptor(); + try { mediaCrypto = new(encryptor, tls.CreateMediaDecryptor()); } + catch { encryptor.Dispose(); throw; } + mediaReady.SetResult(); + } if (tls.IsReady) lifetime.CancelAfter(TimeSpan.FromSeconds(60)); while ((count = tls.ReadPlaintext(plaintext)) > 0) Parse(plaintext.AsSpan(0, count)); await FlushAsync(sendBuffer, cancellationToken).ConfigureAwait(false); @@ -104,6 +119,7 @@ internal sealed class TlsControlConnection : IAsyncDisposable } finally { + mediaReady.TrySetCanceled(); lifetime.Cancel(); socket.Dispose(); if (receive is not null) @@ -163,7 +179,7 @@ internal sealed class TlsControlConnection : IAsyncDisposable public async ValueTask DisposeAsync() { lifetime.Cancel(); - await Completion.ConfigureAwait(false); - lifetime.Dispose(); + try { await Completion.ConfigureAwait(false); } + finally { Interlocked.Exchange(ref mediaCrypto, null)?.Dispose(); lifetime.Dispose(); } } } diff --git a/dotnet/src/VoiceCat.Server/VoiceServer.cs b/dotnet/src/VoiceCat.Server/VoiceServer.cs index 858886d..1bb71ae 100644 --- a/dotnet/src/VoiceCat.Server/VoiceServer.cs +++ b/dotnet/src/VoiceCat.Server/VoiceServer.cs @@ -13,6 +13,7 @@ namespace VoiceCat.Server; public sealed class VoiceServer : IAsyncDisposable { private readonly Socket listener; + private readonly MediaRelay media; private readonly ServerCredentials credentials; private readonly AccountStore accounts; private readonly IReadOnlyList channels; @@ -24,10 +25,12 @@ public sealed class VoiceServer : IAsyncDisposable private readonly List connections = []; private ulong nextSession; private uint nextUser; + private uint nextSsrc; private readonly Task accepting; private int disposed; public IPEndPoint EndPoint => (IPEndPoint)listener.LocalEndPoint!; + public IPEndPoint MediaEndPoint => media.EndPoint; public event Action? ConnectionFailed; public VoiceServer(string directory, IPEndPoint endpoint, bool allowGuests = true, string name = "VoiceCat Server") @@ -42,6 +45,8 @@ public sealed class VoiceServer : IAsyncDisposable listener = new Socket(endpoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp); listener.Bind(endpoint); listener.Listen(64); + media = new((IPEndPoint)listener.LocalEndPoint!); + media.Failed += exception => ConnectionFailed?.Invoke(exception); } catch { @@ -96,7 +101,8 @@ public sealed class VoiceServer : IAsyncDisposable Reject(session, "Unsupported protocol version or banned address."); break; } - var hello = new ServerHello { ProtoVersion = 2, ServerName = name, ServerVersion = "0.1.0-dotnet", ServerIdentityFingerprint = ByteString.CopyFrom(SHA256.HashData(credentials.Identity.PublicKey)) }; + session.Media = new(RandomNumberGenerator.GetBytes(16), await session.Connection.TakeMediaCryptoAsync(shutdown.Token).ConfigureAwait(false)); + var hello = new ServerHello { ProtoVersion = 2, ServerName = name, ServerVersion = "0.1.0-dotnet", UdpPort = checked((uint)media.EndPoint.Port), ServerIdentityFingerprint = ByteString.CopyFrom(SHA256.HashData(credentials.Identity.PublicKey)) }; if (allowGuests) hello.AuthMethods.Add("guest"); hello.AuthMethods.Add("password"); session.Connection.TrySend(new() { RequestId = envelope.RequestId, ServerHello = hello }); @@ -114,8 +120,14 @@ public sealed class VoiceServer : IAsyncDisposable case Envelope.BodyOneofCase.TextMessage: RelayText(session, envelope.TextMessage); break; case Envelope.BodyOneofCase.Subscribe: SendSnapshot(session); break; case Envelope.BodyOneofCase.JoinChannel: Join(session, envelope.RequestId, envelope.JoinChannel.ChannelId); break; - case Envelope.BodyOneofCase.SubscribeVoice: - session.Connection.TrySend(new() { RequestId = envelope.RequestId, VoiceSubscriptionResult = new() { Error = "Managed media relay is not implemented yet." } }); + case Envelope.BodyOneofCase.SubscribeVoice: SubscribeVoice(session, envelope.RequestId, true); break; + case Envelope.BodyOneofCase.UnsubscribeVoice: SubscribeVoice(session, envelope.RequestId, false); break; + case Envelope.BodyOneofCase.StreamAnnounce: AnnounceStream(session, envelope.RequestId, envelope.StreamAnnounce); break; + case Envelope.BodyOneofCase.StreamStop: StopStream(session, envelope.StreamStop.StreamId); break; + case Envelope.BodyOneofCase.StreamState: UpdateStream(session, envelope.StreamState); break; + case Envelope.BodyOneofCase.UdpBinding: + if (!envelope.UdpBinding.Ack && CryptographicOperations.FixedTimeEquals(envelope.UdpBinding.UdpToken.Span, session.Media!.Token)) + session.Connection.TrySend(new() { RequestId = envelope.RequestId, UdpBinding = new() { Ack = true } }); break; default: session.Connection.TrySend(new() { RequestId = envelope.RequestId, GenericResult = new() { Code = 1, Message = "Operation is not implemented by this server checkpoint." } }); @@ -133,6 +145,8 @@ public sealed class VoiceServer : IAsyncDisposable lock (gate) { sessions.Remove(session.Id); + if (session.User is null) session.Media?.Dispose(); + else PublishMedia(); if (session.User is not null) Broadcast(new() { UserEvent = new() { Kind = UserEvent.Types.Kind.Left, LeftId = session.User.Id } }); } await session.Connection.DisposeAsync().ConfigureAwait(false); @@ -170,9 +184,10 @@ public sealed class VoiceServer : IAsyncDisposable session.User = user; session.Connection.TrySend(new() { RequestId = requestId, AuthResult = new() { - Ok = true, SessionId = session.Id, Self = user.Clone(), + Ok = true, SessionId = session.Id, Self = user.Clone(), UdpToken = ByteString.CopyFrom(session.Media!.Token), Permissions = new() { IsAdmin = admin, CanAdminAccounts = admin, CanBan = admin, CanKick = admin, CanMoveUsers = admin, CanCreateTempChannel = admin } } }); + PublishMedia(); Broadcast(new() { UserEvent = new() { Kind = UserEvent.Types.Kind.Joined, User = user.Clone() } }, session.Id); SendSnapshot(session); } @@ -199,7 +214,9 @@ public sealed class VoiceServer : IAsyncDisposable session.Connection.TrySend(new() { RequestId = requestId, JoinChannelResult = new() { Error = "Channel unavailable." } }); return; } - session.User!.ChannelId = channelId; + if (session.User!.ChannelId != channelId) session.User.Streams.Clear(); + session.User.ChannelId = channelId; + PublishMedia(); var result = new JoinChannelResult { Ok = true, ChannelId = channelId, Audio = channel.Audio.Clone() }; result.Members.Add(sessions.Values.Where(peer => peer.User?.ChannelId == channelId).Select(peer => peer.User!.Clone())); session.Connection.TrySend(new() { RequestId = requestId, JoinChannelResult = result }); @@ -229,6 +246,69 @@ public sealed class VoiceServer : IAsyncDisposable } } + private void PublishMedia() + { + media.Publish(sessions.Values.Where(peer => peer.User is not null).Select(peer => new MediaRoute( + peer.Media!, peer.User!.ChannelId, peer.User.VoiceSubscribed, peer.User.ServerMuted, peer.User.SelfDeafened || peer.User.ServerDeafened, + peer.User.Streams.Select(stream => stream.Ssrc).ToArray())).ToArray()); + } + + private void BroadcastUser(Session session) => Broadcast(new() { UserEvent = new() { Kind = UserEvent.Types.Kind.Updated, User = session.User!.Clone() } }); + + private void SubscribeVoice(Session session, ulong requestId, bool subscribed) + { + lock (gate) + { + session.User!.VoiceSubscribed = subscribed; + if (!subscribed) session.User.Streams.Clear(); + PublishMedia(); + session.Connection.TrySend(new() { RequestId = requestId, VoiceSubscriptionResult = new() { Ok = true, Subscribed = subscribed } }); + BroadcastUser(session); + } + } + + private void AnnounceStream(Session session, ulong requestId, StreamAnnounce request) + { + lock (gate) + { + if (!session.User!.VoiceSubscribed || !Enum.IsDefined(request.Kind) || request.Label.Length > 128 || session.User.Streams.Count >= 16 || + nextSsrc == uint.MaxValue || session.NextStream == uint.MaxValue || request.RequestedAudio?.BitrateBps is > 0 and < 500) + { + session.Connection.TrySend(new() { RequestId = requestId, StreamAnnounceResult = new() { Error = "Voice subscription required, invalid stream, or stream limit reached." } }); + return; + } + AudioConfig audio = channels.First(channel => channel.Id == session.User.ChannelId).Audio.Clone(); + if (request.RequestedAudio?.BitrateBps > 0) audio.BitrateBps = Math.Min(audio.BitrateBps, request.RequestedAudio.BitrateBps); + var stream = new StreamInfo { StreamId = ++session.NextStream, Ssrc = ++nextSsrc, Kind = request.Kind, Label = request.Label, Audio = audio }; + session.User.Streams.Add(stream); + PublishMedia(); + session.Connection.TrySend(new() { RequestId = requestId, StreamAnnounceResult = new() { Ok = true, StreamId = stream.StreamId, Ssrc = stream.Ssrc, EffectiveAudio = audio.Clone() } }); + BroadcastUser(session); + } + } + + private void StopStream(Session session, uint streamId) + { + lock (gate) + { + StreamInfo? stream = session.User!.Streams.FirstOrDefault(candidate => candidate.StreamId == streamId); + if (stream is null) return; + session.User.Streams.Remove(stream); + PublishMedia(); + BroadcastUser(session); + } + } + + private void UpdateStream(Session session, StreamStateUpdate update) + { + lock (gate) + { + StreamInfo? stream = session.User!.Streams.FirstOrDefault(candidate => candidate.StreamId == update.StreamId); + if (stream is null) return; + Broadcast(new() { StreamState = new() { UserId = session.User.Id, StreamId = stream.StreamId, Muted = update.Muted, Talking = update.Talking } }); + } + } + private void Broadcast(Envelope envelope, ulong excluded = 0) { foreach (Session recipient in sessions.Values.Where(peer => peer.Id != excluded && peer.User is not null)) recipient.Connection.TrySend(envelope); @@ -248,9 +328,8 @@ public sealed class VoiceServer : IAsyncDisposable } finally { - accounts.Dispose(); - credentials.Dispose(); - shutdown.Dispose(); + try { await media.DisposeAsync().ConfigureAwait(false); } + finally { accounts.Dispose(); credentials.Dispose(); shutdown.Dispose(); } } } @@ -261,5 +340,7 @@ public sealed class VoiceServer : IAsyncDisposable public string Address { get; } = address; public bool HelloReceived { get; set; } public User? User { get; set; } + public MediaPeer? Media { get; set; } + public uint NextStream; } } diff --git a/dotnet/tests/VoiceCat.Tests/MediaFanoutTests.cs b/dotnet/tests/VoiceCat.Tests/MediaFanoutTests.cs new file mode 100644 index 0000000..07a282a --- /dev/null +++ b/dotnet/tests/VoiceCat.Tests/MediaFanoutTests.cs @@ -0,0 +1,110 @@ +using System.Diagnostics; +using System.Net; +using System.Net.Sockets; +using System.Security.Cryptography; +using VoiceCat.Crypto; +using VoiceCat.Protocol; +using VoiceCat.Server.Transport; +using Xunit.Abstractions; + +namespace VoiceCat.Tests; + +public sealed class MediaFanoutTests(ITestOutputHelper output) +{ + [Fact] + public async Task UdpRelayDeliversFiftyPacketsPerSecondToFiftySubscribers() + { + await using var relay = new MediaRelay(new(IPAddress.Loopback, 0)); + byte[] key = Enumerable.Range(0, 32).Select(i => (byte)i).ToArray(); + Socket[] sockets = Enumerable.Range(0, 51).Select(_ => new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp)).ToArray(); + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + try + { + foreach (Socket socket in sockets) + { + socket.ReceiveBufferSize = 1024 * 1024; + socket.Bind(new IPEndPoint(IPAddress.Loopback, 0)); + } + MediaRoute[] routes = sockets.Select(socket => new MediaRoute( + new(new byte[16], new(new(key), new(key))) { Endpoint = ((IPEndPoint)socket.LocalEndPoint!).Serialize() }, + 1, true, false, false, [42])).ToArray(); + relay.Publish(routes); + byte[] payload = Enumerable.Range(0, 120).Select(i => (byte)i).ToArray(); + Task[] receivers = sockets.Skip(1).Select(async socket => + { + using var decryptor = new MediaDecryptor(key); + byte[] packet = new byte[4000]; + byte[] decoded = new byte[4000]; + for (ulong sequence = 0; sequence < 50; sequence++) + { + int length = await socket.ReceiveAsync(packet, SocketFlags.None, timeout.Token); + Assert.True(decryptor.TryDecrypt(packet.AsSpan(0, length), decoded, out var header, out int bytes)); + Assert.Equal(sequence, header.Sequence); + Assert.Equal(payload, decoded[..bytes]); + } + }).ToArray(); + using var encryptor = new MediaEncryptor(key); + byte[] outgoing = new byte[VoiceFrameHeader.Size + payload.Length + 16]; + var elapsed = Stopwatch.StartNew(); + for (uint index = 0; index < 50; index++) + { + encryptor.Encrypt(new(MediaFrameType.Voice, 0, 0, 42, 0, index * 960), payload, outgoing); + await sockets[0].SendToAsync(outgoing, SocketFlags.None, relay.EndPoint, timeout.Token); + await Task.Delay(20, timeout.Token); + } + await Task.WhenAll(receivers); + output.WriteLine($"Delivered all 2,500 recipient packets in {elapsed.Elapsed.TotalMilliseconds:F1} ms at a paced 50 pps input."); + } + finally { foreach (Socket socket in sockets) socket.Dispose(); } + } + + [PlatformCipherFact] + public void FiftySubscriberFanoutAllocatesNoManagedMemoryAndPreservesPayload() + { + byte[] key = Enumerable.Range(0, 32).Select(i => (byte)i).ToArray(); + MediaRoute[] routes = Enumerable.Range(0, 51).Select(i => new MediaRoute( + new(new byte[16], new(new(key), new(key))) { Endpoint = new IPEndPoint(IPAddress.Loopback, 10000 + i).Serialize() }, + 1, true, false, false, [42])).ToArray(); + using var sender = new MediaEncryptor(key); + using var receiver = new MediaDecryptor(key); + using var fanout = new MediaFanout(); + byte[] payload = Enumerable.Range(0, 120).Select(i => (byte)i).ToArray(); + byte[] packet = new byte[VoiceFrameHeader.Size + payload.Length + 16]; + byte[] decoded = new byte[payload.Length]; + ReadOnlyMemory last = default; + try + { + for (int i = 0; i < 100; i++) Cycle(); + long before = GC.GetAllocatedBytesForCurrentThread(); + long started = Stopwatch.GetTimestamp(); + for (int i = 0; i < 1000; i++) Cycle(); + TimeSpan elapsed = Stopwatch.GetElapsedTime(started); + long allocated = GC.GetAllocatedBytesForCurrentThread() - before; + Assert.Equal(0, allocated); + Assert.True(receiver.TryDecrypt(last.Span, decoded, out var header, out int length)); + Assert.Equal(payload.Length, length); + Assert.Equal(payload, decoded); + Assert.Equal(42U, header.Ssrc); + Assert.Equal(1099UL, header.Sequence); + output.WriteLine($"50,000 recipient seals in {elapsed.TotalMilliseconds:F1} ms; {allocated} managed bytes. Transport scheduling is excluded."); + } + finally { foreach (MediaRoute route in routes) route.Peer.Dispose(); } + + void Cycle() + { + sender.Encrypt(new(MediaFrameType.Voice, 0, 0, 42, 0, 960), payload, packet); + if (!fanout.TryStart(packet, routes[0], routes)) throw new InvalidOperationException("Valid packet rejected."); + int recipients = 0; + while (fanout.TryNext(out var next, out _)) { last = next; recipients++; } + if (recipients != 50) throw new InvalidOperationException("Incorrect fanout."); + } + } + + private sealed class PlatformCipherFactAttribute : FactAttribute + { + public PlatformCipherFactAttribute() + { + if (!ChaCha20Poly1305.IsSupported) Skip = "The allocation guarantee requires platform ChaCha20-Poly1305; fallback conformance is tested separately."; + } + } +} diff --git a/dotnet/tests/VoiceCat.Tests/MediaRelayTests.cs b/dotnet/tests/VoiceCat.Tests/MediaRelayTests.cs new file mode 100644 index 0000000..599ddca --- /dev/null +++ b/dotnet/tests/VoiceCat.Tests/MediaRelayTests.cs @@ -0,0 +1,303 @@ +using System.Net; +using System.Diagnostics; +using System.Net.Sockets; +using VoiceCat.Protocol; +using VoiceCat.Server.Transport; +using Voicecat.V1; +using static VoiceCat.Tests.ServerTests; + +namespace VoiceCat.Tests; + +public sealed class MediaRelayTests +{ + [CppCliVoiceTheory] + [InlineData(1)] + [InlineData(2)] + public async Task TwoCppCliProcessesJoinChatAndExchangeVoice(int channel) + { + await using var fixture = new ServerFixture(); + await using var observer = await fixture.ConnectAsync(); + await observer.LoginAsync("Observer"); + observer.Send(new() { JoinChannel = new() { ChannelId = checked((uint)channel) } }); + Assert.True((await observer.ReadUntilAsync(e => e.JoinChannelResult is not null)).JoinChannelResult.Ok); + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + await Task.WhenAll(RunAsync("Cli Alice"), RunAsync("Cli Bob")); + var first = (await observer.ReadUntilAsync(e => e.TextMessage is not null)).TextMessage; + var second = (await observer.ReadUntilAsync(e => e.TextMessage is not null)).TextMessage; + Assert.Equal("CLI voice checkpoint", first.Body); + Assert.Equal(first.Body, second.Body); + Assert.NotEqual(first.SenderId, second.SenderId); + + async Task RunAsync(string nickname) + { + var start = new ProcessStartInfo(Environment.GetEnvironmentVariable("VOICECAT_VCCLI")!) + { + WorkingDirectory = fixture.Directory, UseShellExecute = false, CreateNoWindow = true, + RedirectStandardOutput = true, RedirectStandardError = true + }; + foreach (string argument in new[] { "--host", "127.0.0.1", "--port", fixture.Server.EndPoint.Port.ToString(System.Globalization.CultureInfo.InvariantCulture), + "--nick", nickname, "--channel", channel.ToString(System.Globalization.CultureInfo.InvariantCulture), "--text", "CLI voice checkpoint", "--test-tone-ms", "4000" }) + start.ArgumentList.Add(argument); + using var process = Process.Start(start)!; + Task stdout = process.StandardOutput.ReadToEndAsync(); + Task stderr = process.StandardError.ReadToEndAsync(); + try + { + await process.WaitForExitAsync(timeout.Token); + string log = await stdout + await stderr; + Assert.True(process.ExitCode == 0, log); + Assert.Contains("[test-tone] received=", log); + } + finally { if (!process.HasExited) { process.Kill(true); await process.WaitForExitAsync(); } } + } + } + + private sealed class CppCliVoiceTheoryAttribute : TheoryAttribute + { + public CppCliVoiceTheoryAttribute() + { + if (string.IsNullOrEmpty(Environment.GetEnvironmentVariable("VOICECAT_VCCLI"))) Skip = "Set VOICECAT_VCCLI to the existing native CLI."; + } + } + + [VoiceOracleTheory] + [InlineData(1)] + [InlineData(2)] + public async Task ExistingCppClientsExchangeBidirectionalVoiceThroughManagedServer(int channel) + { + await using var fixture = new ServerFixture(); + var start = new ProcessStartInfo(Environment.GetEnvironmentVariable("VOICECAT_VOICE_ORACLE")!) + { + WorkingDirectory = fixture.Directory, UseShellExecute = false, CreateNoWindow = true, + RedirectStandardOutput = true, RedirectStandardError = true + }; + start.ArgumentList.Add(fixture.Server.EndPoint.Port.ToString(System.Globalization.CultureInfo.InvariantCulture)); + start.ArgumentList.Add(channel.ToString(System.Globalization.CultureInfo.InvariantCulture)); + using var process = Process.Start(start)!; + Task output = process.StandardOutput.ReadToEndAsync(); + Task error = process.StandardError.ReadToEndAsync(); + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(45)); + try + { + await process.WaitForExitAsync(timeout.Token); + Assert.True(process.ExitCode == 0, await output + await error); + } + finally { if (!process.HasExited) { process.Kill(true); await process.WaitForExitAsync(); } } + } + + private sealed class VoiceOracleTheoryAttribute : TheoryAttribute + { + public VoiceOracleTheoryAttribute() + { + if (string.IsNullOrEmpty(Environment.GetEnvironmentVariable("VOICECAT_VOICE_ORACLE"))) Skip = "Set VOICECAT_VOICE_ORACLE to the native voice conformance executable."; + } + } + + [Fact] + public async Task DisconnectInvalidatesBothBindingAndActiveStreams() + { + await using var fixture = new ServerFixture(); + await using var alice = await VoicePeer.ConnectAsync(fixture, "Alice"); + await using var bob = await VoicePeer.ConnectAsync(fixture, "Bob"); + var stream = await alice.AnnounceAsync(StreamKind.StreamMic); + alice.Client.Send(new() { Disconnect = new() }); + await bob.Client.ReadUntilAsync(e => e.UserEvent?.Kind == UserEvent.Types.Kind.Left && e.UserEvent.LeftId == alice.Client.Authentication!.Self.Id); + await alice.SendAsync(alice.Seal(stream.Ssrc, [1])); + await bob.AssertNoVoiceAsync(); + } + + [Fact] + public async Task AnnounceRequiresSubscriptionAndUsesAuthoritativeMusicSettings() + { + await using var fixture = new ServerFixture(); + await using var client = await fixture.ConnectAsync(); + await client.LoginAsync("Alice"); + client.Send(new() { StreamAnnounce = new() { Kind = StreamKind.StreamMic } }); + Assert.False((await client.ReadUntilAsync(e => e.StreamAnnounceResult is not null)).StreamAnnounceResult.Ok); + client.Send(new() { SubscribeVoice = new() }); + await client.ReadUntilAsync(e => e.VoiceSubscriptionResult is not null); + client.Send(new() { JoinChannel = new() { ChannelId = 2 } }); + await client.ReadUntilAsync(e => e.JoinChannelResult is not null); + client.Send(new() { RequestId = 21, StreamAnnounce = new() { Kind = StreamKind.StreamScreenAudio, RequestedAudio = new() { SampleRate = 8000, BitrateBps = 64000 } } }); + var announced = await client.ReadUntilAsync(e => e.StreamAnnounceResult is not null); + Assert.Equal(21UL, announced.RequestId); + Assert.True(announced.StreamAnnounceResult.Ok); + Assert.Equal(64000U, announced.StreamAnnounceResult.EffectiveAudio.BitrateBps); + Assert.Equal(48000U, announced.StreamAnnounceResult.EffectiveAudio.SampleRate); + Assert.Equal(ChannelMode.ModeStereo, announced.StreamAnnounceResult.EffectiveAudio.Mode); + client.Send(new() { StreamAnnounce = new() { Kind = (StreamKind)99 } }); + Assert.False((await client.ReadUntilAsync(e => e.StreamAnnounceResult is not null)).StreamAnnounceResult.Ok); + } + + [Fact] + public async Task EncryptedOpusIsResealedWithRecipientCountersAcrossMultipleStreamsAndSenders() + { + await using var fixture = new ServerFixture(); + await using var alice = await VoicePeer.ConnectAsync(fixture, "Alice"); + await using var bob = await VoicePeer.ConnectAsync(fixture, "Bob"); + await using var carol = await VoicePeer.ConnectAsync(fixture, "Carol"); + StreamAnnounceResult mic = await alice.AnnounceAsync(StreamKind.StreamMic); + StreamAnnounceResult screen = await alice.AnnounceAsync(StreamKind.StreamScreenAudio); + StreamAnnounceResult other = await carol.AnnounceAsync(StreamKind.StreamMic); + Assert.NotEqual(mic.StreamId, screen.StreamId); + Assert.NotEqual(mic.Ssrc, screen.Ssrc); + Assert.Equal(48000U, screen.EffectiveAudio.SampleRate); + Assert.Equal(24000U, screen.EffectiveAudio.BitrateBps); + using var encoder = new Codec.OpusEncoder(new()); + short[] samples = Enumerable.Range(0, 960).Select(i => (short)(8000 * Math.Sin(i * 0.1))).ToArray(); + byte[] payload = new byte[4000]; + int length = encoder.Encode(samples, payload); + payload = payload[..length]; + foreach (var (sender, stream) in new[] { (alice, mic), (carol, other), (alice, screen) }) + { + byte[] packet = sender.Seal(stream.Ssrc, payload, 960, VoiceFrameFlags.Marker | VoiceFrameFlags.FecPresent); + await sender.SendAsync(packet); + var received = await bob.ReceiveVoiceAsync(); + Assert.Equal(payload, received.Payload); + Assert.Equal(stream.Ssrc, received.Header.Ssrc); + Assert.Equal(960U, received.Header.Timestamp); + Assert.Equal(VoiceFrameFlags.Marker | VoiceFrameFlags.FecPresent, received.Header.Flags); + } + Assert.Equal(2UL, bob.LastSequence); + } + + [Fact] + public async Task ReplayForgeryAndSpoofedStreamsAreDroppedWithoutBreakingValidMedia() + { + await using var fixture = new ServerFixture(); + await using var alice = await VoicePeer.ConnectAsync(fixture, "Alice"); + await using var bob = await VoicePeer.ConnectAsync(fixture, "Bob"); + StreamAnnounceResult stream = await alice.AnnounceAsync(StreamKind.StreamMic); + byte[] packet = alice.Seal(stream.Ssrc, [1, 2, 3]); + await alice.SendAsync(packet); + Assert.Equal(new byte[] { 1, 2, 3 }, (await bob.ReceiveVoiceAsync()).Payload); + await alice.SendAsync(packet); + byte[] forged = alice.Seal(stream.Ssrc, [4]); + forged[^1] ^= 1; + await alice.SendAsync(forged); + await alice.SendAsync(alice.Seal(stream.Ssrc + 1000, [5])); + await alice.SendAsync([1]); + await alice.SendAsync(alice.Seal(stream.Ssrc, [6])); + Assert.Equal(new byte[] { 6 }, (await bob.ReceiveVoiceAsync()).Payload); + Assert.Equal(1UL, bob.LastSequence); + } + + [Fact] + public async Task SubscriptionChannelMovementAndStreamStopIsolateMedia() + { + await using var fixture = new ServerFixture(); + await using var alice = await VoicePeer.ConnectAsync(fixture, "Alice"); + await using var bob = await VoicePeer.ConnectAsync(fixture, "Bob"); + var mic = await alice.AnnounceAsync(StreamKind.StreamMic); + await bob.SubscribeAsync(false); + await alice.SendAsync(alice.Seal(mic.Ssrc, [1])); + await bob.AssertNoVoiceAsync(); + await bob.SubscribeAsync(true); + bob.Client.Send(new() { JoinChannel = new() { ChannelId = 2 } }); + Assert.True((await bob.Client.ReadUntilAsync(e => e.JoinChannelResult is not null)).JoinChannelResult.Ok); + await alice.SendAsync(alice.Seal(mic.Ssrc, [2])); + await bob.AssertNoVoiceAsync(); + bob.Client.Send(new() { JoinChannel = new() { ChannelId = 1 } }); + await bob.Client.ReadUntilAsync(e => e.JoinChannelResult is not null); + alice.Client.Send(new() { StreamStop = new() { StreamId = mic.StreamId } }); + await alice.Client.ReadUntilAsync(e => e.UserEvent?.User?.Id == alice.Client.Authentication!.Self.Id && e.UserEvent.User.Streams.Count == 0); + await alice.SendAsync(alice.Seal(mic.Ssrc, [3])); + await bob.AssertNoVoiceAsync(); + var replacement = await alice.AnnounceAsync(StreamKind.StreamMic); + await alice.SendAsync(alice.Seal(replacement.Ssrc, [4])); + Assert.Equal(new byte[] { 4 }, (await bob.ReceiveVoiceAsync()).Payload); + } + + [Fact] + public async Task BadTokensCannotBindAndExistingBindingCannotBeStolen() + { + await using var fixture = new ServerFixture(); + await using var alice = await VoicePeer.ConnectAsync(fixture, "Alice"); + using var rogue = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp); + rogue.Bind(new IPEndPoint(IPAddress.Loopback, 0)); + byte[] binding = new byte[VoiceFrameHeader.Size + 16]; + new VoiceFrameHeader(MediaFrameType.UdpBinding, 0, 0, 0, 0, 0).Write(binding); + await rogue.SendToAsync(binding, SocketFlags.None, fixture.Server.MediaEndPoint); + alice.Client.Authentication!.UdpToken.Span.CopyTo(binding.AsSpan(VoiceFrameHeader.Size)); + await rogue.SendToAsync(binding, SocketFlags.None, fixture.Server.MediaEndPoint); + byte[] keepalive = new byte[VoiceFrameHeader.Size]; + new VoiceFrameHeader(MediaFrameType.Keepalive, 0, 0, 0, 0, 0).Write(keepalive); + await rogue.SendToAsync(keepalive, SocketFlags.None, fixture.Server.MediaEndPoint); + using var timeout = new CancellationTokenSource(200); + await Assert.ThrowsAnyAsync(async () => await rogue.ReceiveAsync(new byte[100], SocketFlags.None, timeout.Token)); + await alice.SendAsync(keepalive); + Assert.Equal(keepalive, await alice.ReceivePacketAsync()); + } + + internal sealed class VoicePeer : IAsyncDisposable + { + public Client Client { get; } + private readonly Socket udp = new(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp); + private readonly IPEndPoint endpoint; + private readonly MediaSessionCrypto crypto; + public ulong LastSequence { get; private set; } + private VoicePeer(Client client, IPEndPoint endpoint, MediaSessionCrypto crypto) + { + Client = client; this.endpoint = endpoint; this.crypto = crypto; + udp.Bind(new IPEndPoint(IPAddress.Loopback, 0)); + } + public static async Task ConnectAsync(ServerFixture fixture, string nickname) + { + Client client = await fixture.ConnectAsync(); + await client.LoginAsync(nickname); + var peer = new VoicePeer(client, fixture.Server.MediaEndPoint, await client.TakeMediaCryptoAsync()); + client.Send(new() { UdpBinding = new() { UdpToken = client.Authentication!.UdpToken } }); + Assert.True((await client.ReadUntilAsync(e => e.UdpBinding is not null)).UdpBinding.Ack); + byte[] binding = new byte[VoiceFrameHeader.Size + 16]; + new VoiceFrameHeader(MediaFrameType.UdpBinding, 0, 0, 0, 0, 0).Write(binding); + client.Authentication.UdpToken.Span.CopyTo(binding.AsSpan(VoiceFrameHeader.Size)); + await peer.SendAsync(binding); + byte[] keepalive = new byte[VoiceFrameHeader.Size]; + new VoiceFrameHeader(MediaFrameType.Keepalive, 0, 0, 0, 0, 0).Write(keepalive); + await peer.SendAsync(keepalive); + Assert.Equal(keepalive, await peer.ReceivePacketAsync()); + await peer.SubscribeAsync(true); + return peer; + } + public async Task SubscribeAsync(bool subscribed) + { + Client.Send(subscribed ? new() { SubscribeVoice = new() } : new() { UnsubscribeVoice = new() }); + var result = (await Client.ReadUntilAsync(e => e.VoiceSubscriptionResult is not null)).VoiceSubscriptionResult; + Assert.True(result.Ok); Assert.Equal(subscribed, result.Subscribed); + } + public async Task AnnounceAsync(StreamKind kind) + { + Client.Send(new() { StreamAnnounce = new() { Kind = kind, RequestedAudio = new() { SampleRate = 8000, BitrateBps = 900000 } } }); + var result = (await Client.ReadUntilAsync(e => e.StreamAnnounceResult is not null)).StreamAnnounceResult; + Assert.True(result.Ok, result.Error); + return result; + } + public byte[] Seal(uint ssrc, byte[] payload, uint timestamp = 0, VoiceFrameFlags flags = 0) + { + byte[] packet = new byte[VoiceFrameHeader.Size + payload.Length + 16]; + crypto.Encryptor.Encrypt(new(MediaFrameType.Voice, flags, 0, ssrc, 0, timestamp), payload, packet); + return packet; + } + public async Task SendAsync(byte[] packet) => await udp.SendToAsync(packet, SocketFlags.None, endpoint, Client.Timeout.Token); + public async Task ReceivePacketAsync() + { + byte[] buffer = new byte[65535]; + int size = await udp.ReceiveAsync(buffer, SocketFlags.None, Client.Timeout.Token); + return buffer[..size]; + } + public async Task<(VoiceFrameHeader Header, byte[] Payload)> ReceiveVoiceAsync() + { + byte[] packet = await ReceivePacketAsync(); + byte[] plain = new byte[65535]; + Assert.True(crypto.Decryptor.TryDecrypt(packet, plain, out var header, out int length)); + LastSequence = header.Sequence; + return (header, plain[..length]); + } + public async Task AssertNoVoiceAsync() + { + using var timeout = new CancellationTokenSource(200); + await Assert.ThrowsAnyAsync(async () => await udp.ReceiveAsync(new byte[65535], SocketFlags.None, timeout.Token)); + } + public async ValueTask DisposeAsync() { udp.Dispose(); crypto.Dispose(); await Client.DisposeAsync(); } + } +} diff --git a/dotnet/tests/VoiceCat.Tests/ServerTests.cs b/dotnet/tests/VoiceCat.Tests/ServerTests.cs index 8e344c9..30b066a 100644 --- a/dotnet/tests/VoiceCat.Tests/ServerTests.cs +++ b/dotnet/tests/VoiceCat.Tests/ServerTests.cs @@ -131,7 +131,7 @@ public sealed class ServerTests } } - private sealed class ServerFixture : IAsyncDisposable + internal sealed class ServerFixture : IAsyncDisposable { public string Directory { get; } = Path.Combine(Path.GetTempPath(), "voicecat-server-" + Guid.NewGuid().ToString("N")); public VoiceServer Server { get; } @@ -156,7 +156,7 @@ public sealed class ServerTests } } - private sealed class Client : IAsyncDisposable + internal sealed class Client : IAsyncDisposable { public CancellationTokenSource Timeout { get; } = new(TimeSpan.FromSeconds(30)); private readonly TlsControlConnection connection; @@ -167,6 +167,8 @@ public sealed class ServerTests messages = connection.ReadAsync(Timeout.Token).GetAsyncEnumerator(); } public void Send(Envelope envelope) => Assert.True(connection.TrySend(envelope)); + public AuthResult? Authentication { get; private set; } + public Task TakeMediaCryptoAsync() => connection.TakeMediaCryptoAsync(Timeout.Token); public async Task ReadUntilAsync(Func predicate) { while (await messages.MoveNextAsync()) if (predicate(messages.Current)) return messages.Current; @@ -178,6 +180,7 @@ public sealed class ServerTests Assert.Equal(1UL, (await ReadUntilAsync(e => e.ServerHello is not null)).RequestId); Send(new() { RequestId = 2, AuthRequest = new() { Guest = new() { Nickname = nickname } } }); AuthResult auth = (await ReadUntilAsync(e => e.AuthResult is not null)).AuthResult; + Authentication = auth; Assert.True(auth.Ok, auth.Error); ServerStateSnapshot state = (await ReadUntilAsync(e => e.ServerState is not null)).ServerState; Assert.Equal(2, state.Channels.Count); diff --git a/tools/vccli/src/main.cpp b/tools/vccli/src/main.cpp index e247bce..d313517 100644 --- a/tools/vccli/src/main.cpp +++ b/tools/vccli/src/main.cpp @@ -6,6 +6,7 @@ */ #include #include +#include #include #include #include @@ -13,6 +14,7 @@ #include #include #include +#include #include "voicecat.h" @@ -24,6 +26,11 @@ void on_sigint(int) { g_stop.store(true); } struct Stats { std::atomic auth_done{false}; std::atomic auth_ok{false}; + std::atomic self_id{0}; + std::atomic voice_subscribed{false}; + std::atomic own_stream_ready{false}; + std::atomic pcm_frames{0}; + std::atomic pcm_energy{0}; // Set right after vc_client_create, before vc_connect — lets on_event auto-confirm the // TOFU gate (VC_EVENT_SERVER_IDENTITY below). vccli has no interactive prompt, so it // trusts-on-first-connect unconditionally (prints the fingerprint for visibility). @@ -52,6 +59,7 @@ void on_event(void* user, const vc_event* ev) { vc_confirm_server_identity(st->client, 1); break; case VC_EVENT_AUTH_RESULT: + st->self_id = ev->user_id; st->auth_ok = (ev->result == VC_OK); st->auth_done = true; std::printf("[auth] ok=%d user_id=%u %s\n", st->auth_ok.load(), ev->user_id, @@ -71,9 +79,13 @@ void on_event(void* user, const vc_event* ev) { std::printf("[user] updated: id=%u\n", ev->user_id); break; case VC_EVENT_STREAM_STARTED: + if (ev->user_id == st->self_id.load()) st->own_stream_ready = true; std::printf("[voice] stream started: user_id=%u stream_id=%u\n", ev->user_id, ev->stream_id); break; + case VC_EVENT_VOICE_STATE: + st->voice_subscribed = ev->u32a == 1; + break; case VC_EVENT_STREAM_STOPPED: std::printf("[voice] stream stopped: user_id=%u stream_id=%u\n", ev->user_id, ev->stream_id); @@ -124,6 +136,16 @@ bool wait_until(std::atomic& flag, int timeout_ms) { return true; } +void test_pcm_sink(void* context, uint32_t user, uint32_t, const int16_t* pcm, + size_t samples, uint32_t channels, uint32_t rate) { + auto& stats = *static_cast(context); + if (user == stats.self_id.load() || rate != 48000) return; + long long energy = 0; + for (size_t index = 0; index < samples * channels; ++index) energy += std::abs(static_cast(pcm[index])); + stats.pcm_energy.fetch_add(energy); + stats.pcm_frames.fetch_add(1); +} + vc_result wait_generic_result(Stats& st, int baseline_count, int timeout_ms) { auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(timeout_ms); while (st.generic_result_count.load() <= baseline_count) { @@ -202,6 +224,7 @@ void print_usage() { "\n" "Voice / devices:\n" " --voice start a MIC stream and stay connected until Ctrl+C\n" + " --test-tone-ms MS finite headless voice test (500..60000 ms); requires a peer\n" " --mute start with the mic muted (only meaningful with --voice)\n" " --list-devices print input/output devices (vc_list_devices) and exit\n" " --input-device ID use device ID (from --list-devices) for the MIC stream\n" @@ -309,6 +332,7 @@ int main(int argc, char** argv) { bool have_password = false; uint32_t channel_id = 1; bool voice_mode = false; + uint32_t test_tone_ms = 0; bool start_muted = false; bool self_mute = false; bool self_deafen = false; @@ -370,6 +394,12 @@ int main(int argc, char** argv) { else if (a == "--password") { password = next(); have_password = true; } else if (a == "--channel") { if (!parse_u32(next().c_str(), &channel_id, "--channel")) return 1; } else if (a == "--voice") voice_mode = true; + else if (a == "--test-tone-ms") { + if (!parse_u32(next().c_str(), &test_tone_ms, "--test-tone-ms") || test_tone_ms < 500 || test_tone_ms > 60000) { + std::fprintf(stderr, "--test-tone-ms must be between 500 and 60000\n"); return 1; + } + voice_mode = true; + } else if (a == "--mute") start_muted = true; else if (a == "--self-mute") self_mute = true; else if (a == "--self-deafen") self_deafen = true; @@ -441,6 +471,9 @@ int main(int argc, char** argv) { } // Validate auth mode. + if (test_tone_ms && (start_muted || self_mute || self_deafen || share_screen_audio || have_input_device)) { + std::fprintf(stderr, "--test-tone-ms cannot be combined with mute, deafen or device capture\n"); return 1; + } if (have_username != have_password) { std::fprintf(stderr, "--username and --password must be used together\n"); return 1; @@ -490,6 +523,10 @@ int main(int argc, char** argv) { return 1; } st.client = c; + if (test_tone_ms) { + vc_set_external_playback(c, 1); + vc_set_pcm_sink(c, test_pcm_sink, &st); + } if (list_devices) { // Device enumeration works pre-connect (no server needed) — see docs/voice.md. @@ -641,26 +678,54 @@ int main(int argc, char** argv) { } if (voice_mode) { + r = vc_join_voice(c); + if (r != VC_OK || !wait_until(st.voice_subscribed, 8000)) { + std::fprintf(stderr, "voice subscription failed or timed out\n"); + vc_disconnect(c); vc_client_destroy(c); return 1; + } // Give the async UDP binding handshake (TCP UdpBinding -> ack -> plaintext // bootstrap packet) a moment to land before announcing a stream. std::this_thread::sleep_for(std::chrono::milliseconds(500)); if (start_muted) vc_set_self_mute(c, 1, 0); - r = vc_set_input_mode(c, input_mode); + vc_input_mode effective_input_mode = test_tone_ms ? VC_INPUT_ALWAYS_ON : input_mode; + r = vc_set_input_mode(c, effective_input_mode); std::printf("vc_set_input_mode(%s) -> %d (%s)\n", - input_mode == VC_INPUT_PUSH_TO_TALK ? "ptt" : "vad", r, vc_result_string(r)); + effective_input_mode == VC_INPUT_ALWAYS_ON ? "always-on" : input_mode == VC_INPUT_PUSH_TO_TALK ? "ptt" : "vad", r, vc_result_string(r)); vc_stream_desc desc{}; desc.kind = VC_STREAM_MIC; desc.device_id = have_input_device ? input_device.c_str() : nullptr; desc.label = "Microphone"; + desc.external_feed = test_tone_ms ? 1 : 0; uint32_t stream_id = 0; r = vc_stream_start(c, &desc, &stream_id); std::printf("vc_stream_start -> %d (%s), stream_id=%u\n", r, vc_result_string(r), stream_id); + if (test_tone_ms) { + vc_audio_config audio{}; + bool ok = r == VC_OK && wait_until(st.own_stream_ready, 8000) && + vc_get_stream_audio_config(c, st.self_id.load(), stream_id, &audio) == VC_OK; + uint32_t channels = audio.mode == 1 ? 2 : 1; + std::vector pcm(960 * channels); + for (size_t sample = 0; sample < 960; ++sample) + for (uint32_t side = 0; side < channels; ++side) + pcm[sample * channels + side] = static_cast(12000 * std::sin(sample * (side ? 0.083 : 0.058))); + auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(test_tone_ms); + while (ok && !g_stop.load() && std::chrono::steady_clock::now() < deadline) { + ok = vc_stream_feed_pcm(c, stream_id, pcm.data(), 960, channels) == VC_OK; + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + } + vc_stream_stop(c, stream_id); + vc_disconnect(c); + vc_client_destroy(c); + std::printf("[test-tone] received=%d energy=%lld channels=%u\n", st.pcm_frames.load(), st.pcm_energy.load(), channels); + return ok && st.pcm_frames.load() >= 5 && st.pcm_energy.load() > 0 && !mod_request_error ? 0 : 1; + } + if (have_input_device && r == VC_OK) { r = vc_set_input_device(c, stream_id, input_device.c_str()); std::printf("vc_set_input_device -> %d (%s)\n", r, vc_result_string(r));