share quick connect state between instances
ci/woodpecker/push/ci Pipeline was successful
ci/woodpecker/pr/ci Pipeline was successful

Move the pending requests and authorized secrets out of process so the
initiate, authorize and exchange legs can land on different replicas.
This commit is contained in:
2026-09-24 23:02:56 +10:00
parent 6691b785c3
commit ba4d487c65
14 changed files with 841 additions and 85 deletions
+5 -1
View File
@@ -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"
@@ -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
/// </summary>
private const int Timeout = 10;
private readonly ConcurrentDictionary<string, QuickConnectResult> _currentRequests = new();
private readonly ConcurrentDictionary<string, (DateTime Timestamp, AuthenticationResult AuthenticationResult)> _authorizedSecrets = new();
private readonly IServerConfigurationManager _config;
private readonly ILogger<QuickConnectManager> _logger;
private readonly ISessionManager _sessionManager;
private readonly IQuickConnectStore _store;
/// <summary>
/// Initializes a new instance of the <see cref="QuickConnectManager"/> class.
@@ -44,14 +40,17 @@ namespace Emby.Server.Implementations.QuickConnect
/// <param name="config">Configuration.</param>
/// <param name="logger">Logger.</param>
/// <param name="sessionManager">Session Manager.</param>
/// <param name="store">Quick connect store.</param>
public QuickConnectManager(
IServerConfigurationManager config,
ILogger<QuickConnectManager> logger,
ISessionManager sessionManager)
ISessionManager sessionManager,
IQuickConnectStore store)
{
_config = config;
_logger = logger;
_sessionManager = sessionManager;
_store = store;
}
/// <inheritdoc />
@@ -69,7 +68,7 @@ namespace Emby.Server.Implementations.QuickConnect
}
/// <inheritdoc/>
public QuickConnectResult TryConnect(AuthorizationInfo authorizationInfo)
public async Task<QuickConnectResult> 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;
}
/// <inheritdoc/>
public QuickConnectResult CheckRequestStatus(string secret)
public async Task<QuickConnectResult> 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<bool> 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
}
/// <inheritdoc/>
public AuthenticationResult GetAuthorizedRequest(string secret)
public async Task<AuthenticationResult> 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<byte> bytes = stackalloc byte[length];
@@ -190,42 +189,5 @@ namespace Emby.Server.Implementations.QuickConnect
return Convert.ToHexString(bytes);
}
/// <summary>
/// Expire quick connect requests that are over the time limit. If <paramref name="expireAll"/> is true, all requests are unconditionally expired.
/// </summary>
/// <param name="expireAll">If true, all requests will be expired.</param>
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);
}
}
}
}
}
}
@@ -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;
/// <summary>
/// A Redis-backed <see cref="IQuickConnectStore"/> 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 <c>GETDEL</c> so only one instance can ever hand out a given secret's access token.
/// </summary>
public sealed class RedisQuickConnectStore : IQuickConnectStore
{
private const string KeyPrefix = "jellyfin:quickconnect:";
private readonly IDatabase _db;
private readonly InMemoryQuickConnectStore _fallback;
private readonly ILogger<RedisQuickConnectStore> _logger;
/// <summary>
/// Initializes a new instance of the <see cref="RedisQuickConnectStore"/> class.
/// </summary>
/// <param name="redis">The Redis connection multiplexer.</param>
/// <param name="logger">The logger.</param>
public RedisQuickConnectStore(IConnectionMultiplexer redis, ILogger<RedisQuickConnectStore> logger)
{
ArgumentNullException.ThrowIfNull(redis);
_db = redis.GetDatabase();
_fallback = new InMemoryQuickConnectStore();
_logger = logger;
}
/// <inheritdoc />
public async Task<QuickConnectResult?> GetRequestBySecretAsync(string secret, CancellationToken cancellationToken = default)
{
try
{
var raw = await _db.StringGetAsync(RequestKey(secret)).ConfigureAwait(false);
if (raw.HasValue)
{
return JsonSerializer.Deserialize<QuickConnectResult>(raw.ToString(), JsonDefaults.Options);
}
}
catch (Exception ex)
{
LogDegraded(ex);
}
return await _fallback.GetRequestBySecretAsync(secret, cancellationToken).ConfigureAwait(false);
}
/// <inheritdoc />
public async Task<QuickConnectResult?> 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);
}
/// <inheritdoc />
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);
}
}
/// <inheritdoc />
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);
}
}
/// <inheritdoc />
public async Task<AuthenticationResult?> TryConsumeAuthorizationAsync(string secret, CancellationToken cancellationToken = default)
{
try
{
var raw = await _db.StringGetDeleteAsync(AuthorizationKey(secret)).ConfigureAwait(false);
if (raw.HasValue)
{
return JsonSerializer.Deserialize<AuthenticationResult>(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.");
}
@@ -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<QuickConnectResult> GetQuickConnectState([FromQuery, Required] string secret)
public async Task<ActionResult<QuickConnectResult>> GetQuickConnectState([FromQuery, Required] string secret)
{
try
{
return _quickConnect.CheckRequestStatus(secret);
return await _quickConnect.CheckRequestStatus(secret).ConfigureAwait(false);
}
catch (ResourceNotFoundException)
{
+2 -2
View File
@@ -245,11 +245,11 @@ public class UserController : BaseJellyfinApiController
[HttpPost("AuthenticateWithQuickConnect")]
[ProducesResponseType(StatusCodes.Status200OK)]
[Tags("Authentication")]
public ActionResult<AuthenticationResult> AuthenticateWithQuickConnect([FromBody, Required] QuickConnectDto request)
public async Task<ActionResult<AuthenticationResult>> AuthenticateWithQuickConnect([FromBody, Required] QuickConnectDto request)
{
try
{
return _quickConnectManager.GetAuthorizedRequest(request.Secret);
return await _quickConnectManager.GetAuthorizedRequest(request.Secret).ConfigureAwait(false);
}
catch (SecurityException e)
{
+4
View File
@@ -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<ILyricProvider>())
{
serviceCollection.AddSingleton(typeof(ILyricProvider), type);
@@ -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;
/// <summary>
/// Extensions for registering the quick connect store.
/// </summary>
public static class QuickConnectStoreServiceCollectionExtensions
{
/// <summary>
/// Registers the quick connect store, Redis-backed when a connection string is configured and
/// process-local otherwise, and reports the selected store at <see cref="LogLevel.Information"/>.
/// </summary>
/// <remarks>
/// 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.
/// </remarks>
/// <param name="serviceCollection">The service collection.</param>
/// <param name="configuration">The configuration to read the Redis connection string from.</param>
/// <param name="logger">The logger to report the selected store on.</param>
/// <returns>The updated service collection.</returns>
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<IQuickConnectStore, InMemoryQuickConnectStore>();
}
logger.LogInformation(
"Quick connect store: {Store}. Quick connect flows complete across any instance.",
nameof(RedisQuickConnectStore));
return serviceCollection.AddSingleton<IQuickConnectStore>(sp =>
{
try
{
return new RedisQuickConnectStore(
sp.GetRequiredService<IConnectionMultiplexer>(),
sp.GetRequiredService<ILogger<RedisQuickConnectStore>>());
}
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<ILogger<CoreAppHost>>().LogError(
ex,
"Redis is configured but unavailable, so quick connect flows will not complete across instances. Check {Key}.",
TranscodeStoreOptions.RedisConnectionStringKey);
return new InMemoryQuickConnectStore();
}
});
}
}
@@ -21,14 +21,14 @@ namespace MediaBrowser.Controller.QuickConnect
/// </summary>
/// <param name="authorizationInfo">The initiator authorization info.</param>
/// <returns>A quick connect result with tokens to proceed or throws an exception if not active.</returns>
QuickConnectResult TryConnect(AuthorizationInfo authorizationInfo);
Task<QuickConnectResult> TryConnect(AuthorizationInfo authorizationInfo);
/// <summary>
/// Checks the status of an individual request.
/// </summary>
/// <param name="secret">Unique secret identifier of the request.</param>
/// <returns>Quick connect result.</returns>
QuickConnectResult CheckRequestStatus(string secret);
Task<QuickConnectResult> CheckRequestStatus(string secret);
/// <summary>
/// Authorizes a quick connect request to connect as the calling user.
@@ -43,6 +43,6 @@ namespace MediaBrowser.Controller.QuickConnect
/// </summary>
/// <param name="secret">The secret.</param>
/// <returns>The authentication result.</returns>
AuthenticationResult GetAuthorizedRequest(string secret);
Task<AuthenticationResult> GetAuthorizedRequest(string secret);
}
}
@@ -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;
/// <summary>
/// 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.
/// </summary>
public interface IQuickConnectStore
{
/// <summary>
/// Looks up a pending request by the secret handed to the initiating client.
/// </summary>
/// <param name="secret">The request secret.</param>
/// <param name="cancellationToken">A cancellation token.</param>
/// <returns>The request, or <c>null</c> when it is unknown or has expired.</returns>
Task<QuickConnectResult?> GetRequestBySecretAsync(string secret, CancellationToken cancellationToken = default);
/// <summary>
/// Looks up a pending request by the code shown to the user.
/// </summary>
/// <param name="code">The user facing code.</param>
/// <param name="cancellationToken">A cancellation token.</param>
/// <returns>The request, or <c>null</c> when it is unknown or has expired.</returns>
Task<QuickConnectResult?> GetRequestByCodeAsync(string code, CancellationToken cancellationToken = default);
/// <summary>
/// Stores a new or updated request until <paramref name="expiresUtc"/>.
/// </summary>
/// <param name="request">The request to store.</param>
/// <param name="expiresUtc">The instant the request stops being resolvable.</param>
/// <param name="cancellationToken">A cancellation token.</param>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
Task SetRequestAsync(QuickConnectResult request, DateTime expiresUtc, CancellationToken cancellationToken = default);
/// <summary>
/// Stores the authentication minted for an authorized request until <paramref name="expiresUtc"/>.
/// </summary>
/// <param name="secret">The request secret the client exchanges.</param>
/// <param name="authenticationResult">The authentication to hand out.</param>
/// <param name="expiresUtc">The instant the authentication stops being exchangeable.</param>
/// <param name="cancellationToken">A cancellation token.</param>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
Task SetAuthorizationAsync(string secret, AuthenticationResult authenticationResult, DateTime expiresUtc, CancellationToken cancellationToken = default);
/// <summary>
/// Atomically takes the authentication for <paramref name="secret"/> and removes it, so that two
/// instances racing on the same secret cannot both hand out an access token.
/// </summary>
/// <param name="secret">The request secret.</param>
/// <param name="cancellationToken">A cancellation token.</param>
/// <returns>The authentication, or <c>null</c> when the secret is unknown, expired or already exchanged.</returns>
Task<AuthenticationResult?> TryConsumeAuthorizationAsync(string secret, CancellationToken cancellationToken = default);
}
@@ -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;
/// <summary>
/// A process-local <see cref="IQuickConnectStore"/>. 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.
/// </summary>
public sealed class InMemoryQuickConnectStore : IQuickConnectStore
{
private readonly ConcurrentDictionary<string, Entry<QuickConnectResult>> _requests = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, Entry<AuthenticationResult>> _authorizations = new(StringComparer.Ordinal);
/// <inheritdoc />
public Task<QuickConnectResult?> GetRequestBySecretAsync(string secret, CancellationToken cancellationToken = default)
{
Expire();
return Task.FromResult(_requests.TryGetValue(secret, out var entry) ? entry.Value : null);
}
/// <inheritdoc />
public Task<QuickConnectResult?> 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)));
}
/// <inheritdoc />
public Task SetRequestAsync(QuickConnectResult request, DateTime expiresUtc, CancellationToken cancellationToken = default)
{
ArgumentNullException.ThrowIfNull(request);
Expire();
_requests[request.Secret] = new Entry<QuickConnectResult>(expiresUtc, request);
return Task.CompletedTask;
}
/// <inheritdoc />
public Task SetAuthorizationAsync(string secret, AuthenticationResult authenticationResult, DateTime expiresUtc, CancellationToken cancellationToken = default)
{
Expire();
_authorizations[secret] = new Entry<AuthenticationResult>(expiresUtc, authenticationResult);
return Task.CompletedTask;
}
/// <inheritdoc />
public Task<AuthenticationResult?> TryConsumeAuthorizationAsync(string secret, CancellationToken cancellationToken = default)
{
Expire();
if (!_authorizations.TryRemove(secret, out var entry) || entry.ExpiresUtc <= DateTime.UtcNow)
{
return Task.FromResult<AuthenticationResult?>(null);
}
return Task.FromResult<AuthenticationResult?>(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<T>(DateTime ExpiresUtc, T Value);
}
@@ -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<IQuickConnectStore>(new InMemoryQuickConnectStore());
// User object contains circular references.
_fixture.Behaviors.OfType<ThrowingRecursionBehavior>().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<ArgumentException>(() => _quickConnectManager.TryConnect(
public async Task TryConnect_InvalidAuthorizationInfo_ThrowsArgumentException(string device, string deviceId, string client, string version)
=> await Assert.ThrowsAsync<ArgumentException>(() => _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<AuthenticationException>(() => _quickConnectManager.TryConnect(_quickConnectAuthInfo));
await Assert.ThrowsAsync<AuthenticationException>(() => _quickConnectManager.TryConnect(_quickConnectAuthInfo));
}
[Fact]
public void CheckRequestStatus_QuickConnectUnavailable_ThrowsAuthenticationException()
public async Task CheckRequestStatus_QuickConnectUnavailable_ThrowsAuthenticationException()
{
_config.QuickConnectAvailable = false;
Assert.Throws<AuthenticationException>(() => _quickConnectManager.CheckRequestStatus(string.Empty));
await Assert.ThrowsAsync<AuthenticationException>(() => _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<AuthenticationException>(() => _quickConnectManager.GetAuthorizedRequest(string.Empty));
await Assert.ThrowsAsync<AuthenticationException>(() => _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<ResourceNotFoundException>(() => _quickConnectManager.CheckRequestStatus("Unknown secret"));
await Assert.ThrowsAsync<ResourceNotFoundException>(() => _quickConnectManager.CheckRequestStatus("Unknown secret"));
}
[Fact]
public void GetAuthorizedRequest_UnknownSecret_ThrowsResourceNotFoundException()
public async Task GetAuthorizedRequest_UnknownSecret_ThrowsResourceNotFoundException()
{
_config.QuickConnectAvailable = true;
Assert.Throws<ResourceNotFoundException>(() => _quickConnectManager.GetAuthorizedRequest("Unknown secret"));
await Assert.ThrowsAsync<ResourceNotFoundException>(() => _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<ResourceNotFoundException>(() => _quickConnectManager.GetAuthorizedRequest(res.Secret));
}
}
}
@@ -0,0 +1,91 @@
using System;
using System.Threading.Tasks;
using StackExchange.Redis;
using Testcontainers.Redis;
namespace Jellyfin.Server.Tests.HighAvailability;
/// <summary>
/// Hands out a Redis server for the tests that need one. A server named by <c>JELLYFIN_TEST_REDIS</c> 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.
/// </summary>
public sealed class RedisTestServer : IAsyncDisposable
{
/// <summary>
/// The connection string of an already running server.
/// </summary>
public const string ConnectionStringVariable = "JELLYFIN_TEST_REDIS";
private readonly RedisContainer? _container;
private RedisTestServer(RedisContainer? container, string connectionString)
{
_container = container;
ConnectionString = connectionString;
}
/// <summary>
/// Gets the connection string of the running server.
/// </summary>
public string ConnectionString { get; }
/// <summary>
/// Starts or attaches to a Redis server and waits until it accepts connections.
/// </summary>
/// <returns>The running server.</returns>
public static async Task<RedisTestServer> 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;
}
/// <summary>
/// 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.
/// </summary>
/// <returns>A new multiplexer.</returns>
public async Task<IConnectionMultiplexer> ConnectAsync()
=> await ConnectionMultiplexer.ConnectAsync(ConnectionString).ConfigureAwait(false);
/// <inheritdoc/>
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);
}
}
}
}
@@ -12,6 +12,7 @@
<PackageReference Include="Microsoft.NET.Test.Sdk" />
<PackageReference Include="Npgsql" />
<PackageReference Include="Testcontainers.PostgreSql" />
<PackageReference Include="Testcontainers.Redis" />
<PackageReference Include="xunit.v3" />
<PackageReference Include="xunit.runner.visualstudio">
<PrivateAssets>all</PrivateAssets>
@@ -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;
/// <summary>
/// Three independently constructed <see cref="QuickConnectManager"/> 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.
/// </summary>
[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<IConnectionMultiplexer> _connections = new();
private PostgreSqlTestServer _postgres = null!;
private RedisTestServer _redis = null!;
/// <inheritdoc/>
public async ValueTask InitializeAsync()
{
_postgres = await PostgreSqlTestServer.StartAsync().ConfigureAwait(false);
_redis = await RedisTestServer.StartAsync().ConfigureAwait(false);
}
/// <inheritdoc/>
public async ValueTask DisposeAsync()
{
foreach (var connection in _connections)
{
await connection.DisposeAsync().ConfigureAwait(false);
}
await _redis.DisposeAsync().ConfigureAwait(false);
await _postgres.DisposeAsync().ConfigureAwait(false);
}
/// <summary>
/// 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.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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);
}
/// <summary>
/// A secret is single use across the whole deployment: two replicas racing to exchange it must not
/// both hand out an access token.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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<ResourceNotFoundException>(() => replicaB.Manager.GetAuthorizedRequest(initiated.Secret));
}
/// <summary>
/// An expired request is rejected on a replica that never saw it created, rather than resolving to a
/// stale authorization.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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<ResourceNotFoundException>(() => replicaB.Manager.CheckRequestStatus(initiated.Secret));
await Assert.ThrowsAsync<ResourceNotFoundException>(() => replicaB.Manager.AuthorizeRequest(user.Id, initiated.Code));
}
/// <summary>
/// An authorization that was never exchanged expires too, so a code authorized and then abandoned
/// cannot be redeemed later from another replica.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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<ResourceNotFoundException>(() => replicaB.Manager.GetAuthorizedRequest(initiated.Secret));
}
private static async Task<AuthenticationResult?> ExchangeAsync(IQuickConnect manager, string secret)
{
try
{
return await manager.GetAuthorizedRequest(secret).ConfigureAwait(false);
}
catch (ResourceNotFoundException)
{
return null;
}
}
private static async Task<User> 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<JellyfinDbContext>();
var provider = new PostgreSqlDatabaseProvider(dataSource);
provider.Initialise(optionsBuilder, new DatabaseConfigurationOptions { DatabaseType = "PostgreSQL" });
return new JellyfinDbContext(
optionsBuilder.Options,
NullLogger<JellyfinDbContext>.Instance,
provider,
new NoLockBehavior(NullLogger<NoLockBehavior>.Instance));
}
private async Task<Replica> CreateReplicaAsync(NpgsqlDataSource dataSource, User user)
{
var connection = await _redis.ConnectAsync().ConfigureAwait(false);
_connections.Add(connection);
var userManager = new Mock<IUserManager>();
userManager.Setup(manager => manager.GetUserById(user.Id)).Returns(user);
var deviceManager = new DeviceManager(new DataSourceContextFactory(dataSource), userManager.Object);
var configManager = new Mock<IServerConfigurationManager>();
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<ISessionManager>();
sessionManager
.Setup(manager => manager.AuthenticateDirect(It.IsAny<AuthenticationRequest>()))
.Returns<AuthenticationRequest>(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<RedisQuickConnectStore>.Instance);
var manager = new QuickConnectManager(
configManager.Object,
NullLogger<QuickConnectManager>.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<JellyfinDbContext>
{
private readonly NpgsqlDataSource _dataSource;
public DataSourceContextFactory(NpgsqlDataSource dataSource)
{
_dataSource = dataSource;
}
public JellyfinDbContext CreateDbContext() => CreateContext(_dataSource);
}
}