From ba4d487c6512894f4605588972cf8e19cc1aceca Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Thu, 24 Sep 2026 23:02:56 +1000 Subject: [PATCH] share quick connect state between instances Move the pending requests and authorized secrets out of process so the initiate, authorize and exchange legs can land on different replicas. --- .woodpecker/ci.yaml | 6 +- .../QuickConnect/QuickConnectManager.cs | 80 ++--- .../QuickConnect/RedisQuickConnectStore.cs | 151 +++++++++ .../Controllers/QuickConnectController.cs | 6 +- Jellyfin.Api/Controllers/UserController.cs | 4 +- Jellyfin.Server/CoreAppHost.cs | 4 + ...ConnectStoreServiceCollectionExtensions.cs | 72 ++++ .../QuickConnect/IQuickConnect.cs | 6 +- .../QuickConnect/IQuickConnectStore.cs | 59 ++++ .../QuickConnect/InMemoryQuickConnectStore.cs | 89 +++++ .../QuickConnect/QuickConnectManagerTests.cs | 49 ++- .../HighAvailability/RedisTestServer.cs | 91 ++++++ .../Jellyfin.Server.Tests.csproj | 1 + .../QuickConnect/QuickConnectReplicaTests.cs | 308 ++++++++++++++++++ 14 files changed, 841 insertions(+), 85 deletions(-) create mode 100644 Emby.Server.Implementations/QuickConnect/RedisQuickConnectStore.cs create mode 100644 Jellyfin.Server/Extensions/QuickConnectStoreServiceCollectionExtensions.cs create mode 100644 MediaBrowser.Controller/QuickConnect/IQuickConnectStore.cs create mode 100644 MediaBrowser.Controller/QuickConnect/InMemoryQuickConnectStore.cs create mode 100644 tests/Jellyfin.Server.Tests/HighAvailability/RedisTestServer.cs create mode 100644 tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectReplicaTests.cs diff --git a/.woodpecker/ci.yaml b/.woodpecker/ci.yaml index c1c5e741f0..668be16db5 100644 --- a/.woodpecker/ci.yaml +++ b/.woodpecker/ci.yaml @@ -47,6 +47,7 @@ steps: # Its data directory lives on the step's ephemeral storage, not on the workspace volume. # Both projects attach to it through JELLYFIN_TEST_POSTGRES and give every test a database of # its own, so nothing here depends on a docker daemon. + # Valkey runs in the step for the same reason, attached through JELLYFIN_TEST_REDIS. - name: postgres-migration-chain image: mcr.microsoft.com/dotnet/sdk:10.0 depends_on: @@ -55,13 +56,16 @@ steps: DOTNET_CLI_TELEMETRY_OPTOUT: "1" DOTNET_NOLOGO: "1" JELLYFIN_TEST_POSTGRES: "Host=127.0.0.1;Port=5432;Database=postgres;Username=postgres" + JELLYFIN_TEST_REDIS: "127.0.0.1:6379" commands: - apt-get -o Acquire::Retries=3 update || apt-get -o Acquire::Retries=3 update - - DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends postgresql + - DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends postgresql valkey-server - install -d -o postgres -g postgres /tmp/pgdata /tmp/pgrun - PGBIN=$(ls -d /usr/lib/postgresql/*/bin | tail -1) - su postgres -c "$PGBIN/initdb -D /tmp/pgdata -A trust -U postgres" - su postgres -c "$PGBIN/pg_ctl -D /tmp/pgdata -o \"-c listen_addresses=127.0.0.1 -k /tmp/pgrun\" -l /tmp/pg.log -w start" + - valkey-server --daemonize yes --bind 127.0.0.1 --port 6379 --save '' + - valkey-cli ping - dotnet build tests/Jellyfin.Server.Tests/Jellyfin.Server.Tests.csproj -c Release - dotnet build tests/Jellyfin.Database.Tests.PostgreSQL/Jellyfin.Database.Tests.PostgreSQL.csproj -c Release - dotnet test tests/Jellyfin.Server.Tests/Jellyfin.Server.Tests.csproj -c Release --no-build --verbosity minimal --filter "Category=RequiresDocker" diff --git a/Emby.Server.Implementations/QuickConnect/QuickConnectManager.cs b/Emby.Server.Implementations/QuickConnect/QuickConnectManager.cs index c4bda96371..855205e4aa 100644 --- a/Emby.Server.Implementations/QuickConnect/QuickConnectManager.cs +++ b/Emby.Server.Implementations/QuickConnect/QuickConnectManager.cs @@ -1,7 +1,5 @@ using System; -using System.Collections.Concurrent; using System.Globalization; -using System.Linq; using System.Security.Cryptography; using System.Threading.Tasks; using MediaBrowser.Common.Extensions; @@ -30,12 +28,10 @@ namespace Emby.Server.Implementations.QuickConnect /// private const int Timeout = 10; - private readonly ConcurrentDictionary _currentRequests = new(); - private readonly ConcurrentDictionary _authorizedSecrets = new(); - private readonly IServerConfigurationManager _config; private readonly ILogger _logger; private readonly ISessionManager _sessionManager; + private readonly IQuickConnectStore _store; /// /// Initializes a new instance of the class. @@ -44,14 +40,17 @@ namespace Emby.Server.Implementations.QuickConnect /// Configuration. /// Logger. /// Session Manager. + /// Quick connect store. public QuickConnectManager( IServerConfigurationManager config, ILogger logger, - ISessionManager sessionManager) + ISessionManager sessionManager, + IQuickConnectStore store) { _config = config; _logger = logger; _sessionManager = sessionManager; + _store = store; } /// @@ -69,7 +68,7 @@ namespace Emby.Server.Implementations.QuickConnect } /// - public QuickConnectResult TryConnect(AuthorizationInfo authorizationInfo) + public async Task TryConnect(AuthorizationInfo authorizationInfo) { ArgumentException.ThrowIfNullOrEmpty(authorizationInfo.DeviceId); ArgumentException.ThrowIfNullOrEmpty(authorizationInfo.Device); @@ -77,7 +76,6 @@ namespace Emby.Server.Implementations.QuickConnect ArgumentException.ThrowIfNullOrEmpty(authorizationInfo.Version); AssertActive(); - ExpireRequests(); var secret = GenerateSecureRandom(); var code = GenerateCode(); @@ -90,19 +88,17 @@ namespace Emby.Server.Implementations.QuickConnect authorizationInfo.Client, authorizationInfo.Version); - _currentRequests[code] = result; + await _store.SetRequestAsync(result, ExpiryOf(result)).ConfigureAwait(false); return result; } /// - public QuickConnectResult CheckRequestStatus(string secret) + public async Task CheckRequestStatus(string secret) { AssertActive(); - ExpireRequests(); - string code = _currentRequests.Where(x => x.Value.Secret == secret).Select(x => x.Value.Code).DefaultIfEmpty(string.Empty).First(); - - if (!_currentRequests.TryGetValue(code, out QuickConnectResult? result)) + var result = await _store.GetRequestBySecretAsync(secret).ConfigureAwait(false); + if (result is null) { throw new ResourceNotFoundException("Unable to find request with provided secret"); } @@ -136,9 +132,9 @@ namespace Emby.Server.Implementations.QuickConnect public async Task AuthorizeRequest(Guid userId, string code) { AssertActive(); - ExpireRequests(); - if (!_currentRequests.TryGetValue(code, out QuickConnectResult? result)) + var result = await _store.GetRequestByCodeAsync(code).ConfigureAwait(false); + if (result is null) { throw new ResourceNotFoundException("Unable to find request"); } @@ -160,9 +156,10 @@ namespace Emby.Server.Implementations.QuickConnect AppVersion = result.AppVersion }).ConfigureAwait(false); - _authorizedSecrets[result.Secret] = (DateTime.UtcNow, authenticationResult); result.Authenticated = true; - _currentRequests[code] = result; + + await _store.SetAuthorizationAsync(result.Secret, authenticationResult, DateTime.UtcNow.AddMinutes(Timeout)).ConfigureAwait(false); + await _store.SetRequestAsync(result, ExpiryOf(result)).ConfigureAwait(false); _logger.LogDebug("Authorizing device with code {Code} to login as user {UserId}", code, userId); @@ -170,19 +167,21 @@ namespace Emby.Server.Implementations.QuickConnect } /// - public AuthenticationResult GetAuthorizedRequest(string secret) + public async Task GetAuthorizedRequest(string secret) { AssertActive(); - ExpireRequests(); - if (!_authorizedSecrets.TryGetValue(secret, out var result)) + var result = await _store.TryConsumeAuthorizationAsync(secret).ConfigureAwait(false); + if (result is null) { throw new ResourceNotFoundException("Unable to find request"); } - return result.AuthenticationResult; + return result; } + private static DateTime ExpiryOf(QuickConnectResult request) => request.DateAdded.AddMinutes(Timeout); + private string GenerateSecureRandom(int length = 32) { Span bytes = stackalloc byte[length]; @@ -190,42 +189,5 @@ namespace Emby.Server.Implementations.QuickConnect return Convert.ToHexString(bytes); } - - /// - /// Expire quick connect requests that are over the time limit. If is true, all requests are unconditionally expired. - /// - /// If true, all requests will be expired. - private void ExpireRequests(bool expireAll = false) - { - // All requests before this timestamp have expired - var minTime = DateTime.UtcNow.AddMinutes(-Timeout); - - // Expire stale connection requests - foreach (var (_, currentRequest) in _currentRequests) - { - if (expireAll || currentRequest.DateAdded < minTime) - { - var code = currentRequest.Code; - _logger.LogDebug("Removing expired request {Code}", code); - - if (!_currentRequests.TryRemove(code, out _)) - { - _logger.LogWarning("Request {Code} already expired", code); - } - } - } - - foreach (var (secret, (timestamp, _)) in _authorizedSecrets) - { - if (expireAll || timestamp < minTime) - { - _logger.LogDebug("Removing expired secret {Secret}", secret); - if (!_authorizedSecrets.TryRemove(secret, out _)) - { - _logger.LogWarning("Secret {Secret} already expired", secret); - } - } - } - } } } diff --git a/Emby.Server.Implementations/QuickConnect/RedisQuickConnectStore.cs b/Emby.Server.Implementations/QuickConnect/RedisQuickConnectStore.cs new file mode 100644 index 0000000000..715e0a6c6e --- /dev/null +++ b/Emby.Server.Implementations/QuickConnect/RedisQuickConnectStore.cs @@ -0,0 +1,151 @@ +using System; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using Jellyfin.Extensions.Json; +using MediaBrowser.Controller.Authentication; +using MediaBrowser.Controller.QuickConnect; +using MediaBrowser.Model.QuickConnect; +using Microsoft.Extensions.Logging; +using StackExchange.Redis; + +namespace Emby.Server.Implementations.QuickConnect; + +/// +/// A Redis-backed that lets the initiate, authorize and exchange legs +/// of a quick connect flow land on different instances. Expiry is the key TTL, and an authorization is +/// consumed with GETDEL so only one instance can ever hand out a given secret's access token. +/// +public sealed class RedisQuickConnectStore : IQuickConnectStore +{ + private const string KeyPrefix = "jellyfin:quickconnect:"; + + private readonly IDatabase _db; + private readonly InMemoryQuickConnectStore _fallback; + private readonly ILogger _logger; + + /// + /// Initializes a new instance of the class. + /// + /// The Redis connection multiplexer. + /// The logger. + public RedisQuickConnectStore(IConnectionMultiplexer redis, ILogger logger) + { + ArgumentNullException.ThrowIfNull(redis); + + _db = redis.GetDatabase(); + _fallback = new InMemoryQuickConnectStore(); + _logger = logger; + } + + /// + public async Task GetRequestBySecretAsync(string secret, CancellationToken cancellationToken = default) + { + try + { + var raw = await _db.StringGetAsync(RequestKey(secret)).ConfigureAwait(false); + if (raw.HasValue) + { + return JsonSerializer.Deserialize(raw.ToString(), JsonDefaults.Options); + } + } + catch (Exception ex) + { + LogDegraded(ex); + } + + return await _fallback.GetRequestBySecretAsync(secret, cancellationToken).ConfigureAwait(false); + } + + /// + public async Task GetRequestByCodeAsync(string code, CancellationToken cancellationToken = default) + { + try + { + var secret = await _db.StringGetAsync(CodeKey(code)).ConfigureAwait(false); + if (secret.HasValue) + { + return await GetRequestBySecretAsync(secret.ToString(), cancellationToken).ConfigureAwait(false); + } + } + catch (Exception ex) + { + LogDegraded(ex); + } + + return await _fallback.GetRequestByCodeAsync(code, cancellationToken).ConfigureAwait(false); + } + + /// + public async Task SetRequestAsync(QuickConnectResult request, DateTime expiresUtc, CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(request); + + var ttl = expiresUtc - DateTime.UtcNow; + if (ttl <= TimeSpan.Zero) + { + return; + } + + try + { + var json = JsonSerializer.Serialize(request, JsonDefaults.Options); + await _db.StringSetAsync(RequestKey(request.Secret), json, ttl).ConfigureAwait(false); + await _db.StringSetAsync(CodeKey(request.Code), request.Secret, ttl).ConfigureAwait(false); + } + catch (Exception ex) + { + LogDegraded(ex); + await _fallback.SetRequestAsync(request, expiresUtc, cancellationToken).ConfigureAwait(false); + } + } + + /// + public async Task SetAuthorizationAsync(string secret, AuthenticationResult authenticationResult, DateTime expiresUtc, CancellationToken cancellationToken = default) + { + var ttl = expiresUtc - DateTime.UtcNow; + if (ttl <= TimeSpan.Zero) + { + return; + } + + try + { + var json = JsonSerializer.Serialize(authenticationResult, JsonDefaults.Options); + await _db.StringSetAsync(AuthorizationKey(secret), json, ttl).ConfigureAwait(false); + } + catch (Exception ex) + { + LogDegraded(ex); + await _fallback.SetAuthorizationAsync(secret, authenticationResult, expiresUtc, cancellationToken).ConfigureAwait(false); + } + } + + /// + public async Task TryConsumeAuthorizationAsync(string secret, CancellationToken cancellationToken = default) + { + try + { + var raw = await _db.StringGetDeleteAsync(AuthorizationKey(secret)).ConfigureAwait(false); + if (raw.HasValue) + { + return JsonSerializer.Deserialize(raw.ToString(), JsonDefaults.Options); + } + } + catch (Exception ex) + { + LogDegraded(ex); + } + + return await _fallback.TryConsumeAuthorizationAsync(secret, cancellationToken).ConfigureAwait(false); + } + + private static string RequestKey(string secret) => KeyPrefix + "request:" + secret; + + private static string CodeKey(string code) => KeyPrefix + "code:" + code; + + private static string AuthorizationKey(string secret) => KeyPrefix + "auth:" + secret; + + private void LogDegraded(Exception exception) + => _logger.LogWarning(exception, "Quick connect state could not be shared through Redis; falling back to this instance only."); +} diff --git a/Jellyfin.Api/Controllers/QuickConnectController.cs b/Jellyfin.Api/Controllers/QuickConnectController.cs index 5c7b38e137..eebda12ff3 100644 --- a/Jellyfin.Api/Controllers/QuickConnectController.cs +++ b/Jellyfin.Api/Controllers/QuickConnectController.cs @@ -59,7 +59,7 @@ public class QuickConnectController : BaseJellyfinApiController try { var auth = await _authContext.GetAuthorizationInfo(Request).ConfigureAwait(false); - return _quickConnect.TryConnect(auth); + return await _quickConnect.TryConnect(auth).ConfigureAwait(false); } catch (AuthenticationException) { @@ -77,11 +77,11 @@ public class QuickConnectController : BaseJellyfinApiController [HttpGet("Connect")] [ProducesResponseType(StatusCodes.Status200OK)] [ProducesResponseType(StatusCodes.Status404NotFound)] - public ActionResult GetQuickConnectState([FromQuery, Required] string secret) + public async Task> GetQuickConnectState([FromQuery, Required] string secret) { try { - return _quickConnect.CheckRequestStatus(secret); + return await _quickConnect.CheckRequestStatus(secret).ConfigureAwait(false); } catch (ResourceNotFoundException) { diff --git a/Jellyfin.Api/Controllers/UserController.cs b/Jellyfin.Api/Controllers/UserController.cs index 657bda4d15..c8aeb616a4 100644 --- a/Jellyfin.Api/Controllers/UserController.cs +++ b/Jellyfin.Api/Controllers/UserController.cs @@ -245,11 +245,11 @@ public class UserController : BaseJellyfinApiController [HttpPost("AuthenticateWithQuickConnect")] [ProducesResponseType(StatusCodes.Status200OK)] [Tags("Authentication")] - public ActionResult AuthenticateWithQuickConnect([FromBody, Required] QuickConnectDto request) + public async Task> AuthenticateWithQuickConnect([FromBody, Required] QuickConnectDto request) { try { - return _quickConnectManager.GetAuthorizedRequest(request.Secret); + return await _quickConnectManager.GetAuthorizedRequest(request.Secret).ConfigureAwait(false); } catch (SecurityException e) { diff --git a/Jellyfin.Server/CoreAppHost.cs b/Jellyfin.Server/CoreAppHost.cs index 9ac86bb8ed..4e30328870 100644 --- a/Jellyfin.Server/CoreAppHost.cs +++ b/Jellyfin.Server/CoreAppHost.cs @@ -116,6 +116,10 @@ namespace Jellyfin.Server // to the other instances. Redis-backed when configured, no-op otherwise. serviceCollection.AddConfigurationInvalidationBus(_startupConfig, Logger); + // Quick connect store: shares in-flight quick connect requests so the initiate, authorize and + // exchange legs can land on different instances. Redis-backed when configured, local otherwise. + serviceCollection.AddQuickConnectStore(_startupConfig, Logger); + foreach (var type in GetExportTypes()) { serviceCollection.AddSingleton(typeof(ILyricProvider), type); diff --git a/Jellyfin.Server/Extensions/QuickConnectStoreServiceCollectionExtensions.cs b/Jellyfin.Server/Extensions/QuickConnectStoreServiceCollectionExtensions.cs new file mode 100644 index 0000000000..36f270e66e --- /dev/null +++ b/Jellyfin.Server/Extensions/QuickConnectStoreServiceCollectionExtensions.cs @@ -0,0 +1,72 @@ +using System; +using Emby.Server.Implementations.QuickConnect; +using MediaBrowser.Controller.MediaEncoding; +using MediaBrowser.Controller.QuickConnect; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using StackExchange.Redis; + +namespace Jellyfin.Server.Extensions; + +/// +/// Extensions for registering the quick connect store. +/// +public static class QuickConnectStoreServiceCollectionExtensions +{ + /// + /// Registers the quick connect store, Redis-backed when a connection string is configured and + /// process-local otherwise, and reports the selected store at . + /// + /// + /// The connection string is only set for a multi-instance deployment, which is the only shape where + /// the initiate, authorize and exchange legs of one flow can land on different instances. + /// + /// The service collection. + /// The configuration to read the Redis connection string from. + /// The logger to report the selected store on. + /// The updated service collection. + public static IServiceCollection AddQuickConnectStore( + this IServiceCollection serviceCollection, + IConfiguration configuration, + ILogger logger) + { + ArgumentNullException.ThrowIfNull(configuration); + ArgumentNullException.ThrowIfNull(logger); + + if (string.IsNullOrEmpty(configuration[TranscodeStoreOptions.RedisConnectionStringKey])) + { + logger.LogInformation( + "Quick connect store: {Store}. A quick connect flow has to complete against one instance; set {Key} to share it.", + nameof(InMemoryQuickConnectStore), + TranscodeStoreOptions.RedisConnectionStringKey); + + return serviceCollection.AddSingleton(); + } + + logger.LogInformation( + "Quick connect store: {Store}. Quick connect flows complete across any instance.", + nameof(RedisQuickConnectStore)); + + return serviceCollection.AddSingleton(sp => + { + try + { + return new RedisQuickConnectStore( + sp.GetRequiredService(), + sp.GetRequiredService>()); + } + catch (Exception ex) + { + // Fail open: an unreachable Redis degrades to the single-instance behaviour of a flow + // having to complete against one instance, rather than taking quick connect down. + sp.GetRequiredService>().LogError( + ex, + "Redis is configured but unavailable, so quick connect flows will not complete across instances. Check {Key}.", + TranscodeStoreOptions.RedisConnectionStringKey); + + return new InMemoryQuickConnectStore(); + } + }); + } +} diff --git a/MediaBrowser.Controller/QuickConnect/IQuickConnect.cs b/MediaBrowser.Controller/QuickConnect/IQuickConnect.cs index ec3706773a..b585c30dbd 100644 --- a/MediaBrowser.Controller/QuickConnect/IQuickConnect.cs +++ b/MediaBrowser.Controller/QuickConnect/IQuickConnect.cs @@ -21,14 +21,14 @@ namespace MediaBrowser.Controller.QuickConnect /// /// The initiator authorization info. /// A quick connect result with tokens to proceed or throws an exception if not active. - QuickConnectResult TryConnect(AuthorizationInfo authorizationInfo); + Task TryConnect(AuthorizationInfo authorizationInfo); /// /// Checks the status of an individual request. /// /// Unique secret identifier of the request. /// Quick connect result. - QuickConnectResult CheckRequestStatus(string secret); + Task CheckRequestStatus(string secret); /// /// Authorizes a quick connect request to connect as the calling user. @@ -43,6 +43,6 @@ namespace MediaBrowser.Controller.QuickConnect /// /// The secret. /// The authentication result. - AuthenticationResult GetAuthorizedRequest(string secret); + Task GetAuthorizedRequest(string secret); } } diff --git a/MediaBrowser.Controller/QuickConnect/IQuickConnectStore.cs b/MediaBrowser.Controller/QuickConnect/IQuickConnectStore.cs new file mode 100644 index 0000000000..2d47c90507 --- /dev/null +++ b/MediaBrowser.Controller/QuickConnect/IQuickConnectStore.cs @@ -0,0 +1,59 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using MediaBrowser.Controller.Authentication; +using MediaBrowser.Model.QuickConnect; + +namespace MediaBrowser.Controller.QuickConnect; + +/// +/// Holds the state of in-flight quick connect requests. The three legs of a quick connect flow - +/// initiate, authorize and exchange - can each land on a different instance, so the state has to be +/// reachable from all of them. +/// +public interface IQuickConnectStore +{ + /// + /// Looks up a pending request by the secret handed to the initiating client. + /// + /// The request secret. + /// A cancellation token. + /// The request, or null when it is unknown or has expired. + Task GetRequestBySecretAsync(string secret, CancellationToken cancellationToken = default); + + /// + /// Looks up a pending request by the code shown to the user. + /// + /// The user facing code. + /// A cancellation token. + /// The request, or null when it is unknown or has expired. + Task GetRequestByCodeAsync(string code, CancellationToken cancellationToken = default); + + /// + /// Stores a new or updated request until . + /// + /// The request to store. + /// The instant the request stops being resolvable. + /// A cancellation token. + /// A representing the asynchronous operation. + Task SetRequestAsync(QuickConnectResult request, DateTime expiresUtc, CancellationToken cancellationToken = default); + + /// + /// Stores the authentication minted for an authorized request until . + /// + /// The request secret the client exchanges. + /// The authentication to hand out. + /// The instant the authentication stops being exchangeable. + /// A cancellation token. + /// A representing the asynchronous operation. + Task SetAuthorizationAsync(string secret, AuthenticationResult authenticationResult, DateTime expiresUtc, CancellationToken cancellationToken = default); + + /// + /// Atomically takes the authentication for and removes it, so that two + /// instances racing on the same secret cannot both hand out an access token. + /// + /// The request secret. + /// A cancellation token. + /// The authentication, or null when the secret is unknown, expired or already exchanged. + Task TryConsumeAuthorizationAsync(string secret, CancellationToken cancellationToken = default); +} diff --git a/MediaBrowser.Controller/QuickConnect/InMemoryQuickConnectStore.cs b/MediaBrowser.Controller/QuickConnect/InMemoryQuickConnectStore.cs new file mode 100644 index 0000000000..0b42bd08d3 --- /dev/null +++ b/MediaBrowser.Controller/QuickConnect/InMemoryQuickConnectStore.cs @@ -0,0 +1,89 @@ +using System; +using System.Collections.Concurrent; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using MediaBrowser.Controller.Authentication; +using MediaBrowser.Model.QuickConnect; + +namespace MediaBrowser.Controller.QuickConnect; + +/// +/// A process-local . It is the single-instance default, and the +/// fallback a shared store degrades to while its backend is unreachable, so quick connect keeps +/// working for clients whose three legs happen to land on one instance. +/// +public sealed class InMemoryQuickConnectStore : IQuickConnectStore +{ + private readonly ConcurrentDictionary> _requests = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary> _authorizations = new(StringComparer.Ordinal); + + /// + public Task GetRequestBySecretAsync(string secret, CancellationToken cancellationToken = default) + { + Expire(); + return Task.FromResult(_requests.TryGetValue(secret, out var entry) ? entry.Value : null); + } + + /// + public Task GetRequestByCodeAsync(string code, CancellationToken cancellationToken = default) + { + Expire(); + return Task.FromResult(_requests.Values + .Select(entry => entry.Value) + .FirstOrDefault(request => string.Equals(request.Code, code, StringComparison.Ordinal))); + } + + /// + public Task SetRequestAsync(QuickConnectResult request, DateTime expiresUtc, CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(request); + + Expire(); + _requests[request.Secret] = new Entry(expiresUtc, request); + return Task.CompletedTask; + } + + /// + public Task SetAuthorizationAsync(string secret, AuthenticationResult authenticationResult, DateTime expiresUtc, CancellationToken cancellationToken = default) + { + Expire(); + _authorizations[secret] = new Entry(expiresUtc, authenticationResult); + return Task.CompletedTask; + } + + /// + public Task TryConsumeAuthorizationAsync(string secret, CancellationToken cancellationToken = default) + { + Expire(); + if (!_authorizations.TryRemove(secret, out var entry) || entry.ExpiresUtc <= DateTime.UtcNow) + { + return Task.FromResult(null); + } + + return Task.FromResult(entry.Value); + } + + private void Expire() + { + var now = DateTime.UtcNow; + + foreach (var (secret, entry) in _requests) + { + if (entry.ExpiresUtc <= now) + { + _requests.TryRemove(secret, out _); + } + } + + foreach (var (secret, entry) in _authorizations) + { + if (entry.ExpiresUtc <= now) + { + _authorizations.TryRemove(secret, out _); + } + } + } + + private sealed record Entry(DateTime ExpiresUtc, T Value); +} diff --git a/tests/Jellyfin.Server.Implementations.Tests/QuickConnect/QuickConnectManagerTests.cs b/tests/Jellyfin.Server.Implementations.Tests/QuickConnect/QuickConnectManagerTests.cs index 30f72f5957..dffede2851 100644 --- a/tests/Jellyfin.Server.Implementations.Tests/QuickConnect/QuickConnectManagerTests.cs +++ b/tests/Jellyfin.Server.Implementations.Tests/QuickConnect/QuickConnectManagerTests.cs @@ -8,6 +8,7 @@ using MediaBrowser.Common.Extensions; using MediaBrowser.Controller.Authentication; using MediaBrowser.Controller.Configuration; using MediaBrowser.Controller.Net; +using MediaBrowser.Controller.QuickConnect; using MediaBrowser.Model.Configuration; using Moq; using Xunit; @@ -40,6 +41,8 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect ConfigureMembers = true }).Inject(configManager.Object); + _fixture.Inject(new InMemoryQuickConnectStore()); + // User object contains circular references. _fixture.Behaviors.OfType().ToList() .ForEach(b => _fixture.Behaviors.Remove(b)); @@ -60,8 +63,8 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect [InlineData("Device", "", "Client", "1.0.0")] [InlineData("Device", "DeviceId", "", "1.0.0")] [InlineData("Device", "DeviceId", "Client", "")] - public void TryConnect_InvalidAuthorizationInfo_ThrowsArgumentException(string device, string deviceId, string client, string version) - => Assert.Throws(() => _quickConnectManager.TryConnect( + public async Task TryConnect_InvalidAuthorizationInfo_ThrowsArgumentException(string device, string deviceId, string client, string version) + => await Assert.ThrowsAsync(() => _quickConnectManager.TryConnect( new AuthorizationInfo { Device = device, @@ -71,17 +74,17 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect })); [Fact] - public void TryConnect_QuickConnectUnavailable_ThrowsAuthenticationException() + public async Task TryConnect_QuickConnectUnavailable_ThrowsAuthenticationException() { _config.QuickConnectAvailable = false; - Assert.Throws(() => _quickConnectManager.TryConnect(_quickConnectAuthInfo)); + await Assert.ThrowsAsync(() => _quickConnectManager.TryConnect(_quickConnectAuthInfo)); } [Fact] - public void CheckRequestStatus_QuickConnectUnavailable_ThrowsAuthenticationException() + public async Task CheckRequestStatus_QuickConnectUnavailable_ThrowsAuthenticationException() { _config.QuickConnectAvailable = false; - Assert.Throws(() => _quickConnectManager.CheckRequestStatus(string.Empty)); + await Assert.ThrowsAsync(() => _quickConnectManager.CheckRequestStatus(string.Empty)); } [Fact] @@ -92,10 +95,10 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect } [Fact] - public void GetAuthorizedRequest_QuickConnectUnavailable_ThrowsAuthenticationException() + public async Task GetAuthorizedRequest_QuickConnectUnavailable_ThrowsAuthenticationException() { _config.QuickConnectAvailable = false; - Assert.Throws(() => _quickConnectManager.GetAuthorizedRequest(string.Empty)); + await Assert.ThrowsAsync(() => _quickConnectManager.GetAuthorizedRequest(string.Empty)); } [Fact] @@ -106,34 +109,46 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect } [Fact] - public void CheckRequestStatus_QuickConnectAvailable_Success() + public async Task CheckRequestStatus_QuickConnectAvailable_Success() { _config.QuickConnectAvailable = true; - var res1 = _quickConnectManager.TryConnect(_quickConnectAuthInfo); - var res2 = _quickConnectManager.CheckRequestStatus(res1.Secret); - Assert.Equal(res1, res2); + var res1 = await _quickConnectManager.TryConnect(_quickConnectAuthInfo); + var res2 = await _quickConnectManager.CheckRequestStatus(res1.Secret); + Assert.Equal(res1.Secret, res2.Secret); + Assert.Equal(res1.Code, res2.Code); } [Fact] - public void CheckRequestStatus_UnknownSecret_ThrowsResourceNotFoundException() + public async Task CheckRequestStatus_UnknownSecret_ThrowsResourceNotFoundException() { _config.QuickConnectAvailable = true; - Assert.Throws(() => _quickConnectManager.CheckRequestStatus("Unknown secret")); + await Assert.ThrowsAsync(() => _quickConnectManager.CheckRequestStatus("Unknown secret")); } [Fact] - public void GetAuthorizedRequest_UnknownSecret_ThrowsResourceNotFoundException() + public async Task GetAuthorizedRequest_UnknownSecret_ThrowsResourceNotFoundException() { _config.QuickConnectAvailable = true; - Assert.Throws(() => _quickConnectManager.GetAuthorizedRequest("Unknown secret")); + await Assert.ThrowsAsync(() => _quickConnectManager.GetAuthorizedRequest("Unknown secret")); } [Fact] public async Task AuthorizeRequest_QuickConnectAvailable_Success() { _config.QuickConnectAvailable = true; - var res = _quickConnectManager.TryConnect(_quickConnectAuthInfo); + var res = await _quickConnectManager.TryConnect(_quickConnectAuthInfo); Assert.True(await _quickConnectManager.AuthorizeRequest(Guid.Empty, res.Code)); } + + [Fact] + public async Task GetAuthorizedRequest_SecondExchange_ThrowsResourceNotFoundException() + { + _config.QuickConnectAvailable = true; + var res = await _quickConnectManager.TryConnect(_quickConnectAuthInfo); + await _quickConnectManager.AuthorizeRequest(Guid.Empty, res.Code); + + Assert.NotNull(await _quickConnectManager.GetAuthorizedRequest(res.Secret)); + await Assert.ThrowsAsync(() => _quickConnectManager.GetAuthorizedRequest(res.Secret)); + } } } diff --git a/tests/Jellyfin.Server.Tests/HighAvailability/RedisTestServer.cs b/tests/Jellyfin.Server.Tests/HighAvailability/RedisTestServer.cs new file mode 100644 index 0000000000..a17850522c --- /dev/null +++ b/tests/Jellyfin.Server.Tests/HighAvailability/RedisTestServer.cs @@ -0,0 +1,91 @@ +using System; +using System.Threading.Tasks; +using StackExchange.Redis; +using Testcontainers.Redis; + +namespace Jellyfin.Server.Tests.HighAvailability; + +/// +/// Hands out a Redis server for the tests that need one. A server named by JELLYFIN_TEST_REDIS is +/// used as is, so CI can run one beside the step instead of a docker daemon of its own; without it a +/// container is started through testcontainers. +/// +public sealed class RedisTestServer : IAsyncDisposable +{ + /// + /// The connection string of an already running server. + /// + public const string ConnectionStringVariable = "JELLYFIN_TEST_REDIS"; + + private readonly RedisContainer? _container; + + private RedisTestServer(RedisContainer? container, string connectionString) + { + _container = container; + ConnectionString = connectionString; + } + + /// + /// Gets the connection string of the running server. + /// + public string ConnectionString { get; } + + /// + /// Starts or attaches to a Redis server and waits until it accepts connections. + /// + /// The running server. + public static async Task StartAsync() + { + var provided = Environment.GetEnvironmentVariable(ConnectionStringVariable); + if (!string.IsNullOrWhiteSpace(provided)) + { + var attached = new RedisTestServer(null, provided); + await attached.WaitUntilReadyAsync().ConfigureAwait(false); + return attached; + } + + var container = new RedisBuilder("redis:7-alpine").Build(); + await container.StartAsync().ConfigureAwait(false); + + var started = new RedisTestServer(container, container.GetConnectionString()); + await started.WaitUntilReadyAsync().ConfigureAwait(false); + return started; + } + + /// + /// Opens a connection of its own, so each in-process stand-in for a replica talks to the server the + /// way a separate pod would. + /// + /// A new multiplexer. + public async Task ConnectAsync() + => await ConnectionMultiplexer.ConnectAsync(ConnectionString).ConfigureAwait(false); + + /// + public async ValueTask DisposeAsync() + { + if (_container is not null) + { + await _container.DisposeAsync().ConfigureAwait(false); + } + } + + private async Task WaitUntilReadyAsync() + { + for (var attempt = 1; ; attempt++) + { + try + { + var connection = await ConnectionMultiplexer.ConnectAsync(ConnectionString).ConfigureAwait(false); + await using (connection.ConfigureAwait(false)) + { + await connection.GetDatabase().PingAsync().ConfigureAwait(false); + return; + } + } + catch (RedisException) when (attempt < 60) + { + await Task.Delay(TimeSpan.FromSeconds(1)).ConfigureAwait(false); + } + } + } +} diff --git a/tests/Jellyfin.Server.Tests/Jellyfin.Server.Tests.csproj b/tests/Jellyfin.Server.Tests/Jellyfin.Server.Tests.csproj index a96840baa1..b5d8f219d6 100644 --- a/tests/Jellyfin.Server.Tests/Jellyfin.Server.Tests.csproj +++ b/tests/Jellyfin.Server.Tests/Jellyfin.Server.Tests.csproj @@ -12,6 +12,7 @@ + all diff --git a/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectReplicaTests.cs b/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectReplicaTests.cs new file mode 100644 index 0000000000..26fff9a1f1 --- /dev/null +++ b/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectReplicaTests.cs @@ -0,0 +1,308 @@ +using System; +using System.Collections.Generic; +using System.Globalization; +using System.Threading; +using System.Threading.Tasks; +using Emby.Server.Implementations.QuickConnect; +using Jellyfin.Data.Queries; +using Jellyfin.Database.Implementations; +using Jellyfin.Database.Implementations.DbConfiguration; +using Jellyfin.Database.Implementations.Entities; +using Jellyfin.Database.Implementations.Entities.Security; +using Jellyfin.Database.Implementations.Locking; +using Jellyfin.Database.Providers.PostgreSQL; +using Jellyfin.Server.Implementations.Devices; +using Jellyfin.Server.Tests.HighAvailability; +using Jellyfin.Server.Tests.Migrations; +using MediaBrowser.Common.Extensions; +using MediaBrowser.Controller.Authentication; +using MediaBrowser.Controller.Configuration; +using MediaBrowser.Controller.Devices; +using MediaBrowser.Controller.Library; +using MediaBrowser.Controller.Net; +using MediaBrowser.Controller.QuickConnect; +using MediaBrowser.Controller.Session; +using MediaBrowser.Model.Configuration; +using MediaBrowser.Model.Dto; +using MediaBrowser.Model.QuickConnect; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using Npgsql; +using StackExchange.Redis; +using Xunit; + +namespace Jellyfin.Server.Tests.QuickConnect; + +/// +/// Three independently constructed instances over one PostgreSQL +/// database and one Redis are the in-process stand-in for three replicas without sticky sessions: the +/// initiate, authorize and exchange legs of one flow each land on a different one. +/// +[Trait("Category", "RequiresDocker")] +public sealed class QuickConnectReplicaTests : IAsyncLifetime +{ + private static readonly AuthorizationInfo _authorizationInfo = new AuthorizationInfo + { + Device = "Living Room TV", + DeviceId = "device-1", + Client = "Jellyfin Web", + Version = "1.0.0" + }; + + private readonly List _connections = new(); + + private PostgreSqlTestServer _postgres = null!; + private RedisTestServer _redis = null!; + + /// + public async ValueTask InitializeAsync() + { + _postgres = await PostgreSqlTestServer.StartAsync().ConfigureAwait(false); + _redis = await RedisTestServer.StartAsync().ConfigureAwait(false); + } + + /// + public async ValueTask DisposeAsync() + { + foreach (var connection in _connections) + { + await connection.DisposeAsync().ConfigureAwait(false); + } + + await _redis.DisposeAsync().ConfigureAwait(false); + await _postgres.DisposeAsync().ConfigureAwait(false); + } + + /// + /// The three legs of a quick connect flow land on three different replicas, and the token the third + /// one hands out is the one the second one minted into the shared database. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task InitiateAuthorizeExchange_AcrossThreeReplicas_Succeeds() + { + var cancellationToken = TestContext.Current.CancellationToken; + var connectionString = await _postgres.CreateDatabaseAsync("quickconnect_replica_flow", cancellationToken); + + await using var dataSource = new NpgsqlDataSourceBuilder(connectionString).Build(); + var user = await CreateSchemaWithUserAsync(dataSource, cancellationToken); + + var replicaA = await CreateReplicaAsync(dataSource, user); + var replicaB = await CreateReplicaAsync(dataSource, user); + var replicaC = await CreateReplicaAsync(dataSource, user); + + var initiated = await replicaA.Manager.TryConnect(_authorizationInfo); + + // The code is shown to the user on whichever replica serves the dashboard. + Assert.True(await replicaB.Manager.AuthorizeRequest(user.Id, initiated.Code)); + + var polled = await replicaC.Manager.CheckRequestStatus(initiated.Secret); + Assert.True(polled.Authenticated); + Assert.Equal(initiated.Code, polled.Code); + Assert.Equal(_authorizationInfo.DeviceId, polled.DeviceId); + + var exchanged = await replicaC.Manager.GetAuthorizedRequest(initiated.Secret); + + Assert.False(string.IsNullOrEmpty(exchanged.AccessToken)); + Assert.Equal(user.Id, exchanged.User.Id); + + var devices = await replicaA.Devices.GetDevices(new DeviceQuery { AccessToken = exchanged.AccessToken }); + Assert.Equal(user.Id, Assert.Single(devices.Items).UserId); + } + + /// + /// A secret is single use across the whole deployment: two replicas racing to exchange it must not + /// both hand out an access token. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Exchange_RacedOnTwoReplicas_SucceedsOnce() + { + var cancellationToken = TestContext.Current.CancellationToken; + var connectionString = await _postgres.CreateDatabaseAsync("quickconnect_replica_race", cancellationToken); + + await using var dataSource = new NpgsqlDataSourceBuilder(connectionString).Build(); + var user = await CreateSchemaWithUserAsync(dataSource, cancellationToken); + + var replicaA = await CreateReplicaAsync(dataSource, user); + var replicaB = await CreateReplicaAsync(dataSource, user); + var replicaC = await CreateReplicaAsync(dataSource, user); + + var initiated = await replicaA.Manager.TryConnect(_authorizationInfo); + await replicaB.Manager.AuthorizeRequest(user.Id, initiated.Code); + + var onA = ExchangeAsync(replicaA.Manager, initiated.Secret); + var onC = ExchangeAsync(replicaC.Manager, initiated.Secret); + var outcomes = await Task.WhenAll(onA, onC); + + Assert.Single(outcomes, outcome => outcome is not null); + + // And it stays consumed for every later attempt, on any replica. + await Assert.ThrowsAsync(() => replicaB.Manager.GetAuthorizedRequest(initiated.Secret)); + } + + /// + /// An expired request is rejected on a replica that never saw it created, rather than resolving to a + /// stale authorization. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task ExpiredRequest_IsRejectedOnEveryReplica() + { + var cancellationToken = TestContext.Current.CancellationToken; + var connectionString = await _postgres.CreateDatabaseAsync("quickconnect_replica_expiry", cancellationToken); + + await using var dataSource = new NpgsqlDataSourceBuilder(connectionString).Build(); + var user = await CreateSchemaWithUserAsync(dataSource, cancellationToken); + + var replicaA = await CreateReplicaAsync(dataSource, user); + var replicaB = await CreateReplicaAsync(dataSource, user); + + var initiated = await replicaA.Manager.TryConnect(_authorizationInfo); + Assert.NotNull(await replicaB.Manager.CheckRequestStatus(initiated.Secret)); + + // Shorten the stored expiry instead of waiting out the ten minute timeout. + await replicaA.Store.SetRequestAsync(initiated, DateTime.UtcNow.AddSeconds(1), cancellationToken); + await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken); + + await Assert.ThrowsAsync(() => replicaB.Manager.CheckRequestStatus(initiated.Secret)); + await Assert.ThrowsAsync(() => replicaB.Manager.AuthorizeRequest(user.Id, initiated.Code)); + } + + /// + /// An authorization that was never exchanged expires too, so a code authorized and then abandoned + /// cannot be redeemed later from another replica. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task ExpiredAuthorization_IsRejectedOnEveryReplica() + { + var cancellationToken = TestContext.Current.CancellationToken; + var connectionString = await _postgres.CreateDatabaseAsync("quickconnect_replica_auth_expiry", cancellationToken); + + await using var dataSource = new NpgsqlDataSourceBuilder(connectionString).Build(); + var user = await CreateSchemaWithUserAsync(dataSource, cancellationToken); + + var replicaA = await CreateReplicaAsync(dataSource, user); + var replicaB = await CreateReplicaAsync(dataSource, user); + + var initiated = await replicaA.Manager.TryConnect(_authorizationInfo); + await replicaA.Manager.AuthorizeRequest(user.Id, initiated.Code); + + var stored = await replicaA.Store.GetRequestBySecretAsync(initiated.Secret, cancellationToken); + Assert.True(stored?.Authenticated); + + await replicaA.Store.SetAuthorizationAsync( + initiated.Secret, + new AuthenticationResult { AccessToken = "stale" }, + DateTime.UtcNow.AddSeconds(1), + cancellationToken); + await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken); + + await Assert.ThrowsAsync(() => replicaB.Manager.GetAuthorizedRequest(initiated.Secret)); + } + + private static async Task ExchangeAsync(IQuickConnect manager, string secret) + { + try + { + return await manager.GetAuthorizedRequest(secret).ConfigureAwait(false); + } + catch (ResourceNotFoundException) + { + return null; + } + } + + private static async Task CreateSchemaWithUserAsync(NpgsqlDataSource dataSource, CancellationToken cancellationToken) + { + var context = CreateContext(dataSource); + await using (context.ConfigureAwait(false)) + { + await context.Database.EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); + + var user = new User("quickconnect-user", "provider", "provider"); + context.Users.Add(user); + await context.SaveChangesAsync(cancellationToken).ConfigureAwait(false); + + return user; + } + } + + private static JellyfinDbContext CreateContext(NpgsqlDataSource dataSource) + { + var optionsBuilder = new DbContextOptionsBuilder(); + var provider = new PostgreSqlDatabaseProvider(dataSource); + provider.Initialise(optionsBuilder, new DatabaseConfigurationOptions { DatabaseType = "PostgreSQL" }); + return new JellyfinDbContext( + optionsBuilder.Options, + NullLogger.Instance, + provider, + new NoLockBehavior(NullLogger.Instance)); + } + + private async Task CreateReplicaAsync(NpgsqlDataSource dataSource, User user) + { + var connection = await _redis.ConnectAsync().ConfigureAwait(false); + _connections.Add(connection); + + var userManager = new Mock(); + userManager.Setup(manager => manager.GetUserById(user.Id)).Returns(user); + var deviceManager = new DeviceManager(new DataSourceContextFactory(dataSource), userManager.Object); + + var configManager = new Mock(); + configManager.Setup(manager => manager.Configuration).Returns(new ServerConfiguration { QuickConnectAvailable = true }); + + // Stands in for SessionManager.AuthenticateDirect: the token has to be minted into the shared + // database, because the replica that exchanges the secret is not the one that authorized it. + var sessionManager = new Mock(); + sessionManager + .Setup(manager => manager.AuthenticateDirect(It.IsAny())) + .Returns(async request => + { + var device = await deviceManager.CreateDevice( + new Device(request.UserId, request.App, request.AppVersion, request.DeviceName, request.DeviceId)).ConfigureAwait(false); + + return new AuthenticationResult + { + AccessToken = device.AccessToken, + ServerId = "server-1", + User = new UserDto { Id = user.Id, Name = user.Username, ServerId = "server-1" }, + SessionInfo = new SessionInfoDto + { + Id = device.Id.ToString(CultureInfo.InvariantCulture), + UserId = user.Id, + UserName = user.Username, + Client = request.App, + DeviceId = request.DeviceId, + DeviceName = request.DeviceName, + ApplicationVersion = request.AppVersion + } + }; + }); + + var store = new RedisQuickConnectStore(connection, NullLogger.Instance); + var manager = new QuickConnectManager( + configManager.Object, + NullLogger.Instance, + sessionManager.Object, + store); + + return new Replica(manager, store, deviceManager); + } + + private sealed record Replica(IQuickConnect Manager, IQuickConnectStore Store, IDeviceManager Devices); + + private sealed class DataSourceContextFactory : IDbContextFactory + { + private readonly NpgsqlDataSource _dataSource; + + public DataSourceContextFactory(NpgsqlDataSource dataSource) + { + _dataSource = dataSource; + } + + public JellyfinDbContext CreateDbContext() => CreateContext(_dataSource); + } +}