diff --git a/.gitignore b/.gitignore index 6aa19847..bc0d9b81 100644 --- a/.gitignore +++ b/.gitignore @@ -130,3 +130,6 @@ src/Surface.Rendering/wwwroot/surface-app/ custom/plugins/*/node_modules/ custom/static-plugins/*/node_modules/ +# Archived plugins keep their own app trees; their dependencies are never tracked. +custom/static-plugins/_archive/*/node_modules/ +custom/static-plugins/*/app/*/node_modules/ diff --git a/custom/static-plugins/Communication/registry.json b/custom/static-plugins/Communication/registry.json index 6ca9848e..c8fdb911 100644 --- a/custom/static-plugins/Communication/registry.json +++ b/custom/static-plugins/Communication/registry.json @@ -14,6 +14,9 @@ "communication.video", "communication.webrtc" ], + "sensitiveFields": [ + "remoteParty" + ], "dependencies": { "Callora.Core": ">=0.1.0-local", "Callora.Plugin.Communication.Abstractions": ">=0.1.0-local" diff --git a/custom/static-plugins/Communication/src/Abstractions/src/Voice/AudioFormat.cs b/custom/static-plugins/Communication/src/Abstractions/src/Voice/AudioFormat.cs index eab40432..91c51252 100644 --- a/custom/static-plugins/Communication/src/Abstractions/src/Voice/AudioFormat.cs +++ b/custom/static-plugins/Communication/src/Abstractions/src/Voice/AudioFormat.cs @@ -8,4 +8,12 @@ public sealed record AudioFormat(AudioCodec Codec, int SampleRateHz, int FrameMi { /// SIP/PSTN-Standard: G.711 µ-law, 8 kHz, 20-ms-Frames. public static AudioFormat G711Ulaw8k20ms { get; } = new(AudioCodec.G711Ulaw, 8000, 20); + + /// + /// Exakte Frame-Größe in Bytes. G.711 kodiert ein Sample je Byte, also + /// SampleRateHz × FrameMilliseconds / 1000 — für 8 kHz/20 ms sind das 160 Bytes. + /// Das ausgehandelte Format wird damit prüfbar: eingehende Frames dürfen genau + /// diese Größe haben, statt beliebig groß zu sein (#108). + /// + public int BytesPerFrame => SampleRateHz * FrameMilliseconds / 1000; } diff --git a/custom/static-plugins/Communication/src/Api/WebSocket/WebRtcSignalingChannel.cs b/custom/static-plugins/Communication/src/Api/WebSocket/WebRtcSignalingChannel.cs index fd4eef8e..4fe6e102 100644 --- a/custom/static-plugins/Communication/src/Api/WebSocket/WebRtcSignalingChannel.cs +++ b/custom/static-plugins/Communication/src/Api/WebSocket/WebRtcSignalingChannel.cs @@ -1,5 +1,7 @@ using System.Net.WebSockets; using System.Text; +using Callora.Plugin.Communication.Application.Streaming; +using Callora.Plugin.Communication.Infrastructure.Transport; namespace Callora.Plugin.Communication.Api.WebSocket; @@ -38,28 +40,17 @@ public async ValueTask SendAsync(WebRtcSignalMessage message, CancellationToken /// /// Reads the next whole text message and returns its raw JSON, or when the - /// socket closes. Fragmented frames are reassembled; parsing/validation is the caller's concern so a - /// single malformed frame can be logged and ignored without ending the stream. + /// socket closes. Fragmented frames are reassembled under a hard byte cap and an idle timeout + /// (#108); parsing/validation is the caller's concern so a single malformed frame can be logged + /// and ignored without ending the stream. /// - public async ValueTask ReceiveTextAsync(CancellationToken cancellationToken = default) - { - using var message = new MemoryStream(); - - WebSocketReceiveResult result; - do - { - result = await socket.ReceiveAsync(_receiveBuffer, cancellationToken).ConfigureAwait(false); - if (result.MessageType == WebSocketMessageType.Close) - { - return null; - } - - message.Write(_receiveBuffer, 0, result.Count); - } - while (!result.EndOfMessage); - - return Encoding.UTF8.GetString(message.GetBuffer(), 0, (int)message.Length); - } + public ValueTask ReceiveTextAsync(CancellationToken cancellationToken = default) => + BoundedWebSocketReader.ReadTextAsync( + socket, + _receiveBuffer, + CommunicationStreamLimits.MaxSignalingMessageBytes, + CommunicationStreamLimits.IdleTimeout, + cancellationToken); /// public void Dispose() => _sendLock.Dispose(); diff --git a/custom/static-plugins/Communication/src/Application/Admin/Calls/CallAdminScope.cs b/custom/static-plugins/Communication/src/Application/Admin/Calls/CallAdminScope.cs index 002540e8..77e1e0be 100644 --- a/custom/static-plugins/Communication/src/Application/Admin/Calls/CallAdminScope.cs +++ b/custom/static-plugins/Communication/src/Application/Admin/Calls/CallAdminScope.cs @@ -3,11 +3,12 @@ namespace Callora.Plugin.Communication.Application.Admin.Calls; /// -/// Resolves the workspace a call-control admin request operates on. The caller's token-bound workspace -/// (, set authoritatively by the host) always wins, so a -/// workspace-scoped operator can never reach another workspace. A platform operator (no bound -/// workspace) must name the target explicitly via ?workspaceKey=; absent that the request is -/// rejected rather than defaulting to something dangerous. Mirrors the SIP-account scope helper. +/// Reads the workspace a call-control admin request operates on. The host resolves it +/// authoritatively into — the caller's bound +/// workspace when it has one, otherwise the workspace a platform operator named explicitly — +/// and has already confirmed the plugin is available there (#109). The plugin therefore +/// never reads a workspace from the query itself; that would bypass the host's gate. +/// Mirrors the SIP-account scope helper. /// internal static class CallAdminScope { @@ -16,22 +17,16 @@ public static bool TryResolve( out string workspaceKey, out HostAdminApiResponse? error) { - var resolved = request.WorkspaceKey; - if (string.IsNullOrWhiteSpace(resolved) && - request.Query.TryGetValue("workspaceKey", out var values) && - values.Length > 0) - { - resolved = values[0]; - } + ArgumentNullException.ThrowIfNull(request); - if (string.IsNullOrWhiteSpace(resolved)) + if (string.IsNullOrWhiteSpace(request.WorkspaceKey)) { workspaceKey = string.Empty; error = new HostAdminApiResponse(400, new { error = "A workspace is required." }); return false; } - workspaceKey = resolved.Trim(); + workspaceKey = request.WorkspaceKey.Trim(); error = null; return true; } diff --git a/custom/static-plugins/Communication/src/Application/Admin/CommunicationAdminApiExtensionContributor.cs b/custom/static-plugins/Communication/src/Application/Admin/CommunicationAdminApiExtensionContributor.cs index d684b7a3..5b149bbb 100644 --- a/custom/static-plugins/Communication/src/Application/Admin/CommunicationAdminApiExtensionContributor.cs +++ b/custom/static-plugins/Communication/src/Application/Admin/CommunicationAdminApiExtensionContributor.cs @@ -24,11 +24,15 @@ public CommunicationAdminApiExtensionContributor(IReadOnlyList public async ValueTask HandleAsync( @@ -53,6 +58,13 @@ public async ValueTask HandleAsync( return Bad("displayName is required."); } + // Refuse an authentication method the provider cannot connect before anything is + // persisted (#111) — accepting it would create an account that is silently skipped. + if (SipAuthMethodValidation.Reject(body.AuthMethod) is { } unsupported) + { + return unsupported; + } + if (!_connectionFactory.TryBuild(body, existing: null, out var connection, out var error)) { return Bad(error!); @@ -73,7 +85,10 @@ public async ValueTask HandleAsync( body.Enabled ?? true); await store.AddAsync(account, cancellationToken).ConfigureAwait(false); - return new HostAdminApiResponse(201, SipAccountResponse.FromDomain(account)); + + // A created-and-enabled account must register now, not at the next restart (#110). + var runtimeFailure = await _runtime.ReconcileAsync(account, cancellationToken).ConfigureAwait(false); + return runtimeFailure ?? new HostAdminApiResponse(201, SipAccountResponse.FromDomain(account)); } private static HostAdminApiResponse Bad(string message) => new(400, new { error = message }); diff --git a/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/DeleteSipAccountRouteHandler.cs b/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/DeleteSipAccountRouteHandler.cs index a0efbfa8..884f81f1 100644 --- a/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/DeleteSipAccountRouteHandler.cs +++ b/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/DeleteSipAccountRouteHandler.cs @@ -1,11 +1,17 @@ using Callora.Core.Application.Plugins.Contracts; using Callora.Plugin.Communication.Application.Accounts; +using Callora.Plugin.Communication.Application.Voice; namespace Callora.Plugin.Communication.Application.Admin.SipAccounts; /// Handles DELETE sip-accounts/{accountId} — removes an account of the caller's workspace. -public sealed class DeleteSipAccountRouteHandler(ISipAccountStore store) : IHostAdminApiRouteHandler +public sealed class DeleteSipAccountRouteHandler( + ISipAccountStore store, + ISipAccountRuntimeReconciler? reconciler = null) : IHostAdminApiRouteHandler { + private readonly SipAccountRuntimeCoordinator _runtime = + new(store, reconciler, TimeProvider.System); + /// public async ValueTask HandleAsync( HostAdminApiRequest request, @@ -19,6 +25,12 @@ public async ValueTask HandleAsync( var accountId = request.RouteValues.TryGetValue("accountId", out var value) ? value : string.Empty; var deleted = await store.DeleteAsync(workspaceKey, accountId, cancellationToken).ConfigureAwait(false); + if (deleted) + { + // Deregister before returning: a deleted account must not keep a live + // registration until the next restart (#110). + await _runtime.RemoveAsync(workspaceKey, accountId, cancellationToken).ConfigureAwait(false); + } return deleted ? new HostAdminApiResponse(204) diff --git a/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/SetSipAccountEnabledRouteHandler.cs b/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/SetSipAccountEnabledRouteHandler.cs index abb3d069..8ae7a200 100644 --- a/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/SetSipAccountEnabledRouteHandler.cs +++ b/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/SetSipAccountEnabledRouteHandler.cs @@ -1,5 +1,6 @@ using Callora.Core.Application.Plugins.Contracts; using Callora.Plugin.Communication.Application.Accounts; +using Callora.Plugin.Communication.Application.Voice; namespace Callora.Plugin.Communication.Application.Admin.SipAccounts; @@ -8,8 +9,15 @@ namespace Callora.Plugin.Communication.Application.Admin.SipAccounts; /// is provisioned. One handler class, registered once per target state, so enable and disable share the /// same guarded lookup and persistence. /// -public sealed class SetSipAccountEnabledRouteHandler(ISipAccountStore store, bool enabled) : IHostAdminApiRouteHandler +public sealed class SetSipAccountEnabledRouteHandler( + ISipAccountStore store, + bool enabled, + ISipAccountRuntimeReconciler? reconciler = null, + TimeProvider? timeProvider = null) : IHostAdminApiRouteHandler { + private readonly SipAccountRuntimeCoordinator _runtime = + new(store, reconciler, timeProvider ?? TimeProvider.System); + /// public async ValueTask HandleAsync( HostAdminApiRequest request, @@ -38,6 +46,10 @@ public async ValueTask HandleAsync( } await store.UpdateAsync(account, cancellationToken).ConfigureAwait(false); - return new HostAdminApiResponse(200, SipAccountResponse.FromDomain(account)); + + // Enabling registers now; disabling deregisters now, so a disabled account stops + // taking calls immediately rather than at the next restart (#110). + var runtimeFailure = await _runtime.ReconcileAsync(account, cancellationToken).ConfigureAwait(false); + return runtimeFailure ?? new HostAdminApiResponse(200, SipAccountResponse.FromDomain(account)); } } diff --git a/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/SipAccountAdminRoutes.cs b/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/SipAccountAdminRoutes.cs index 474359be..0b39a263 100644 --- a/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/SipAccountAdminRoutes.cs +++ b/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/SipAccountAdminRoutes.cs @@ -1,6 +1,7 @@ using Callora.Core.Application.Plugins.Contracts; using Callora.Core.Application.Secrets.Contracts; using Callora.Plugin.Communication.Application.Accounts; +using Callora.Plugin.Communication.Application.Voice; namespace Callora.Plugin.Communication.Application.Admin.SipAccounts; @@ -13,10 +14,16 @@ namespace Callora.Plugin.Communication.Application.Admin.SipAccounts; public static class SipAccountAdminRoutes { /// Creates the route registrations bound to the given store, data protector and plugin id. + /// + /// Brings the live runtime in line after every successful mutation (#110). Null in a + /// deployment that operates no voice runtime, in which case the routes are pure + /// persistence — which is then the truthful behaviour. + /// public static IReadOnlyList Build( ISipAccountStore store, IPluginDataProtector dataProtector, - string pluginId) + string pluginId, + ISipAccountRuntimeReconciler? reconciler = null) { ArgumentNullException.ThrowIfNull(store); ArgumentNullException.ThrowIfNull(dataProtector); @@ -32,19 +39,19 @@ public static IReadOnlyList Build( new GetSipAccountRouteHandler(store)), new HostAdminApiRouteRegistration( "POST", "sip-accounts", CommunicationPermissionKeys.AccountsManage, - new CreateSipAccountRouteHandler(store, dataProtector, pluginId)), + new CreateSipAccountRouteHandler(store, dataProtector, pluginId, reconciler)), new HostAdminApiRouteRegistration( "PUT", "sip-accounts/{accountId}", CommunicationPermissionKeys.AccountsManage, - new UpdateSipAccountRouteHandler(store, dataProtector, pluginId)), + new UpdateSipAccountRouteHandler(store, dataProtector, pluginId, reconciler)), new HostAdminApiRouteRegistration( "POST", "sip-accounts/{accountId}/enable", CommunicationPermissionKeys.AccountsManage, - new SetSipAccountEnabledRouteHandler(store, enabled: true)), + new SetSipAccountEnabledRouteHandler(store, enabled: true, reconciler)), new HostAdminApiRouteRegistration( "POST", "sip-accounts/{accountId}/disable", CommunicationPermissionKeys.AccountsManage, - new SetSipAccountEnabledRouteHandler(store, enabled: false)), + new SetSipAccountEnabledRouteHandler(store, enabled: false, reconciler)), new HostAdminApiRouteRegistration( "DELETE", "sip-accounts/{accountId}", CommunicationPermissionKeys.AccountsManage, - new DeleteSipAccountRouteHandler(store)), + new DeleteSipAccountRouteHandler(store, reconciler)), ]; } } diff --git a/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/SipAccountAdminScope.cs b/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/SipAccountAdminScope.cs index 41dd4890..f90eee3d 100644 --- a/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/SipAccountAdminScope.cs +++ b/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/SipAccountAdminScope.cs @@ -3,11 +3,11 @@ namespace Callora.Plugin.Communication.Application.Admin.SipAccounts; /// -/// Resolves the workspace a SIP-account admin request operates on. The caller's token-bound workspace -/// (, set authoritatively by the host) always wins, so a -/// workspace-scoped operator can never reach another workspace. A platform operator (no bound -/// workspace) must name the target explicitly via ?workspaceKey=; absent that, the request is -/// rejected rather than defaulting to something dangerous. +/// Reads the workspace a SIP-account admin request operates on. The host resolves it +/// authoritatively into — the caller's bound +/// workspace when it has one, otherwise the workspace a platform operator named explicitly — +/// and has already confirmed the plugin is available there (#109). The plugin therefore +/// never reads a workspace from the query itself; that would bypass the host's gate. /// internal static class SipAccountAdminScope { @@ -16,22 +16,16 @@ public static bool TryResolve( out string workspaceKey, out HostAdminApiResponse? error) { - var resolved = request.WorkspaceKey; - if (string.IsNullOrWhiteSpace(resolved) && - request.Query.TryGetValue("workspaceKey", out var values) && - values.Length > 0) - { - resolved = values[0]; - } + ArgumentNullException.ThrowIfNull(request); - if (string.IsNullOrWhiteSpace(resolved)) + if (string.IsNullOrWhiteSpace(request.WorkspaceKey)) { workspaceKey = string.Empty; error = new HostAdminApiResponse(400, new { error = "A workspace is required." }); return false; } - workspaceKey = resolved.Trim(); + workspaceKey = request.WorkspaceKey.Trim(); error = null; return true; } diff --git a/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/SipAccountRuntimeCoordinator.cs b/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/SipAccountRuntimeCoordinator.cs new file mode 100644 index 00000000..e6d96c84 --- /dev/null +++ b/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/SipAccountRuntimeCoordinator.cs @@ -0,0 +1,69 @@ +using Callora.Core.Application.Plugins.Contracts; +using Callora.Plugin.Communication.Abstractions; +using Callora.Plugin.Communication.Application.Accounts; +using Callora.Plugin.Communication.Application.Voice; +using Callora.Plugin.Communication.Domain.Accounts; + +namespace Callora.Plugin.Communication.Application.Admin.SipAccounts; + +/// +/// Couples an admin mutation to the live runtime (#110): persist, reconcile, and let the +/// response tell the truth about both. +/// +/// Before this existed the handlers only wrote rows, so a created account did not register +/// until the next restart and a disabled one kept taking calls. Now every successful write is +/// followed by a reconciliation, and a runtime failure is written back onto the account as a +/// status with its reason — so the persisted state and +/// the response agree with what the runtime actually did. +/// +/// +/// Deployments without a voice runtime (no SDK client configured) pass a null reconciler; the +/// handlers then behave as pure persistence, which is the honest behaviour for a host that +/// operates no channels. +/// +/// +internal sealed class SipAccountRuntimeCoordinator( + ISipAccountStore store, + ISipAccountRuntimeReconciler? reconciler, + TimeProvider timeProvider) +{ + /// + /// Reconciles after it was persisted. On failure the account's + /// status is written back as failed and the caller receives a 502 carrying both the + /// reason and the account, so an operator sees the configuration that exists and + /// that it is not live. Returns null when the runtime matches the desired state. + /// + public async Task ReconcileAsync( + SipAccount account, + CancellationToken cancellationToken) + { + if (reconciler is null) + { + return null; + } + + var result = await reconciler.ApplyAsync(account, cancellationToken).ConfigureAwait(false); + if (result.IsSuccess) + { + return null; + } + + account.ReportStatus(SipAccountStatus.Failed, result.Error, timeProvider.GetUtcNow()); + await store.UpdateAsync(account, cancellationToken).ConfigureAwait(false); + + return new HostAdminApiResponse(502, new + { + error = result.Error, + account = SipAccountResponse.FromDomain(account) + }); + } + + /// + /// Removes the account from the runtime. Deprovisioning is local teardown — deregister the + /// channel, stop new calls — so it cannot fail in a way the caller could act on. + /// + public Task RemoveAsync(string workspaceKey, string accountId, CancellationToken cancellationToken) => + reconciler is null + ? Task.CompletedTask + : reconciler.RemoveAsync(workspaceKey, accountId, cancellationToken); +} diff --git a/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/SipAuthMethodValidation.cs b/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/SipAuthMethodValidation.cs new file mode 100644 index 00000000..d57008f7 --- /dev/null +++ b/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/SipAuthMethodValidation.cs @@ -0,0 +1,37 @@ +using Callora.Core.Application.Plugins.Contracts; +using Callora.Plugin.Communication.Application.Voice; +using Callora.Plugin.Communication.Domain.Accounts; + +namespace Callora.Plugin.Communication.Application.Admin.SipAccounts; + +/// +/// Refuses SIP accounts the voice provider cannot connect, at the edge (#111). +/// +/// 422 Unprocessable Content rather than 400: the request is well-formed and the +/// authentication method is a legitimate SIP deployment — this platform just cannot operate it +/// yet. The distinction matters to a client, because a 400 says "fix your request" while a 422 +/// says "this is understood but unsupported", and the message names the upstream gap. +/// +/// +internal static class SipAuthMethodValidation +{ + /// + /// Returns the refusal response when cannot be connected, or null + /// when the account may be created or updated. + /// + public static HostAdminApiResponse? Reject(SipAuthMethod? method) + { + var effective = method ?? SipAuthMethod.Digest; + if (SipAuthMethodSupport.DescribeUnsupported(effective) is not { } reason) + { + return null; + } + + return new HostAdminApiResponse(422, new + { + error = reason, + authMethod = effective.ToString(), + supportedAuthMethods = SipAuthMethodSupport.Supported.Select(x => x.ToString()).ToArray() + }); + } +} diff --git a/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/UpdateSipAccountRouteHandler.cs b/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/UpdateSipAccountRouteHandler.cs index 20e4f813..d5274ce8 100644 --- a/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/UpdateSipAccountRouteHandler.cs +++ b/custom/static-plugins/Communication/src/Application/Admin/SipAccounts/UpdateSipAccountRouteHandler.cs @@ -3,6 +3,7 @@ using Callora.Core.Application.Plugins.Contracts; using Callora.Core.Application.Secrets.Contracts; using Callora.Plugin.Communication.Application.Accounts; +using Callora.Plugin.Communication.Application.Voice; namespace Callora.Plugin.Communication.Application.Admin.SipAccounts; @@ -16,12 +17,16 @@ namespace Callora.Plugin.Communication.Application.Admin.SipAccounts; public sealed class UpdateSipAccountRouteHandler( ISipAccountStore store, IPluginDataProtector dataProtector, - string pluginId) : IHostAdminApiRouteHandler + string pluginId, + ISipAccountRuntimeReconciler? reconciler = null, + TimeProvider? timeProvider = null) : IHostAdminApiRouteHandler { private static readonly JsonSerializerOptions SerializerOptions = new(JsonSerializerDefaults.Web) { Converters = { new JsonStringEnumConverter() } }; private readonly SipAccountConnectionFactory _connectionFactory = new(dataProtector, pluginId); + private readonly SipAccountRuntimeCoordinator _runtime = + new(store, reconciler, timeProvider ?? TimeProvider.System); /// public async ValueTask HandleAsync( @@ -61,6 +66,14 @@ public async ValueTask HandleAsync( return Bad("displayName is required."); } + // An update must not move an account onto an unsupported method either (#111). An + // omitted method keeps the stored one, so an already-unsupported account can still be + // edited towards a supported configuration. + if (SipAuthMethodValidation.Reject(body.AuthMethod ?? account.Connection.Authentication.Method) is { } unsupported) + { + return unsupported; + } + // Reuse the existing authentication so omitted secrets are kept rather than dropped. if (!_connectionFactory.TryBuild(body, account.Connection.Authentication, out var connection, out var error)) { @@ -75,7 +88,10 @@ public async ValueTask HandleAsync( account.Reconfigure(body.DisplayName!, connection!, maxConcurrentCalls); await store.UpdateAsync(account, cancellationToken).ConfigureAwait(false); - return new HostAdminApiResponse(200, SipAccountResponse.FromDomain(account)); + + // Credential, endpoint or capacity changes reconnect the live channel (#110). + var runtimeFailure = await _runtime.ReconcileAsync(account, cancellationToken).ConfigureAwait(false); + return runtimeFailure ?? new HostAdminApiResponse(200, SipAccountResponse.FromDomain(account)); } private static HostAdminApiResponse Bad(string message) => new(400, new { error = message }); diff --git a/custom/static-plugins/Communication/src/Application/Streaming/CommunicationStreamLimits.cs b/custom/static-plugins/Communication/src/Application/Streaming/CommunicationStreamLimits.cs new file mode 100644 index 00000000..ec5ffe50 --- /dev/null +++ b/custom/static-plugins/Communication/src/Application/Streaming/CommunicationStreamLimits.cs @@ -0,0 +1,53 @@ +namespace Callora.Plugin.Communication.Application.Streaming; + +/// +/// Hard resource bounds for the plugin's WebSocket surfaces (#108). A valid ticket +/// holder is still an untrusted peer: without these, one connection can grow host +/// memory without limit through fragmented messages, oversized audio payloads or a +/// producer that outruns the paced sender. +/// +/// The values are deliberately generous against legitimate traffic and tiny against +/// abuse: a 20 ms µ-law frame is 160 bytes (~216 base64), and an SDP offer with +/// bundled ICE candidates stays far below 64 KiB. +/// +/// +public static class CommunicationStreamLimits +{ + /// + /// Largest media-protocol message accepted, across all fragments. Exceeding it + /// aborts the connection rather than truncating — a peer sending more is either + /// broken or hostile. + /// + public const int MaxMediaMessageBytes = 64 * 1024; + + /// Largest signalling message accepted, across all fragments. + public const int MaxSignalingMessageBytes = 64 * 1024; + + /// + /// Largest decoded audio frame accepted. Well above any 20–60 ms frame in the + /// supported formats, and far below anything that could pressure memory. + /// + public const int MaxAudioFrameBytes = 8 * 1024; + + /// + /// Total bytes the paced outbound buffer may hold. Bounds the queue by size as + /// well as by frame count, so many small frames cannot bypass the count cap. + /// + public const int MaxPacedBufferBytes = 512 * 1024; + + /// + /// How long a socket may stay silent before it is torn down. Frees sockets that a + /// peer opened and abandoned, which would otherwise hold their buffers forever. + /// + public static readonly TimeSpan IdleTimeout = TimeSpan.FromSeconds(60); + + /// How long a connect token stays redeemable after the session is minted. + public static readonly TimeSpan ConnectTokenTimeToLive = TimeSpan.FromMinutes(2); + + /// + /// Sessions are removed this long after they closed or their token expired. Keeps + /// spent tickets from accumulating and shrinks the window in which a leaked row is + /// worth anything. + /// + public static readonly TimeSpan SessionRetention = TimeSpan.FromHours(24); +} diff --git a/custom/static-plugins/Communication/src/Application/Streaming/IMediaStreamSessionStore.cs b/custom/static-plugins/Communication/src/Application/Streaming/IMediaStreamSessionStore.cs index a0eb2485..ccb8af88 100644 --- a/custom/static-plugins/Communication/src/Application/Streaming/IMediaStreamSessionStore.cs +++ b/custom/static-plugins/Communication/src/Application/Streaming/IMediaStreamSessionStore.cs @@ -25,6 +25,14 @@ public interface IMediaStreamSessionStore /// Resolves a workspace-scoped session by id. Task GetAsync(string workspaceKey, string sessionId, CancellationToken cancellationToken = default); + /// + /// Deletes sessions that closed, or whose ticket has been unusable, for longer than + /// . Returns the count. Spent and expired tickets must not + /// accumulate (#108). + /// + Task PurgeExpiredAsync( + DateTimeOffset now, TimeSpan retention, CancellationToken cancellationToken = default); + /// Deletes all sessions of a workspace (used by the GDPR purge contributor). Returns the count. Task DeleteByWorkspaceAsync(string workspaceKey, CancellationToken cancellationToken = default); } diff --git a/custom/static-plugins/Communication/src/Application/Streaming/MediaBridge.cs b/custom/static-plugins/Communication/src/Application/Streaming/MediaBridge.cs index 62590768..258bd35e 100644 --- a/custom/static-plugins/Communication/src/Application/Streaming/MediaBridge.cs +++ b/custom/static-plugins/Communication/src/Application/Streaming/MediaBridge.cs @@ -103,7 +103,10 @@ private async Task PumpConsumerToCallAsync(PacedAudioSender pacer, CancellationT switch (message.Event) { - case MediaStreamEventType.Media when TryDecodePayload(message.Payload, out var frame): + case MediaStreamEventType.Media when TryDecodePayload( + message.Payload, + audioStream.Format.BytesPerFrame, + out var frame): // Buffer for paced emission rather than sending straight through. pacer.Enqueue(frame); break; @@ -129,23 +132,45 @@ private async Task PumpConsumerToCallAsync(PacedAudioSender pacer, CancellationT } } - private static bool TryDecodePayload(string? base64, out byte[] frame) + /// + /// Decodes one base64 audio payload, enforcing the negotiated frame size (#108). + /// The encoded length is checked before decoding, so an oversized payload + /// never gets allocated; the decoded frame must then match the format exactly, + /// because a stream that agreed on 20 ms µ-law has no reason to send anything else. + /// A violating frame is dropped, not fatal — a misbehaving consumer must not kill + /// the call. + /// + private static bool TryDecodePayload(string? base64, int expectedFrameBytes, out byte[] frame) { - if (!string.IsNullOrEmpty(base64)) + frame = []; + if (string.IsNullOrEmpty(base64)) { - try - { - frame = Convert.FromBase64String(base64); - return true; - } - catch (FormatException) - { - // A misbehaving consumer must not kill the stream — drop the frame. - } + return false; } - frame = []; - return false; + // 4 base64 chars per 3 bytes; reject before allocating. + if ((long)base64.Length / 4 * 3 > CommunicationStreamLimits.MaxAudioFrameBytes) + { + return false; + } + + byte[] decoded; + try + { + decoded = Convert.FromBase64String(base64); + } + catch (FormatException) + { + return false; + } + + if (decoded.Length != expectedFrameBytes || decoded.Length > CommunicationStreamLimits.MaxAudioFrameBytes) + { + return false; + } + + frame = decoded; + return true; } private static async Task ObserveAsync(Task task) diff --git a/custom/static-plugins/Communication/src/Application/Streaming/MediaStreamSessionPurgeJobHandler.cs b/custom/static-plugins/Communication/src/Application/Streaming/MediaStreamSessionPurgeJobHandler.cs new file mode 100644 index 00000000..10292045 --- /dev/null +++ b/custom/static-plugins/Communication/src/Application/Streaming/MediaStreamSessionPurgeJobHandler.cs @@ -0,0 +1,24 @@ +using Callora.Core.Application.Jobs.Contracts; + +namespace Callora.Plugin.Communication.Application.Streaming; + +/// +/// Removes spent and expired media-stream tickets (#108). A ticket row is a lookup key +/// for a two-minute credential; keeping it after the credential died only grows the +/// table and widens the window in which a leaked row is worth analysing. +/// +public sealed class MediaStreamSessionPurgeJobHandler(IMediaStreamSessionStore sessionStore) : IBackgroundJobHandler +{ + /// Job type key this handler is registered under. + public const string JobTypeName = "communication.media-session-purge"; + + /// + public string JobType => JobTypeName; + + /// + public Task ExecuteAsync(BackgroundJobExecutionContext context, CancellationToken cancellationToken = default) => + sessionStore.PurgeExpiredAsync( + DateTimeOffset.UtcNow, + CommunicationStreamLimits.SessionRetention, + cancellationToken); +} diff --git a/custom/static-plugins/Communication/src/Application/Streaming/MediaStreamSessionPurgeRecurringJobProvider.cs b/custom/static-plugins/Communication/src/Application/Streaming/MediaStreamSessionPurgeRecurringJobProvider.cs new file mode 100644 index 00000000..7c2389b0 --- /dev/null +++ b/custom/static-plugins/Communication/src/Application/Streaming/MediaStreamSessionPurgeRecurringJobProvider.cs @@ -0,0 +1,19 @@ +using Callora.Core.Application.Jobs.Contracts; + +namespace Callora.Plugin.Communication.Application.Streaming; + +/// +/// Schedules the media-session purge. Hourly: rows only become droppable once the +/// retention window has passed, so a tighter cadence would just re-scan. +/// +public sealed class MediaStreamSessionPurgeRecurringJobProvider : IRecurringJobProvider +{ + /// + public IReadOnlyList GetDefinitions() => + [ + new RecurringJobDefinition( + MediaStreamSessionPurgeJobHandler.JobTypeName, + PayloadJson: "{}", + Interval: TimeSpan.FromHours(1)) + ]; +} diff --git a/custom/static-plugins/Communication/src/Application/Streaming/Pacing/PacedAudioSender.cs b/custom/static-plugins/Communication/src/Application/Streaming/Pacing/PacedAudioSender.cs index 3abbb512..3663aa8c 100644 --- a/custom/static-plugins/Communication/src/Application/Streaming/Pacing/PacedAudioSender.cs +++ b/custom/static-plugins/Communication/src/Application/Streaming/Pacing/PacedAudioSender.cs @@ -12,28 +12,40 @@ namespace Callora.Plugin.Communication.Application.Streaming.Pacing; public sealed class PacedAudioSender( Func, CancellationToken, ValueTask> sendFrameAsync, IPacingClock clock, - int maxBufferedFrames = 500) + int maxBufferedFrames = 500, + int maxBufferedBytes = CommunicationStreamLimits.MaxPacedBufferBytes) { private readonly ConcurrentQueue _queue = new(); + private int _bufferedBytes; - /// Queues one outbound frame, dropping the oldest if the safety cap is exceeded. + /// Bytes currently held in the buffer — the quantity the byte cap bounds. + public int BufferedBytes => Volatile.Read(ref _bufferedBytes); + + /// + /// Queues one outbound frame, dropping the oldest until both the frame count and + /// the total byte cap are satisfied. Bounding by count alone was not enough (#108): + /// a producer sending many large frames stays under the count while the buffer grows. + /// public void Enqueue(byte[] frame) { ArgumentNullException.ThrowIfNull(frame); - while (_queue.Count >= maxBufferedFrames && _queue.TryDequeue(out _)) + _queue.Enqueue(frame); + Interlocked.Add(ref _bufferedBytes, frame.Length); + + while ((_queue.Count > maxBufferedFrames || Volatile.Read(ref _bufferedBytes) > maxBufferedBytes) && + _queue.TryDequeue(out var dropped)) { - // Safety cap: bound the buffer so a runaway producer cannot exhaust memory. + Interlocked.Add(ref _bufferedBytes, -dropped.Length); } - - _queue.Enqueue(frame); } /// Drops all queued audio — barge-in: the agent's pending playback stops at once. public void Flush() { - while (_queue.TryDequeue(out _)) + while (_queue.TryDequeue(out var dropped)) { + Interlocked.Add(ref _bufferedBytes, -dropped.Length); } } @@ -44,6 +56,7 @@ public async Task RunAsync(CancellationToken cancellationToken = default) { if (_queue.TryDequeue(out var frame)) { + Interlocked.Add(ref _bufferedBytes, -frame.Length); await sendFrameAsync(frame, cancellationToken).ConfigureAwait(false); } } diff --git a/custom/static-plugins/Communication/src/Application/Voice/ISipAccountRuntimeReconciler.cs b/custom/static-plugins/Communication/src/Application/Voice/ISipAccountRuntimeReconciler.cs new file mode 100644 index 00000000..eb527490 --- /dev/null +++ b/custom/static-plugins/Communication/src/Application/Voice/ISipAccountRuntimeReconciler.cs @@ -0,0 +1,38 @@ +using Callora.Plugin.Communication.Domain.Accounts; + +namespace Callora.Plugin.Communication.Application.Voice; + +/// +/// Brings the live voice runtime in line with a persisted (#110). +/// +/// Persisting an account is not the same as operating it: before this port existed, create, +/// update, enable, disable and delete only wrote rows, so a new account did not register +/// until the next restart and a disabled one kept taking calls. Every successful mutation +/// now runs through here, and so does startup — one reconciler, one code path, so the two +/// cannot drift. +/// +/// +/// Operations are idempotent and state-based: callers declare the account's desired +/// state and the reconciler works out whether that means connect, reconnect or nothing at all. +/// Repeating a request is therefore safe, and concurrent requests for the same account are +/// serialized. +/// +/// +public interface ISipAccountRuntimeReconciler +{ + /// + /// Makes the runtime match : an enabled account ends up connected + /// and registered with its current configuration; a disabled one ends up removed. A + /// configuration change reconnects; an unchanged enabled account is a no-op. + /// + Task ApplyAsync(SipAccount account, CancellationToken cancellationToken = default); + + /// + /// Removes the account from the runtime — deregisters the channel and stops new calls + /// immediately. Idempotent: removing an account that was never provisioned succeeds. + /// + Task RemoveAsync( + string workspaceKey, + string accountId, + CancellationToken cancellationToken = default); +} diff --git a/custom/static-plugins/Communication/src/Application/Voice/SipAuthMethodSupport.cs b/custom/static-plugins/Communication/src/Application/Voice/SipAuthMethodSupport.cs new file mode 100644 index 00000000..d88e57e3 --- /dev/null +++ b/custom/static-plugins/Communication/src/Application/Voice/SipAuthMethodSupport.cs @@ -0,0 +1,50 @@ +using Callora.Plugin.Communication.Domain.Accounts; + +namespace Callora.Plugin.Communication.Application.Voice; + +/// +/// The authentication methods the voice provider can actually connect (#111) — one place, so +/// the admin UI, the API and the provisioner cannot disagree. +/// +/// The domain models three methods because they are real SIP deployments. The provider behind +/// currently connects only digest. Advertising the other +/// two anyway produced accounts that were accepted, then silently skipped at provisioning +/// while the UI sat on Connecting forever. This type turns that mismatch into an +/// explicit, testable boundary: unsupported methods are refused at the edge with an +/// actionable reason instead of failing later and invisibly. +/// +/// +/// Both gaps are tracked upstream and this type is what shrinks as they land: +/// IP-authenticated (registration-less) trunks need +/// SDK #104, +/// mutual TLS needs per-line certificates from +/// SDK #183 +/// (the SDK's TLS configuration is client-wide and file-path based today). +/// +/// +public static class SipAuthMethodSupport +{ + /// Methods the provider can connect today. + public static IReadOnlyList Supported { get; } = [SipAuthMethod.Digest]; + + /// Whether an account using can be connected. + public static bool IsSupported(SipAuthMethod method) => method == SipAuthMethod.Digest; + + /// + /// Operator-facing reason why cannot be used, or null when it can. + /// Names the upstream gap so the message is actionable rather than a bare refusal. + /// + public static string? DescribeUnsupported(SipAuthMethod method) => method switch + { + SipAuthMethod.Digest => null, + SipAuthMethod.IpAuthenticated => + "IP-authenticated trunks are not supported: the voice provider always registers and has no " + + "registration-less mode (callora-voip-sdk#104). Use digest authentication — most trunk " + + "providers offer a registering variant.", + SipAuthMethod.MutualTls => + "Mutual-TLS accounts are not supported: the voice provider's TLS configuration is per client, " + + "not per account, and loads its certificate from a file rather than the secret store " + + "(callora-voip-sdk#183). Use digest authentication over a TLS transport instead.", + _ => $"Authentication method '{method}' is not supported by the voice provider." + }; +} diff --git a/custom/static-plugins/Communication/src/Application/Voice/SipRuntimeReconciliation.cs b/custom/static-plugins/Communication/src/Application/Voice/SipRuntimeReconciliation.cs new file mode 100644 index 00000000..1e342c8f --- /dev/null +++ b/custom/static-plugins/Communication/src/Application/Voice/SipRuntimeReconciliation.cs @@ -0,0 +1,27 @@ +namespace Callora.Plugin.Communication.Application.Voice; + +/// +/// Outcome of one reconciliation. The API result must reflect the runtime, not just the +/// write that preceded it (#110) — a handler reporting success while the account failed to +/// register is exactly the lie this type exists to prevent. +/// +/// What the runtime looks like now. +/// +/// Redacted, operator-facing reason when is +/// ; null otherwise. +/// +public sealed record SipRuntimeReconciliation(SipRuntimeState State, string? Error = null) +{ + /// Whether the runtime now matches the account's desired state. + public bool IsSuccess => State != SipRuntimeState.Failed; + + /// The account is connected and registered with its current configuration. + public static SipRuntimeReconciliation Connected { get; } = new(SipRuntimeState.Connected); + + /// The account is not provisioned — disabled, deleted, or never connected. + public static SipRuntimeReconciliation Removed { get; } = new(SipRuntimeState.Removed); + + /// The runtime could not reach the desired state; the account is not live. + public static SipRuntimeReconciliation Failed(string error) => + new(SipRuntimeState.Failed, string.IsNullOrWhiteSpace(error) ? "The voice runtime rejected the account." : error); +} diff --git a/custom/static-plugins/Communication/src/Application/Voice/SipRuntimeState.cs b/custom/static-plugins/Communication/src/Application/Voice/SipRuntimeState.cs new file mode 100644 index 00000000..465d6a53 --- /dev/null +++ b/custom/static-plugins/Communication/src/Application/Voice/SipRuntimeState.cs @@ -0,0 +1,14 @@ +namespace Callora.Plugin.Communication.Application.Voice; + +/// What the voice runtime holds for an account after a reconciliation. +public enum SipRuntimeState +{ + /// Connected and registered with the account's current configuration. + Connected = 0, + + /// Not provisioned — disabled, deleted, or never connected. + Removed = 1, + + /// The desired state could not be reached; the account is not live. + Failed = 2 +} diff --git a/custom/static-plugins/Communication/src/CommunicationPlugin.cs b/custom/static-plugins/Communication/src/CommunicationPlugin.cs index 8dd98fe8..4bba8fe5 100644 --- a/custom/static-plugins/Communication/src/CommunicationPlugin.cs +++ b/custom/static-plugins/Communication/src/CommunicationPlugin.cs @@ -1,4 +1,5 @@ using Callora.Core.Application.Events.Contracts; +using Callora.Core.Application.Jobs.Contracts; using Callora.Core.Application.Mcp.Contracts; using Callora.Core.Application.Persistence.Contracts; using Callora.Core.Application.Plugins.Contracts; @@ -15,6 +16,7 @@ using Callora.Plugin.Communication.Application.Conference; using Callora.Plugin.Communication.Application.Mcp; using Callora.Plugin.Communication.Application.RealtimeMedia; +using Callora.Plugin.Communication.Application.Streaming; using Callora.Plugin.Communication.Infrastructure.Capabilities; using Callora.Plugin.Communication.Infrastructure.Channels; using Callora.Plugin.Communication.Infrastructure.Persistence; @@ -54,7 +56,10 @@ public sealed class CommunicationPlugin : IHostManagedPlugin // Set during StartAsync when the media/voice surface is wired; torn down on stop. private SdkCallAudioRegistrar? _audioRegistrar; - private VoiceChannelProvisioner? _voiceProvisioner; + + // The single path from a persisted account to a live channel (#110): startup and every + // admin mutation go through it, so the runtime cannot drift from the database. + private SipAccountRuntimeReconciler? _sipRuntimeReconciler; private CommunicationRuntimeCapabilitySource? _capabilitySource; // Call-control primitive, exported for in-process consumers (and the REST adapter); set when the @@ -121,12 +126,22 @@ public async ValueTask StartAsync(IHostPluginContext context, CancellationToken _incomingCallObserver.Start(); } + // Live-call audio surface and the SIP runtime reconciler are built here, before the admin + // routes, because those routes must reconcile the runtime on every mutation (#110). Both + // degrade cleanly: no data protector or no voice runtime means no reconciler, and the + // routes fall back to pure persistence. + var audioStreamProvider = new SdkCallAudioStreamProvider(); + _audioRegistrar = new SdkCallAudioRegistrar( + audioStreamProvider, ResolveLogger(context.Services)); + _sipRuntimeReconciler = TryCreateSipRuntimeReconciler(context, dataProtector); + // Operator Admin-API: the status route always; the SIP-account management routes only when // persistence and a data protector are present (credentials must be protectable) plus the // call-control routes above. These read/write the DB that this same StartAsync migrates below. IReadOnlyList accountRoutes = dbContextFactory is not null && dataProtector is not null - ? SipAccountAdminRoutes.Build(new EfSipAccountStore(dbContextFactory), dataProtector, Id) + ? SipAccountAdminRoutes.Build( + new EfSipAccountStore(dbContextFactory), dataProtector, Id, _sipRuntimeReconciler) : []; context.Export( new CommunicationAdminApiExtensionContributor([.. accountRoutes, .. callRoutes])); @@ -201,14 +216,17 @@ dbContextFactory is not null && dataProtector is not null new CommunicationWorkspaceDataPurger(dbContextFactory))); // Media WebSocket surface (/ws/communication/media/{connectToken}) backed by the live-call - // audio provider; the registrar populates it as tracked calls connect. - var audioStreamProvider = new SdkCallAudioStreamProvider(); - _audioRegistrar = new SdkCallAudioRegistrar( - audioStreamProvider, ResolveLogger(context.Services)); + // audio provider built above; the registrar populates it as tracked calls connect. + var mediaStreamSessionStore = new EfMediaStreamSessionStore(dbContextFactory); context.Export(new CommunicationMediaWebSocketContributor( - new EfMediaStreamSessionStore(dbContextFactory), audioStreamProvider)); + mediaStreamSessionStore, audioStreamProvider)); - await ProvisionVoiceChannelsAsync(context, dbContextFactory, cancellationToken).ConfigureAwait(false); + // Spent and expired media tickets are swept hourly (#108); without this the + // table only ever grows. + context.Export(new MediaStreamSessionPurgeJobHandler(mediaStreamSessionStore)); + context.Export(new MediaStreamSessionPurgeRecurringJobProvider()); + + await ProvisionVoiceChannelsAsync(dbContextFactory, cancellationToken).ConfigureAwait(false); } /// @@ -216,7 +234,7 @@ public async ValueTask StopAsync(CancellationToken cancellationToken = default) { // Deregister and dispose provisioned channels, release live audio streams, then drop all // channel registrations so nothing dangles past unload. - _voiceProvisioner?.Teardown(); + _sipRuntimeReconciler?.Dispose(); if (_audioRegistrar is not null) { await _audioRegistrar.ClearAsync().ConfigureAwait(false); @@ -251,20 +269,19 @@ public async ValueTask StopAsync(CancellationToken cancellationToken = default) // Voice provisioning is opt-in: it needs the plugin data protector (to resolve credentials) and a // voice runtime — either injected by the host or built by the plugin when voice is configured. // Without both the plugin serves the foundation surface only — no voice channels. - private async Task ProvisionVoiceChannelsAsync( + private SipAccountRuntimeReconciler? TryCreateSipRuntimeReconciler( IHostPluginContext context, - IPluginDbContextFactory dbContextFactory, - CancellationToken cancellationToken) + IPluginDataProtector? dataProtector) { - if (context.Services.GetService(typeof(IPluginDataProtector)) is not IPluginDataProtector dataProtector) + if (dataProtector is null) { - return; + return null; } var voiceRuntime = ResolveVoiceRuntime(context.Services, context.PluginConfiguration); if (voiceRuntime is null) { - return; + return null; } var connector = new SdkVoiceChannelConnector( @@ -272,13 +289,42 @@ private async Task ProvisionVoiceChannelsAsync( voiceRuntime, Id, ResolveLogger(context.Services)); - _voiceProvisioner = new VoiceChannelProvisioner( - connector, _channelRegistry, _audioRegistrar!, ResolveLogger(context.Services)); - var enabledAccounts = await new EfSipAccountStore(dbContextFactory) - .ListEnabledAsync(cancellationToken) - .ConfigureAwait(false); - await _voiceProvisioner.ProvisionAsync(enabledAccounts, cancellationToken).ConfigureAwait(false); + return new SipAccountRuntimeReconciler( + connector, + _channelRegistry, + _audioRegistrar!, + ResolveLogger(context.Services)); + } + + // Startup uses the same reconciler as the admin mutations, so there is one provisioning + // path rather than two that can disagree (#110). A failure is written back onto the account, + // so an operator sees why it is dark instead of a permanent "Connecting" (#111/#112). + private async Task ProvisionVoiceChannelsAsync( + IPluginDbContextFactory dbContextFactory, + CancellationToken cancellationToken) + { + if (_sipRuntimeReconciler is null) + { + return; + } + + var store = new EfSipAccountStore(dbContextFactory); + var enabledAccounts = await store.ListEnabledAsync(cancellationToken).ConfigureAwait(false); + + foreach (var account in enabledAccounts) + { + var result = await _sipRuntimeReconciler.ApplyAsync(account, cancellationToken).ConfigureAwait(false); + if (result.IsSuccess) + { + continue; + } + + // Accounts created before the unsupported-method guard existed live on in the + // database; this is where they stop being invisible (#111). + account.ReportStatus(SipAccountStatus.Failed, result.Error, DateTimeOffset.UtcNow); + await store.UpdateAsync(account, cancellationToken).ConfigureAwait(false); + } } // An explicitly injected runtime (tests/custom hosts) always wins. Otherwise, when the deployment diff --git a/custom/static-plugins/Communication/src/Domain/Streaming/MediaStreamSession.cs b/custom/static-plugins/Communication/src/Domain/Streaming/MediaStreamSession.cs index 92809d98..c070ed22 100644 --- a/custom/static-plugins/Communication/src/Domain/Streaming/MediaStreamSession.cs +++ b/custom/static-plugins/Communication/src/Domain/Streaming/MediaStreamSession.cs @@ -1,13 +1,19 @@ +using System.Security.Cryptography; +using System.Text; using Callora.Plugin.Communication.Abstractions; namespace Callora.Plugin.Communication.Domain.Streaming; /// /// Binds a live call to the WebSocket media stream of an external consumer (Twilio-Media- -/// Streams-style). Metadata only — no audio is persisted. The is a -/// short-lived, single-use credential the host validates when the consumer opens the socket; -/// it is consumed by the transition, so one token authorizes exactly -/// one connect. +/// Streams-style). Metadata only — no audio is persisted. +/// +/// The connect token is a short-lived, single-use credential the host validates when the +/// consumer opens the socket; it is consumed by the transition, so one +/// token authorizes exactly one connect. Only its is kept +/// (#108): the row is a lookup key, not a copy of a live credential, so a leaked database +/// row hands out no working ticket. +/// /// public sealed class MediaStreamSession { @@ -33,7 +39,7 @@ public MediaStreamSession( CallId = callId; WorkspaceKey = workspaceKey; ConsumerRef = consumerRef; - ConnectToken = connectToken; + ConnectTokenHash = HashToken(connectToken); Format = format; Direction = direction; CreatedAt = createdAt; @@ -58,8 +64,23 @@ private MediaStreamSession() /// The external consumer this stream serves (for example ai-agent). public string ConsumerRef { get; } - /// Short-lived, single-use credential the host validates on WS connect. - public string ConnectToken { get; } + /// + /// SHA-256 of the connect token, hex-encoded. The plaintext exists only in the + /// response that mints the session; it is never stored. + /// + public string ConnectTokenHash { get; } + + /// + /// One-way, deterministic hash of a connect token — deterministic so the store can + /// look a presented token up, one-way so the stored value is not a credential. No + /// salt: the token is high-entropy already, and a per-row salt would make lookup + /// impossible. + /// + public static string HashToken(string connectToken) + { + ArgumentException.ThrowIfNullOrWhiteSpace(connectToken); + return Convert.ToHexStringLower(SHA256.HashData(Encoding.UTF8.GetBytes(connectToken.Trim()))); + } /// Audio frame format negotiated for this stream. public AudioFormat Format { get; private set; } @@ -81,11 +102,22 @@ private MediaStreamSession() /// /// Whether the connect token may still be redeemed at : the session is - /// still and within - /// of creation. + /// still and its creation lies within + /// — and in the past. A future would + /// otherwise satisfy a bare lower-bound check forever (#108). /// public bool CanActivate(DateTimeOffset now, TimeSpan timeToLive) => - Status == MediaStreamSessionStatus.Pending && now - CreatedAt <= timeToLive; + Status == MediaStreamSessionStatus.Pending && + CreatedAt <= now && + now - CreatedAt <= timeToLive; + + /// + /// Whether the session may be purged at : it is closed, or its + /// ticket has been unusable for longer than . Spent and + /// expired tickets must not accumulate (#108). + /// + public bool CanPurge(DateTimeOffset now, TimeSpan retention) => + (EndedAt ?? CreatedAt) + retention <= now; /// /// Consumes the connect token and marks the stream live. Single-use: only a diff --git a/custom/static-plugins/Communication/src/Infrastructure/Persistence/Configurations/MediaStreamSessionConfiguration.cs b/custom/static-plugins/Communication/src/Infrastructure/Persistence/Configurations/MediaStreamSessionConfiguration.cs index 725ab609..d9ad8af2 100644 --- a/custom/static-plugins/Communication/src/Infrastructure/Persistence/Configurations/MediaStreamSessionConfiguration.cs +++ b/custom/static-plugins/Communication/src/Infrastructure/Persistence/Configurations/MediaStreamSessionConfiguration.cs @@ -17,7 +17,7 @@ public void Configure(EntityTypeBuilder builder) builder.Property(x => x.CallId).HasMaxLength(64).IsRequired(); builder.Property(x => x.WorkspaceKey).HasMaxLength(120).IsRequired(); builder.Property(x => x.ConsumerRef).HasMaxLength(200).IsRequired(); - builder.Property(x => x.ConnectToken).HasMaxLength(200).IsRequired(); + builder.Property(x => x.ConnectTokenHash).HasMaxLength(64).IsRequired(); builder.Property(x => x.Direction).HasConversion().HasMaxLength(15).IsRequired(); builder.Property(x => x.Status).HasConversion().HasMaxLength(15).IsRequired(); builder.Property(x => x.CreatedAt).IsRequired(); @@ -32,7 +32,7 @@ public void Configure(EntityTypeBuilder builder) // Single-use connect token → unique lookup key for WS-connect authorization. Atomic // single-use under a concurrent double-connect is enforced by a conditional UPDATE in the // store (EfMediaStreamSessionStore.TryActivateByConnectTokenAsync), not a mapping concern. - builder.HasIndex(x => x.ConnectToken).IsUnique(); + builder.HasIndex(x => x.ConnectTokenHash).IsUnique(); builder.HasIndex(x => x.WorkspaceKey); } } diff --git a/custom/static-plugins/Communication/src/Infrastructure/Persistence/Migrations/20260805112000_HashMediaStreamConnectTokens.Designer.cs b/custom/static-plugins/Communication/src/Infrastructure/Persistence/Migrations/20260805112000_HashMediaStreamConnectTokens.Designer.cs new file mode 100644 index 00000000..3ea88a7f --- /dev/null +++ b/custom/static-plugins/Communication/src/Infrastructure/Persistence/Migrations/20260805112000_HashMediaStreamConnectTokens.Designer.cs @@ -0,0 +1,344 @@ +// +using System; +using Callora.Plugin.Communication.Infrastructure.Persistence; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Migrations; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; +using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata; + +#nullable disable + +namespace Callora.Plugin.Communication.Infrastructure.Persistence.Migrations +{ + [DbContext(typeof(CommunicationDbContext))] + [Migration("20260805112000_HashMediaStreamConnectTokens")] + partial class HashMediaStreamConnectTokens + { + protected override void BuildTargetModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasDefaultSchema("plugin_communication") + .HasAnnotation("ProductVersion", "10.0.9") + .HasAnnotation("Relational:MaxIdentifierLength", 63); + + NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder); + + modelBuilder.Entity("Callora.Plugin.Communication.Domain.Accounts.SipAccount", b => + { + b.Property("Id") + .HasMaxLength(64) + .HasColumnType("character varying(64)"); + + b.Property("DisplayName") + .IsRequired() + .HasMaxLength(200) + .HasColumnType("character varying(200)"); + + b.Property("Enabled") + .HasColumnType("boolean"); + + b.Property("LastError") + .HasMaxLength(500) + .HasColumnType("character varying(500)"); + + b.Property("LastStatusChangeAt") + .HasColumnType("timestamp with time zone"); + + b.Property("MaxConcurrentCalls") + .HasColumnType("integer"); + + b.Property("Status") + .IsRequired() + .HasMaxLength(20) + .HasColumnType("character varying(20)"); + + b.Property("WorkspaceKey") + .IsRequired() + .HasMaxLength(120) + .HasColumnType("character varying(120)"); + + b.HasKey("Id"); + + b.HasIndex("WorkspaceKey"); + + b.ToTable("sip_accounts", "plugin_communication"); + }); + + modelBuilder.Entity("Callora.Plugin.Communication.Domain.Calls.CallLog", b => + { + b.Property("Id") + .HasMaxLength(64) + .HasColumnType("character varying(64)"); + + b.Property("AccountId") + .HasMaxLength(64) + .HasColumnType("character varying(64)"); + + b.Property("AnsweredAt") + .HasColumnType("timestamp with time zone"); + + b.Property("CorrelationId") + .HasMaxLength(128) + .HasColumnType("character varying(128)"); + + b.Property("Direction") + .IsRequired() + .HasMaxLength(20) + .HasColumnType("character varying(20)"); + + b.Property("DisconnectCause") + .HasMaxLength(200) + .HasColumnType("character varying(200)"); + + b.Property("DurationSeconds") + .HasColumnType("integer"); + + b.Property("EndedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("HandledBy") + .HasMaxLength(100) + .HasColumnType("character varying(100)"); + + b.Property("LineId") + .HasMaxLength(64) + .HasColumnType("character varying(64)"); + + b.Property("LocalIdentity") + .IsRequired() + .HasMaxLength(200) + .HasColumnType("character varying(200)"); + + b.Property("Outcome") + .IsRequired() + .HasMaxLength(20) + .HasColumnType("character varying(20)"); + + b.Property("RemoteParty") + .IsRequired() + .HasMaxLength(200) + .HasColumnType("character varying(200)"); + + b.Property("StartedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("WorkspaceKey") + .IsRequired() + .HasMaxLength(120) + .HasColumnType("character varying(120)"); + + b.HasKey("Id"); + + b.HasIndex("WorkspaceKey", "StartedAt"); + + b.ToTable("call_logs", "plugin_communication"); + }); + + modelBuilder.Entity("Callora.Plugin.Communication.Domain.Lines.SipLine", b => + { + b.Property("Id") + .HasMaxLength(64) + .HasColumnType("character varying(64)"); + + b.Property("AccountId") + .IsRequired() + .HasMaxLength(64) + .HasColumnType("character varying(64)"); + + b.Property("Enabled") + .HasColumnType("boolean"); + + b.Property("InboundRoutingTarget") + .HasMaxLength(200) + .HasColumnType("character varying(200)"); + + b.Property("Label") + .IsRequired() + .HasMaxLength(200) + .HasColumnType("character varying(200)"); + + b.Property("PrimaryNumber") + .HasMaxLength(64) + .HasColumnType("character varying(64)"); + + b.Property("SipUri") + .IsRequired() + .HasMaxLength(255) + .HasColumnType("character varying(255)"); + + b.Property("WorkspaceKey") + .IsRequired() + .HasMaxLength(120) + .HasColumnType("character varying(120)"); + + b.HasKey("Id"); + + b.HasIndex("WorkspaceKey"); + + b.HasIndex("WorkspaceKey", "AccountId"); + + b.ToTable("sip_lines", "plugin_communication"); + }); + + modelBuilder.Entity("Callora.Plugin.Communication.Domain.Streaming.MediaStreamSession", b => + { + b.Property("Id") + .HasMaxLength(64) + .HasColumnType("character varying(64)"); + + b.Property("CallId") + .IsRequired() + .HasMaxLength(64) + .HasColumnType("character varying(64)"); + + b.Property("ConnectTokenHash") + .IsRequired() + .HasMaxLength(64) + .HasColumnType("character varying(64)"); + + b.Property("ConsumerRef") + .IsRequired() + .HasMaxLength(200) + .HasColumnType("character varying(200)"); + + b.Property("CreatedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("Direction") + .IsRequired() + .HasMaxLength(15) + .HasColumnType("character varying(15)"); + + b.Property("EndedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("StartedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("Status") + .IsRequired() + .HasMaxLength(15) + .HasColumnType("character varying(15)"); + + b.Property("WorkspaceKey") + .IsRequired() + .HasMaxLength(120) + .HasColumnType("character varying(120)"); + + b.HasKey("Id"); + + b.HasIndex("ConnectTokenHash") + .IsUnique(); + + b.HasIndex("WorkspaceKey"); + + b.ToTable("media_stream_sessions", "plugin_communication"); + }); + + modelBuilder.Entity("Callora.Plugin.Communication.Domain.Accounts.SipAccount", b => + { + b.OwnsOne("Callora.Plugin.Communication.Domain.Accounts.SipConnection", "Connection", b1 => + { + b1.Property("SipAccountId") + .HasColumnType("character varying(64)"); + + b1.Property("Authentication") + .IsRequired() + .HasColumnType("text") + .HasColumnName("authentication"); + + b1.Property("Host") + .IsRequired() + .HasMaxLength(255) + .HasColumnType("character varying(255)") + .HasColumnName("host"); + + b1.Property("InboundNumbers") + .HasColumnType("text") + .HasColumnName("inbound_numbers"); + + b1.Property("Mode") + .IsRequired() + .HasMaxLength(10) + .HasColumnType("character varying(10)") + .HasColumnName("mode"); + + b1.Property("OutboundProxy") + .HasMaxLength(255) + .HasColumnType("character varying(255)") + .HasColumnName("outbound_proxy"); + + b1.Property("Port") + .HasColumnType("integer") + .HasColumnName("port"); + + b1.Property("RegistrationExpirySeconds") + .HasColumnType("integer") + .HasColumnName("registration_expiry_seconds"); + + b1.Property("Transport") + .IsRequired() + .HasMaxLength(10) + .HasColumnType("character varying(10)") + .HasColumnName("transport"); + + b1.HasKey("SipAccountId"); + + b1.ToTable("sip_accounts", "plugin_communication"); + + b1.WithOwner() + .HasForeignKey("SipAccountId"); + }); + + b.Navigation("Connection") + .IsRequired(); + }); + + modelBuilder.Entity("Callora.Plugin.Communication.Domain.Lines.SipLine", b => + { + b.HasOne("Callora.Plugin.Communication.Domain.Accounts.SipAccount", null) + .WithMany() + .HasForeignKey("WorkspaceKey", "AccountId") + .HasPrincipalKey("WorkspaceKey", "Id") + .OnDelete(DeleteBehavior.Restrict) + .IsRequired(); + }); + + modelBuilder.Entity("Callora.Plugin.Communication.Domain.Streaming.MediaStreamSession", b => + { + b.OwnsOne("Callora.Plugin.Communication.Abstractions.AudioFormat", "Format", b1 => + { + b1.Property("MediaStreamSessionId") + .HasColumnType("character varying(64)"); + + b1.Property("Codec") + .IsRequired() + .HasMaxLength(20) + .HasColumnType("character varying(20)") + .HasColumnName("audio_codec"); + + b1.Property("FrameMilliseconds") + .HasColumnType("integer") + .HasColumnName("audio_frame_ms"); + + b1.Property("SampleRateHz") + .HasColumnType("integer") + .HasColumnName("audio_sample_rate_hz"); + + b1.HasKey("MediaStreamSessionId"); + + b1.ToTable("media_stream_sessions", "plugin_communication"); + + b1.WithOwner() + .HasForeignKey("MediaStreamSessionId"); + }); + + b.Navigation("Format") + .IsRequired(); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/custom/static-plugins/Communication/src/Infrastructure/Persistence/Migrations/20260805112000_HashMediaStreamConnectTokens.cs b/custom/static-plugins/Communication/src/Infrastructure/Persistence/Migrations/20260805112000_HashMediaStreamConnectTokens.cs new file mode 100644 index 00000000..a7d0e5da --- /dev/null +++ b/custom/static-plugins/Communication/src/Infrastructure/Persistence/Migrations/20260805112000_HashMediaStreamConnectTokens.cs @@ -0,0 +1,85 @@ +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace Callora.Plugin.Communication.Infrastructure.Persistence.Migrations; + +/// +public partial class HashMediaStreamConnectTokens : Migration +{ + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + // Connect tokens are no longer stored in the clear (#108): the row keeps only a + // SHA-256 lookup key. Existing rows carry plaintext tokens that cannot be + // converted (hashing them would be correct, but they are two-minute tickets and + // any pending one is already worthless), so the table is emptied rather than + // migrated — that is also the safest outcome for a leaked ticket. + migrationBuilder.Sql( + """ + DELETE FROM plugin_communication.media_stream_sessions; + """); + + migrationBuilder.DropIndex( + name: "IX_media_stream_sessions_ConnectToken", + schema: "plugin_communication", + table: "media_stream_sessions"); + + migrationBuilder.DropColumn( + name: "ConnectToken", + schema: "plugin_communication", + table: "media_stream_sessions"); + + migrationBuilder.AddColumn( + name: "ConnectTokenHash", + schema: "plugin_communication", + table: "media_stream_sessions", + type: "character varying(64)", + maxLength: 64, + nullable: false, + defaultValue: ""); + + migrationBuilder.CreateIndex( + name: "IX_media_stream_sessions_ConnectTokenHash", + schema: "plugin_communication", + table: "media_stream_sessions", + column: "ConnectTokenHash", + unique: true); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + // Downgrading cannot restore plaintext tokens; the table is emptied again. + migrationBuilder.Sql( + """ + DELETE FROM plugin_communication.media_stream_sessions; + """); + + migrationBuilder.DropIndex( + name: "IX_media_stream_sessions_ConnectTokenHash", + schema: "plugin_communication", + table: "media_stream_sessions"); + + migrationBuilder.DropColumn( + name: "ConnectTokenHash", + schema: "plugin_communication", + table: "media_stream_sessions"); + + migrationBuilder.AddColumn( + name: "ConnectToken", + schema: "plugin_communication", + table: "media_stream_sessions", + type: "character varying(200)", + maxLength: 200, + nullable: false, + defaultValue: ""); + + migrationBuilder.CreateIndex( + name: "IX_media_stream_sessions_ConnectToken", + schema: "plugin_communication", + table: "media_stream_sessions", + column: "ConnectToken", + unique: true); + } +} diff --git a/custom/static-plugins/Communication/src/Infrastructure/Persistence/Migrations/CommunicationDbContextModelSnapshot.cs b/custom/static-plugins/Communication/src/Infrastructure/Persistence/Migrations/CommunicationDbContextModelSnapshot.cs index 53d1bcfb..c15c63fb 100644 --- a/custom/static-plugins/Communication/src/Infrastructure/Persistence/Migrations/CommunicationDbContextModelSnapshot.cs +++ b/custom/static-plugins/Communication/src/Infrastructure/Persistence/Migrations/CommunicationDbContextModelSnapshot.cs @@ -191,10 +191,10 @@ protected override void BuildModel(ModelBuilder modelBuilder) .HasMaxLength(64) .HasColumnType("character varying(64)"); - b.Property("ConnectToken") + b.Property("ConnectTokenHash") .IsRequired() - .HasMaxLength(200) - .HasColumnType("character varying(200)"); + .HasMaxLength(64) + .HasColumnType("character varying(64)"); b.Property("ConsumerRef") .IsRequired() @@ -227,7 +227,7 @@ protected override void BuildModel(ModelBuilder modelBuilder) b.HasKey("Id"); - b.HasIndex("ConnectToken") + b.HasIndex("ConnectTokenHash") .IsUnique(); b.HasIndex("WorkspaceKey"); diff --git a/custom/static-plugins/Communication/src/Infrastructure/Persistence/Stores/EfMediaStreamSessionStore.cs b/custom/static-plugins/Communication/src/Infrastructure/Persistence/Stores/EfMediaStreamSessionStore.cs index 8f678c19..27631a14 100644 --- a/custom/static-plugins/Communication/src/Infrastructure/Persistence/Stores/EfMediaStreamSessionStore.cs +++ b/custom/static-plugins/Communication/src/Infrastructure/Persistence/Stores/EfMediaStreamSessionStore.cs @@ -30,9 +30,11 @@ public async Task UpdateAsync(MediaStreamSession session, CancellationToken canc /// public async Task GetByConnectTokenAsync(string connectToken, CancellationToken cancellationToken = default) { + // Only the hash is stored (#108), so the presented token is hashed to look it up. + var tokenHash = MediaStreamSession.HashToken(connectToken); await using var db = dbContextFactory.CreateDbContext(); return await db.MediaStreamSessions.AsNoTracking() - .FirstOrDefaultAsync(x => x.ConnectToken == connectToken, cancellationToken) + .FirstOrDefaultAsync(x => x.ConnectTokenHash == tokenHash, cancellationToken) .ConfigureAwait(false); } @@ -40,17 +42,21 @@ public async Task UpdateAsync(MediaStreamSession session, CancellationToken canc public async Task TryActivateByConnectTokenAsync( string connectToken, DateTimeOffset now, TimeSpan timeToLive, CancellationToken cancellationToken = default) { + var tokenHash = MediaStreamSession.HashToken(connectToken); await using var db = dbContextFactory.CreateDbContext(); // Atomic compare-and-swap: one UPDATE flips Pending → Active only while the token is still - // pending and within its TTL. The predicate mirrors MediaStreamSession.CanActivate; encoding - // it in the WHERE is what makes activation atomic, so a concurrent double-connect cannot both - // win — the loser's UPDATE matches zero rows. + // pending and inside its validity window. The predicate mirrors + // MediaStreamSession.CanActivate — including the upper bound, without which a + // future-dated row would stay redeemable forever (#108). Encoding it in the WHERE is what + // makes activation atomic, so a concurrent double-connect cannot both win — the loser's + // UPDATE matches zero rows. var earliestValidCreation = now - timeToLive; var activated = await db.MediaStreamSessions - .Where(x => x.ConnectToken == connectToken + .Where(x => x.ConnectTokenHash == tokenHash && x.Status == MediaStreamSessionStatus.Pending - && x.CreatedAt >= earliestValidCreation) + && x.CreatedAt >= earliestValidCreation + && x.CreatedAt <= now) .ExecuteUpdateAsync( setters => setters .SetProperty(x => x.Status, MediaStreamSessionStatus.Active) @@ -64,7 +70,22 @@ public async Task UpdateAsync(MediaStreamSession session, CancellationToken canc } return await db.MediaStreamSessions.AsNoTracking() - .FirstOrDefaultAsync(x => x.ConnectToken == connectToken, cancellationToken) + .FirstOrDefaultAsync(x => x.ConnectTokenHash == tokenHash, cancellationToken) + .ConfigureAwait(false); + } + + /// + public async Task PurgeExpiredAsync( + DateTimeOffset now, TimeSpan retention, CancellationToken cancellationToken = default) + { + // Spent and expired tickets must not accumulate (#108): a closed session, or one + // whose token has been unusable for longer than the retention window, is dropped. + var cutoff = now - retention; + await using var db = dbContextFactory.CreateDbContext(); + return await db.MediaStreamSessions + .Where(x => (x.EndedAt != null && x.EndedAt <= cutoff) + || (x.EndedAt == null && x.CreatedAt <= cutoff)) + .ExecuteDeleteAsync(cancellationToken) .ConfigureAwait(false); } diff --git a/custom/static-plugins/Communication/src/Infrastructure/Sdk/ProvisionedVoiceChannel.cs b/custom/static-plugins/Communication/src/Infrastructure/Sdk/ProvisionedVoiceChannel.cs new file mode 100644 index 00000000..dde3b9bb --- /dev/null +++ b/custom/static-plugins/Communication/src/Infrastructure/Sdk/ProvisionedVoiceChannel.cs @@ -0,0 +1,17 @@ +using Callora.Plugin.Communication.Application.Voice; + +namespace Callora.Plugin.Communication.Infrastructure.Sdk; + +/// +/// One account's live registration as the reconciler holds it: the registry handle, the +/// channel it wraps, and the configuration fingerprint it was connected under. The +/// fingerprint is what lets tell "already correct" +/// from "must reconnect" (#110). +/// +/// Registry handle; disposing it deregisters the channel. +/// The audio-registering channel wrapping the provider's voice channel. +/// Runtime-relevant configuration the channel was connected under. +internal sealed record ProvisionedVoiceChannel( + IDisposable Registration, + AudioRegisteringChannel Channel, + string Fingerprint); diff --git a/custom/static-plugins/Communication/src/Infrastructure/Sdk/SipAccountRuntimeReconciler.cs b/custom/static-plugins/Communication/src/Infrastructure/Sdk/SipAccountRuntimeReconciler.cs new file mode 100644 index 00000000..b8e4697e --- /dev/null +++ b/custom/static-plugins/Communication/src/Infrastructure/Sdk/SipAccountRuntimeReconciler.cs @@ -0,0 +1,253 @@ +using System.Collections.Concurrent; +using System.Globalization; +using Callora.Plugin.Communication.Abstractions; +using Callora.Plugin.Communication.Application.Voice; +using Callora.Plugin.Communication.Domain.Accounts; +using Microsoft.Extensions.Logging; + +namespace Callora.Plugin.Communication.Infrastructure.Sdk; + +/// +/// The single path from a persisted to a live, registered voice +/// channel (#110). Startup and every admin mutation call the same +/// , so the runtime cannot drift from the database. +/// +/// Each account is tracked with the fingerprint of the configuration it was connected under. +/// An whose fingerprint is unchanged does nothing — that is what +/// makes repeated calls free. A changed fingerprint tears the old registration down before +/// connecting the new one, because a registrar generally refuses a second registration for +/// the same identity. +/// +/// +/// Per-account locking serializes concurrent mutations, so two operators editing the same +/// account cannot interleave into a half-provisioned state. +/// +/// +public sealed class SipAccountRuntimeReconciler : ISipAccountRuntimeReconciler, IDisposable +{ + private readonly IVoiceChannelConnector _connector; + private readonly ICommunicationChannelRegistry _registry; + private readonly SdkCallAudioRegistrar _registrar; + private readonly ILogger _logger; + + private readonly ConcurrentDictionary _provisioned = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary _locks = new(StringComparer.Ordinal); + + /// Creates a reconciler over the connector seam, channel registry and audio registrar. + public SipAccountRuntimeReconciler( + IVoiceChannelConnector connector, + ICommunicationChannelRegistry registry, + SdkCallAudioRegistrar registrar, + ILogger logger) + { + ArgumentNullException.ThrowIfNull(connector); + ArgumentNullException.ThrowIfNull(registry); + ArgumentNullException.ThrowIfNull(registrar); + ArgumentNullException.ThrowIfNull(logger); + + _connector = connector; + _registry = registry; + _registrar = registrar; + _logger = logger; + } + + /// Accounts currently held as live channels — the runtime's own view. + public int ProvisionedCount => _provisioned.Count; + + /// + public async Task ApplyAsync( + SipAccount account, + CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(account); + + var key = KeyOf(account.WorkspaceKey, account.Id); + var gate = _locks.GetOrAdd(key, _ => new SemaphoreSlim(1, 1)); + await gate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + if (!account.Enabled) + { + TearDown(key); + return SipRuntimeReconciliation.Removed; + } + + // An account the provider cannot connect fails here rather than at the connector, + // with the reason an operator can act on (#111). Accounts predating the edge + // validation reach this path on startup and get that reason persisted. + if (SipAuthMethodSupport.DescribeUnsupported(account.Connection.Authentication.Method) is { } unsupported) + { + TearDown(key); + _logger.LogWarning( + "SIP account {AccountId} uses unsupported authentication {Method}; it stays unprovisioned.", + account.Id, + account.Connection.Authentication.Method); + return SipRuntimeReconciliation.Failed(unsupported); + } + + var fingerprint = Fingerprint(account); + if (_provisioned.TryGetValue(key, out var existing)) + { + if (string.Equals(existing.Fingerprint, fingerprint, StringComparison.Ordinal)) + { + // Nothing changed that the runtime cares about — idempotent no-op. + return SipRuntimeReconciliation.Connected; + } + + // Credentials, endpoint or capacity changed: drop the old registration first so + // the registrar sees one identity, then connect with the new configuration. + TearDown(key); + } + + IVoiceChannel? channel; + try + { + channel = await _connector.ConnectAsync(account, cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) + { + _logger.LogError(ex, "Connecting SIP account {AccountId} failed.", account.Id); + return SipRuntimeReconciliation.Failed("The account could not be connected."); + } + + if (channel is null) + { + _logger.LogWarning("SIP account {AccountId} did not connect.", account.Id); + return SipRuntimeReconciliation.Failed("The account could not be registered."); + } + + var decorated = new AudioRegisteringChannel(channel, _registrar); + var registration = _registry.Register(account.WorkspaceKey, decorated); + _provisioned[key] = new ProvisionedVoiceChannel(registration, decorated, fingerprint); + return SipRuntimeReconciliation.Connected; + } + finally + { + gate.Release(); + } + } + + /// + public async Task RemoveAsync( + string workspaceKey, + string accountId, + CancellationToken cancellationToken = default) + { + ArgumentException.ThrowIfNullOrWhiteSpace(workspaceKey); + ArgumentException.ThrowIfNullOrWhiteSpace(accountId); + + var key = KeyOf(workspaceKey, accountId); + var gate = _locks.GetOrAdd(key, _ => new SemaphoreSlim(1, 1)); + await gate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + TearDown(key); + return SipRuntimeReconciliation.Removed; + } + finally + { + gate.Release(); + } + } + + /// + /// Applies every account in one pass and reports how many ended up live. One account + /// failing never blocks the others — a single unreachable registrar must not take the + /// whole deployment's voice surface down. + /// + public async Task ApplyAllAsync( + IReadOnlyList accounts, + CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(accounts); + + var connected = 0; + foreach (var account in accounts) + { + var result = await ApplyAsync(account, cancellationToken).ConfigureAwait(false); + if (result.State == SipRuntimeState.Connected) + { + connected++; + } + } + + _logger.LogInformation( + "Voice provisioning: {Connected} of {Total} enabled account(s) connected.", + connected, + accounts.Count); + + return new VoiceProvisioningSummary(accounts.Count, connected); + } + + /// Deregisters and disposes every channel this reconciler created (plugin shutdown). + public void Teardown() + { + foreach (var key in _provisioned.Keys.ToArray()) + { + TearDown(key); + } + } + + /// + public void Dispose() + { + Teardown(); + foreach (var gate in _locks.Values) + { + gate.Dispose(); + } + + _locks.Clear(); + } + + private void TearDown(string key) + { + if (!_provisioned.TryRemove(key, out var provisioned)) + { + return; + } + + provisioned.Registration.Dispose(); + provisioned.Channel.Dispose(); + } + + private static string KeyOf(string workspaceKey, string accountId) => + string.Concat(workspaceKey.Trim(), "/", accountId.Trim()); + + /// + /// Everything a live registration depends on. Comparing it is how the reconciler decides + /// between "already correct" and "must reconnect"; the display name is deliberately absent + /// because renaming an account is not a runtime change. + /// + private static string Fingerprint(SipAccount account) + { + var connection = account.Connection; + return string.Join( + '|', + connection.Host, + connection.Port.ToString(CultureInfo.InvariantCulture), + connection.Transport.ToString(), + connection.Mode.ToString(), + connection.RegistrationExpirySeconds?.ToString(CultureInfo.InvariantCulture) ?? "-", + connection.OutboundProxy ?? "-", + string.Join(',', connection.InboundNumbers), + FingerprintAuthentication(connection.Authentication), + account.MaxConcurrentCalls.ToString(CultureInfo.InvariantCulture)); + } + + /// + /// Identity part of the fingerprint. Secret references are compared, never secret + /// values: rotating the stored password behind an unchanged reference is a credential change + /// the reconciler cannot see, so callers rotate the reference alongside it. + /// + private static string FingerprintAuthentication(SipAuthentication authentication) => authentication switch + { + DigestAuthentication digest => + $"digest:{digest.Username}:{digest.AuthId ?? "-"}:{digest.PasswordSecretRef}", + MutualTlsAuthentication mutualTls => + $"mtls:{mutualTls.ClientCertificateSecretRef}", + IpAuthentication => "ip", + _ => authentication.Method.ToString() + }; + +} diff --git a/custom/static-plugins/Communication/src/Infrastructure/Sdk/VoiceChannelProvisioner.cs b/custom/static-plugins/Communication/src/Infrastructure/Sdk/VoiceChannelProvisioner.cs deleted file mode 100644 index a6219073..00000000 --- a/custom/static-plugins/Communication/src/Infrastructure/Sdk/VoiceChannelProvisioner.cs +++ /dev/null @@ -1,111 +0,0 @@ -using Callora.Plugin.Communication.Abstractions; -using Callora.Plugin.Communication.Application.Voice; -using Callora.Plugin.Communication.Domain.Accounts; -using Microsoft.Extensions.Logging; - -namespace Callora.Plugin.Communication.Infrastructure.Sdk; - -/// -/// Turns persisted, enabled s into live, registered voice channels: each -/// account is connected through the seam, wrapped in an -/// so its calls feed the audio surface, and registered in the -/// workspace channel registry. One account failing to connect never blocks the others. -/// reverses it: deregister and dispose every channel it created. -/// -public sealed class VoiceChannelProvisioner -{ - private readonly IVoiceChannelConnector _connector; - private readonly ICommunicationChannelRegistry _registry; - private readonly SdkCallAudioRegistrar _registrar; - private readonly ILogger _logger; - - private readonly List _registrations = []; - private readonly List _channels = []; - - /// Creates a provisioner over the connector seam, channel registry and audio registrar. - public VoiceChannelProvisioner( - IVoiceChannelConnector connector, - ICommunicationChannelRegistry registry, - SdkCallAudioRegistrar registrar, - ILogger logger) - { - ArgumentNullException.ThrowIfNull(connector); - ArgumentNullException.ThrowIfNull(registry); - ArgumentNullException.ThrowIfNull(registrar); - ArgumentNullException.ThrowIfNull(logger); - - _connector = connector; - _registry = registry; - _registrar = registrar; - _logger = logger; - } - - /// - /// Connects and registers a channel for each account. A connect failure (null result or thrown - /// exception) is logged and skipped so the remaining accounts still provision. - /// - public async Task ProvisionAsync( - IReadOnlyList accounts, - CancellationToken cancellationToken = default) - { - ArgumentNullException.ThrowIfNull(accounts); - - var connected = 0; - foreach (var account in accounts) - { - var channel = await ConnectAsync(account, cancellationToken).ConfigureAwait(false); - if (channel is null) - { - continue; - } - - var decorated = new AudioRegisteringChannel(channel, _registrar); - _registrations.Add(_registry.Register(account.WorkspaceKey, decorated)); - _channels.Add(decorated); - connected++; - } - - _logger.LogInformation( - "Voice provisioning: {Connected} of {Total} enabled account(s) connected.", - connected, - accounts.Count); - - return new VoiceProvisioningSummary(accounts.Count, connected); - } - - /// Deregisters and disposes every channel this provisioner created (plugin shutdown). - public void Teardown() - { - foreach (var registration in _registrations) - { - registration.Dispose(); - } - - foreach (var channel in _channels) - { - channel.Dispose(); - } - - _registrations.Clear(); - _channels.Clear(); - } - - private async Task ConnectAsync(SipAccount account, CancellationToken cancellationToken) - { - try - { - var channel = await _connector.ConnectAsync(account, cancellationToken).ConfigureAwait(false); - if (channel is null) - { - _logger.LogWarning("Account {AccountId} did not connect; skipping.", account.Id); - } - - return channel; - } - catch (Exception ex) - { - _logger.LogError(ex, "Connecting account {AccountId} failed; skipping.", account.Id); - return null; - } - } -} diff --git a/custom/static-plugins/Communication/src/Infrastructure/Transport/BoundedWebSocketReader.cs b/custom/static-plugins/Communication/src/Infrastructure/Transport/BoundedWebSocketReader.cs new file mode 100644 index 00000000..ab7fb67b --- /dev/null +++ b/custom/static-plugins/Communication/src/Infrastructure/Transport/BoundedWebSocketReader.cs @@ -0,0 +1,90 @@ +using System.Net.WebSockets; +using System.Text; + +namespace Callora.Plugin.Communication.Infrastructure.Transport; + +/// +/// Reassembles fragmented WebSocket text messages under a hard byte cap and an idle +/// timeout (#108). +/// +/// The naive loop — append every fragment to a growing buffer until +/// EndOfMessage — lets one peer allocate without bound simply by never +/// setting that flag. This reader counts bytes as they arrive and aborts the +/// connection the moment the cap is passed, so the memory a connection can hold is +/// the cap, not the peer's patience. +/// +/// +internal static class BoundedWebSocketReader +{ + /// + /// Reads the next whole text message, or null when the socket closes. + /// + /// + /// The peer exceeded . The socket is closed + /// with first, so the peer learns + /// why rather than seeing an unexplained drop. + /// + public static async ValueTask ReadTextAsync( + WebSocket socket, + byte[] receiveBuffer, + int maxMessageBytes, + TimeSpan idleTimeout, + CancellationToken cancellationToken) + { + using var idle = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + idle.CancelAfter(idleTimeout); + + using var message = new MemoryStream(); + WebSocketReceiveResult result; + do + { + result = await socket.ReceiveAsync(receiveBuffer, idle.Token).ConfigureAwait(false); + if (result.MessageType == WebSocketMessageType.Close) + { + return null; + } + + if (message.Length + result.Count > maxMessageBytes) + { + await CloseTooBigAsync(socket, maxMessageBytes, cancellationToken).ConfigureAwait(false); + throw new WebSocketException( + WebSocketError.HeaderError, + $"The peer exceeded the {maxMessageBytes}-byte message limit."); + } + + message.Write(receiveBuffer, 0, result.Count); + + // Each fragment restarts the idle window: a slow but live peer is fine, + // a silent one is not. + idle.CancelAfter(idleTimeout); + } + while (!result.EndOfMessage); + + return Encoding.UTF8.GetString(message.GetBuffer(), 0, (int)message.Length); + } + + private static async Task CloseTooBigAsync(WebSocket socket, int maxMessageBytes, CancellationToken cancellationToken) + { + if (socket.State is not (WebSocketState.Open or WebSocketState.CloseReceived)) + { + return; + } + + try + { + await socket.CloseOutputAsync( + WebSocketCloseStatus.MessageTooBig, + $"Messages are limited to {maxMessageBytes} bytes.", + cancellationToken) + .ConfigureAwait(false); + } + catch (WebSocketException) + { + // The peer may already be gone; the abort below is what matters. + } + catch (OperationCanceledException) + { + // Shutting down anyway. + } + } +} diff --git a/custom/static-plugins/Communication/src/Infrastructure/Transport/WebSocketMediaFrameChannel.cs b/custom/static-plugins/Communication/src/Infrastructure/Transport/WebSocketMediaFrameChannel.cs index a9aa1f06..62f2ac38 100644 --- a/custom/static-plugins/Communication/src/Infrastructure/Transport/WebSocketMediaFrameChannel.cs +++ b/custom/static-plugins/Communication/src/Infrastructure/Transport/WebSocketMediaFrameChannel.cs @@ -36,25 +36,24 @@ public async ValueTask SendAsync(MediaStreamMessage message, CancellationToken c public async ValueTask ReceiveAsync(CancellationToken cancellationToken = default) { // Loop so a single malformed/unknown frame is skipped rather than ending the stream; - // null is returned only when the socket actually closes. + // null is returned only when the socket actually closes. Reassembly is byte-capped + // and idle-bounded (#108) — an oversized message aborts the connection instead of + // growing the buffer. while (true) { - using var message = new MemoryStream(); - - WebSocketReceiveResult result; - do + var json = await BoundedWebSocketReader + .ReadTextAsync( + socket, + _receiveBuffer, + CommunicationStreamLimits.MaxMediaMessageBytes, + CommunicationStreamLimits.IdleTimeout, + cancellationToken) + .ConfigureAwait(false); + if (json is null) { - result = await socket.ReceiveAsync(_receiveBuffer, cancellationToken).ConfigureAwait(false); - if (result.MessageType == WebSocketMessageType.Close) - { - return null; - } - - message.Write(_receiveBuffer, 0, result.Count); + return null; } - while (!result.EndOfMessage); - var json = Encoding.UTF8.GetString(message.GetBuffer(), 0, (int)message.Length); var decoded = MediaStreamMessageCodec.TryDecode(json); if (decoded is not null) { diff --git a/custom/static-plugins/Communication/src/Resources/app/admin/src/CommunicationAdminPage.vue b/custom/static-plugins/Communication/src/Resources/app/admin/src/CommunicationAdminPage.vue index 0019d190..1916992d 100644 --- a/custom/static-plugins/Communication/src/Resources/app/admin/src/CommunicationAdminPage.vue +++ b/custom/static-plugins/Communication/src/Resources/app/admin/src/CommunicationAdminPage.vue @@ -6,13 +6,17 @@ import { onMounted, reactive, ref } from 'vue' // the target workspace explicitly (?workspaceKey=…), so the page carries that field. const API_BASE = '/api/ext/admin/plugins/communication/' const TRANSPORTS = ['Udp', 'Tcp', 'Tls'] as const -// Matches SipAuthMethod on the backend. The mode (Register/Trunk) and registration -// expiry are derived server-side from the method, so the form only picks the method -// and its credential shape. +// Matches SipAuthMethodSupport.Supported on the backend. The mode (Register/Trunk) and +// registration expiry are derived server-side from the method, so the form only picks the +// method and its credential shape. +// +// Only digest is offered: the voice provider cannot connect IP-authenticated trunks +// (callora-voip-sdk#104, no registration-less mode) or mutual TLS (callora-voip-sdk#183, +// TLS config is client-wide and file-based). The API refuses both with 422, so offering +// them here would only produce accounts that never come up. Re-add an entry when the +// backend adds the method to SipAuthMethodSupport. const AUTH_METHODS = [ { value: 'Digest', label: 'Digest (Registrierung)' }, - { value: 'IpAuthenticated', label: 'IP-Trunk (ohne Zugangsdaten)' }, - { value: 'MutualTls', label: 'Mutual TLS (Client-Zertifikat)' }, ] as const interface SipAccount { @@ -42,13 +46,12 @@ const form = reactive({ username: '', password: '', authId: '', - clientCertificate: '', }) function resetForm(): void { Object.assign(form, { displayName: '', host: '', port: 5060, transport: 'Udp', - authMethod: 'Digest', username: '', password: '', authId: '', clientCertificate: '', + authMethod: 'Digest', username: '', password: '', authId: '', }) } @@ -131,8 +134,8 @@ async function create(): Promise { enabled: true, } - // Credential shape per method (mode/expiry are derived server-side): - // Digest needs username+password, MutualTls a client certificate, IP-trunk nothing. + // Credential shape per method (mode/expiry are derived server-side). Only digest is + // offered; the backend refuses the other methods with 422 until the provider supports them. if (form.authMethod === 'Digest') { if (!form.username.trim() || !form.password) { error.value = 'Für Digest sind Benutzername und Passwort erforderlich.' @@ -143,12 +146,6 @@ async function create(): Promise { if (form.authId.trim()) { body.authId = form.authId.trim() } - } else if (form.authMethod === 'MutualTls') { - if (!form.clientCertificate.trim()) { - error.value = 'Für Mutual TLS ist ein Client-Zertifikat (PEM) erforderlich.' - return - } - body.clientCertificate = form.clientCertificate.trim() } await run(request('POST', 'sip-accounts', body), 'Account angelegt.') @@ -214,18 +211,6 @@ onMounted(() => { -