Add MusicBridge protocol documentation and implement system volume control
- Created PROTOCOL.md to outline the MusicBridge protocol v1, detailing setup, TLS, pairing, state management, and command structure. - Implemented SystemVolume class for managing system audio levels, including methods for setting, reading, changing, and muting volume. - Added unit tests for the MusicBridge agent, covering pairing, identity storage, command validation, and media state management. - Included tests for real-time media reading from Windows and ensured proper handling of artwork caching and retrieval.
This commit is contained in:
@@ -0,0 +1,229 @@
|
||||
using System.Net;
|
||||
using System.Net.WebSockets;
|
||||
using System.Security.Cryptography.X509Certificates;
|
||||
using System.Text.Json;
|
||||
using System.Threading.RateLimiting;
|
||||
using System.Threading.Channels;
|
||||
using Microsoft.AspNetCore.Builder;
|
||||
using Microsoft.AspNetCore.Hosting;
|
||||
using Microsoft.AspNetCore.Http;
|
||||
using Microsoft.AspNetCore.RateLimiting;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using MusicBridge.Agent.Media;
|
||||
|
||||
namespace MusicBridge.Agent.Network;
|
||||
|
||||
internal sealed class RemoteServer : IAsyncDisposable
|
||||
{
|
||||
private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web) { MaxDepth = 8 };
|
||||
private readonly WebApplication app;
|
||||
private readonly IMediaService media;
|
||||
private readonly PairingService pairing;
|
||||
private readonly SemaphoreSlim clients = new(8, 8);
|
||||
private readonly CancellationTokenSource stop = new();
|
||||
private Task? polling;
|
||||
private MediaState latest = new();
|
||||
private readonly Channel<bool> refresh = CreateSignal();
|
||||
private readonly TimeSpan refreshInterval;
|
||||
private event Action? Published;
|
||||
|
||||
public MediaState Latest => Volatile.Read(ref latest);
|
||||
public IEnumerable<string> Addresses => app.Urls;
|
||||
public CancellationToken Stopping => app.Lifetime.ApplicationStopping;
|
||||
|
||||
public RemoteServer(IMediaService media, PairingService pairing, X509Certificate2 certificate, bool lan, int port,
|
||||
TimeSpan? refreshInterval = null)
|
||||
{
|
||||
this.media = media;
|
||||
this.pairing = pairing;
|
||||
this.refreshInterval = refreshInterval ?? TimeSpan.FromMilliseconds(500);
|
||||
var builder = WebApplication.CreateSlimBuilder(new WebApplicationOptions { Args = [] });
|
||||
builder.Logging.ClearProviders(); // Never log pairing bodies or bearer tokens.
|
||||
builder.WebHost.ConfigureKestrel(options =>
|
||||
{
|
||||
options.Limits.MaxRequestBodySize = 4096;
|
||||
options.Limits.MaxConcurrentConnections = 32;
|
||||
options.Limits.MaxConcurrentUpgradedConnections = 8;
|
||||
options.Listen(lan ? IPAddress.Any : IPAddress.Loopback, port, endpoint => endpoint.UseHttps(certificate));
|
||||
});
|
||||
builder.Services.Configure<HostOptions>(options => options.ShutdownTimeout = TimeSpan.FromSeconds(5));
|
||||
builder.Services.AddRateLimiter(options =>
|
||||
{
|
||||
options.RejectionStatusCode = StatusCodes.Status429TooManyRequests;
|
||||
options.GlobalLimiter = PartitionedRateLimiter.Create<HttpContext, string>(_ =>
|
||||
RateLimitPartition.GetFixedWindowLimiter("agent", _ => new FixedWindowRateLimiterOptions
|
||||
{
|
||||
PermitLimit = 60, Window = TimeSpan.FromSeconds(1), QueueLimit = 0
|
||||
}));
|
||||
});
|
||||
app = builder.Build();
|
||||
app.UseRateLimiter();
|
||||
app.UseWebSockets(new WebSocketOptions { KeepAliveInterval = TimeSpan.FromSeconds(20) });
|
||||
app.Use(async (context, next) =>
|
||||
{
|
||||
context.Response.Headers.CacheControl = "no-store";
|
||||
// Native clients have no Origin. Browser pages are deliberately unsupported.
|
||||
if (context.Request.Headers.ContainsKey("Origin"))
|
||||
{
|
||||
context.Response.StatusCode = StatusCodes.Status403Forbidden;
|
||||
return;
|
||||
}
|
||||
if (context.Request.Path != "/v1/pair")
|
||||
{
|
||||
var header = context.Request.Headers.Authorization.ToString();
|
||||
if (!header.StartsWith("Bearer ", StringComparison.Ordinal) || !pairing.Authorize(header[7..]))
|
||||
{
|
||||
context.Response.StatusCode = StatusCodes.Status401Unauthorized;
|
||||
return;
|
||||
}
|
||||
}
|
||||
try { await next(context); }
|
||||
catch (JsonException)
|
||||
{
|
||||
context.Response.StatusCode = StatusCodes.Status400BadRequest;
|
||||
await context.Response.WriteAsJsonAsync(new { code = "invalid_json" }, context.RequestAborted);
|
||||
}
|
||||
});
|
||||
app.MapPost("/v1/pair", async (HttpContext context) =>
|
||||
{
|
||||
var request = await JsonSerializer.DeserializeAsync<PairRequest>(context.Request.Body, JsonOptions, context.RequestAborted);
|
||||
var result = pairing.Pair(request?.Code, request?.DeviceName);
|
||||
return Results.Json(result, statusCode: result.Success ? 200 : result.Code == "storage_failed" ? 503 : 403);
|
||||
});
|
||||
app.MapGet("/v1/state", () => Latest);
|
||||
app.MapGet("/v1/info", () => DiscoveryDescriptor.Create(certificate.GetCertHashString(System.Security.Cryptography.HashAlgorithmName.SHA256)));
|
||||
app.MapGet("/v1/artwork/{id}", (string id, HttpContext context) =>
|
||||
{
|
||||
var image = media.GetArtwork(id);
|
||||
if (image is null) return Results.NotFound();
|
||||
context.Response.Headers.XContentTypeOptions = "nosniff";
|
||||
return Results.Bytes(image.Bytes, image.ContentType);
|
||||
});
|
||||
app.MapPost("/v1/command", async (HttpContext context) =>
|
||||
{
|
||||
var command = await JsonSerializer.DeserializeAsync<MediaCommand>(context.Request.Body, JsonOptions, context.RequestAborted);
|
||||
if (command is null) return Results.BadRequest(new { code = "invalid_command" });
|
||||
var invalid = CommandValidation.Validate(command);
|
||||
if (invalid is not null) return Results.BadRequest(invalid with { Id = command.Id });
|
||||
var result = await media.ExecuteAsync(command, context.RequestAborted);
|
||||
return Results.Json(result);
|
||||
});
|
||||
app.MapGet("/v1/events", StreamAsync);
|
||||
}
|
||||
|
||||
public async Task StartAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
media.Changed += RequestRefresh;
|
||||
Volatile.Write(ref latest, await media.ReadAsync(cancellationToken));
|
||||
await app.StartAsync(cancellationToken);
|
||||
polling = PollAsync(stop.Token);
|
||||
}
|
||||
|
||||
private async Task PollAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
while (!cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
using var deadline = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||||
deadline.CancelAfter(refreshInterval);
|
||||
try { await refresh.Reader.ReadAsync(deadline.Token); }
|
||||
catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) { }
|
||||
// Coalesce bursts of Windows events without accumulating work.
|
||||
await Task.Delay(25, cancellationToken);
|
||||
while (refresh.Reader.TryRead(out _)) { }
|
||||
var state = await media.ReadAsync(cancellationToken);
|
||||
Volatile.Write(ref latest, state);
|
||||
Published?.Invoke();
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { }
|
||||
}
|
||||
|
||||
// WSS is a state stream; commands use HTTPS so each has its own response.
|
||||
private async Task StreamAsync(HttpContext context)
|
||||
{
|
||||
if (!context.WebSockets.IsWebSocketRequest)
|
||||
{
|
||||
context.Response.StatusCode = StatusCodes.Status400BadRequest;
|
||||
return;
|
||||
}
|
||||
if (!await clients.WaitAsync(0, context.RequestAborted))
|
||||
{
|
||||
context.Response.StatusCode = StatusCodes.Status503ServiceUnavailable;
|
||||
return;
|
||||
}
|
||||
try
|
||||
{
|
||||
using var socket = await context.WebSockets.AcceptWebSocketAsync();
|
||||
using var lifetime = CancellationTokenSource.CreateLinkedTokenSource(context.RequestAborted, stop.Token, Stopping);
|
||||
var token = context.Request.Headers.Authorization.ToString()[7..];
|
||||
var sender = SendStatesAsync(socket, token, lifetime.Token);
|
||||
var receiver = ReceiveCloseAsync(socket, lifetime.Token);
|
||||
await Task.WhenAny(sender, receiver);
|
||||
await lifetime.CancelAsync();
|
||||
try { await Task.WhenAll(sender, receiver); }
|
||||
catch (OperationCanceledException) { }
|
||||
catch (WebSocketException) { }
|
||||
if (socket.State is WebSocketState.Open or WebSocketState.CloseReceived)
|
||||
{
|
||||
using var closeTimeout = new CancellationTokenSource(TimeSpan.FromSeconds(2));
|
||||
try
|
||||
{
|
||||
await socket.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, "State stream closed", closeTimeout.Token);
|
||||
}
|
||||
catch (OperationCanceledException) { }
|
||||
catch (WebSocketException) { }
|
||||
}
|
||||
}
|
||||
finally { clients.Release(); }
|
||||
}
|
||||
|
||||
private async Task SendStatesAsync(WebSocket socket, string token, CancellationToken cancellationToken)
|
||||
{
|
||||
var pending = CreateSignal();
|
||||
void Signal() => pending.Writer.TryWrite(true);
|
||||
Published += Signal;
|
||||
pairing.TrustChanged += Signal;
|
||||
Signal();
|
||||
try
|
||||
{
|
||||
while (await pending.Reader.WaitToReadAsync(cancellationToken))
|
||||
{
|
||||
while (pending.Reader.TryRead(out _)) { }
|
||||
if (!pairing.Authorize(token)) return;
|
||||
var bytes = JsonSerializer.SerializeToUtf8Bytes(new { type = "state", state = Latest }, JsonOptions);
|
||||
await socket.SendAsync(bytes.AsMemory(), WebSocketMessageType.Text, true, cancellationToken);
|
||||
}
|
||||
}
|
||||
finally { Published -= Signal; pairing.TrustChanged -= Signal; }
|
||||
}
|
||||
|
||||
private static Channel<bool> CreateSignal() => Channel.CreateBounded<bool>(new BoundedChannelOptions(1)
|
||||
{
|
||||
FullMode = BoundedChannelFullMode.DropOldest, SingleReader = true, SingleWriter = false
|
||||
});
|
||||
private void RequestRefresh() => refresh.Writer.TryWrite(true);
|
||||
|
||||
private static async Task ReceiveCloseAsync(WebSocket socket, CancellationToken cancellationToken)
|
||||
{
|
||||
var buffer = new byte[1];
|
||||
// A receive observes disconnects. No commands or client payloads are accepted here.
|
||||
await socket.ReceiveAsync(buffer.AsMemory(), cancellationToken);
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
media.Changed -= RequestRefresh;
|
||||
await stop.CancelAsync();
|
||||
if (polling is not null) await polling;
|
||||
await app.StopAsync();
|
||||
await app.DisposeAsync();
|
||||
stop.Dispose();
|
||||
clients.Dispose();
|
||||
}
|
||||
|
||||
private sealed record PairRequest(string? Code, string? DeviceName);
|
||||
}
|
||||
Reference in New Issue
Block a user