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..2c619f4178 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,21 +132,27 @@ 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"); } if (result.Authenticated) { - throw new InvalidOperationException("Request is already authorized"); + throw new ConflictException("Request is already authorized"); } // Change the time on the request so it expires one minute into the future. It can't expire immediately as otherwise some clients wouldn't ever see that they have been authenticated. result.DateAdded = DateTime.UtcNow.Add(TimeSpan.FromMinutes(1)); + // The guard above is a read on shared state, so it cannot settle a race between instances; the claim can. + if (!await _store.TryClaimAuthorizationAsync(result.Secret, ExpiryOf(result)).ConfigureAwait(false)) + { + throw await RefusedClaimAsync(result.Secret).ConfigureAwait(false); + } + var authenticationResult = await _sessionManager.AuthenticateDirect(new AuthenticationRequest { UserId = userId, @@ -160,9 +162,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,17 +173,33 @@ 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.GetAuthorizationAsync(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); + + /// + /// Explains a refused claim. The claim outlives a failed mint on purpose, so it can mean either + /// that the request is authorized or that authorizing it did not finish; the two are told apart + /// by re-reading the request rather than reported as the same thing. + /// + private async Task RefusedClaimAsync(string secret) + { + var current = await _store.GetRequestBySecretAsync(secret).ConfigureAwait(false); + + return current?.Authenticated == true + ? new ConflictException("Request is already authorized") + : new ConflictException("Request is being authorized elsewhere, or an earlier attempt to authorize it did not complete. Start quick connect again for a new code."); } private string GenerateSecureRandom(int length = 32) @@ -190,42 +209,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..f536fdb55a --- /dev/null +++ b/Emby.Server.Implementations/QuickConnect/RedisQuickConnectStore.cs @@ -0,0 +1,164 @@ +using System; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using Jellyfin.Extensions.Json; +using MediaBrowser.Common.Extensions; +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 +/// claimed with a Lua check-and-set, so only one instance can ever mint a given secret's access token. +/// +/// +/// There is no local fallback: a call Redis did not answer is inconclusive, and reporting it as a miss +/// would tell a polling client its secret is invalid. Quick connect is unavailable for as long as Redis +/// is, which password login is not. +/// +public sealed class RedisQuickConnectStore : IQuickConnectStore +{ + private const string KeyPrefix = "jellyfin:quickconnect:"; + + /// + /// Lua script writing the two keys a request is resolvable by in one step, so it can never be + /// reachable by its secret while the code the user is reading off the screen resolves to nothing. + /// + private const string SetRequestScript = @" +redis.call('SET', KEYS[1], ARGV[1], 'PX', ARGV[3]) +redis.call('SET', KEYS[2], ARGV[2], 'PX', ARGV[3]) +return 1"; + + /// + /// Lua script for the atomic claim of the sole right to authorize a request: the request has to + /// exist and not already be authorized, and the claim marker is taken with SET NX, so of two + /// instances racing on one code exactly one goes on to mint an access token. + /// + private const string ClaimAuthorizationScript = @" +local raw = redis.call('GET', KEYS[1]) +if not raw then return 0 end +if cjson.decode(raw)['Authenticated'] then return 0 end +if redis.call('SET', KEYS[2], '1', 'NX', 'PX', ARGV[1]) then return 1 end +return 0"; + + private readonly IDatabase _db; + 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(); + _logger = logger; + } + + /// + public async Task GetRequestBySecretAsync(string secret, CancellationToken cancellationToken = default) + { + var raw = await CallAsync(() => _db.StringGetAsync(RequestKey(secret))).ConfigureAwait(false); + + // Deserialization is outside the guard: a malformed stored value is a fault of its own, not Redis + // being unavailable. + return raw.HasValue ? JsonSerializer.Deserialize(raw.ToString(), JsonDefaults.Options) : null; + } + + /// + public async Task GetRequestByCodeAsync(string code, CancellationToken cancellationToken = default) + { + var secret = await CallAsync(() => _db.StringGetAsync(CodeKey(code))).ConfigureAwait(false); + + return secret.HasValue + ? await GetRequestBySecretAsync(secret.ToString(), cancellationToken).ConfigureAwait(false) + : null; + } + + /// + public async Task SetRequestAsync(QuickConnectResult request, DateTime expiresUtc, CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(request); + + var ttl = expiresUtc - DateTime.UtcNow; + if (ttl <= TimeSpan.Zero) + { + return; + } + + var json = JsonSerializer.Serialize(request, JsonDefaults.Options); + await CallAsync(() => _db.ScriptEvaluateAsync( + SetRequestScript, + keys: new RedisKey[] { RequestKey(request.Secret), CodeKey(request.Code) }, + values: new RedisValue[] { json, request.Secret, (long)ttl.TotalMilliseconds })).ConfigureAwait(false); + } + + /// + public async Task TryClaimAuthorizationAsync(string secret, DateTime expiresUtc, CancellationToken cancellationToken = default) + { + var ttl = expiresUtc - DateTime.UtcNow; + if (ttl <= TimeSpan.Zero) + { + return false; + } + + var claimed = (long?)await CallAsync(() => _db.ScriptEvaluateAsync( + ClaimAuthorizationScript, + keys: new RedisKey[] { RequestKey(secret), ClaimKey(secret) }, + values: new RedisValue[] { (long)ttl.TotalMilliseconds })).ConfigureAwait(false); + + return claimed == 1; + } + + /// + public async Task SetAuthorizationAsync(string secret, AuthenticationResult authenticationResult, DateTime expiresUtc, CancellationToken cancellationToken = default) + { + var ttl = expiresUtc - DateTime.UtcNow; + if (ttl <= TimeSpan.Zero) + { + return; + } + + var json = JsonSerializer.Serialize(authenticationResult, JsonDefaults.Options); + await CallAsync(() => _db.StringSetAsync(AuthorizationKey(secret), json, ttl)).ConfigureAwait(false); + } + + /// + public async Task GetAuthorizationAsync(string secret, CancellationToken cancellationToken = default) + { + var raw = await CallAsync(() => _db.StringGetAsync(AuthorizationKey(secret))).ConfigureAwait(false); + + return raw.HasValue + ? JsonSerializer.Deserialize(raw.ToString(), JsonDefaults.Options) + : null; + } + + private static string RequestKey(string secret) => KeyPrefix + "request:" + secret; + + private static string CodeKey(string code) => KeyPrefix + "code:" + code; + + private static string ClaimKey(string secret) => KeyPrefix + "claim:" + secret; + + private static string AuthorizationKey(string secret) => KeyPrefix + "auth:" + secret; + + private async Task CallAsync(Func> call) + { + try + { + return await call().ConfigureAwait(false); + } + catch (Exception exception) when (exception is RedisException or RedisCommandException or TimeoutException) + { + _logger.LogError(exception, "Quick connect state could not be reached in Redis."); + throw new ServiceUnavailableException("Quick connect is temporarily unavailable.", exception); + } + } +} diff --git a/Jellyfin.Api/Controllers/QuickConnectController.cs b/Jellyfin.Api/Controllers/QuickConnectController.cs index 5c7b38e137..54424deb06 100644 --- a/Jellyfin.Api/Controllers/QuickConnectController.cs +++ b/Jellyfin.Api/Controllers/QuickConnectController.cs @@ -50,16 +50,18 @@ public class QuickConnectController : BaseJellyfinApiController /// /// Quick connect request successfully created. /// Quick connect is not active on this server. + /// Quick connect state is unavailable. /// A with a secret and code for future use or an error message. [HttpPost("Initiate")] [ProducesResponseType(StatusCodes.Status200OK)] [ProducesResponseType(StatusCodes.Status401Unauthorized)] + [ProducesResponseType(StatusCodes.Status503ServiceUnavailable)] public async Task> InitiateQuickConnect() { try { var auth = await _authContext.GetAuthorizationInfo(Request).ConfigureAwait(false); - return _quickConnect.TryConnect(auth); + return await _quickConnect.TryConnect(auth).ConfigureAwait(false); } catch (AuthenticationException) { @@ -73,15 +75,17 @@ public class QuickConnectController : BaseJellyfinApiController /// Secret previously returned from the Initiate endpoint. /// Quick connect result returned. /// Unknown quick connect secret. + /// Quick connect state is unavailable. /// An updated . [HttpGet("Connect")] [ProducesResponseType(StatusCodes.Status200OK)] [ProducesResponseType(StatusCodes.Status404NotFound)] - public ActionResult GetQuickConnectState([FromQuery, Required] string secret) + [ProducesResponseType(StatusCodes.Status503ServiceUnavailable)] + public async Task> GetQuickConnectState([FromQuery, Required] string secret) { try { - return _quickConnect.CheckRequestStatus(secret); + return await _quickConnect.CheckRequestStatus(secret).ConfigureAwait(false); } catch (ResourceNotFoundException) { @@ -100,11 +104,15 @@ public class QuickConnectController : BaseJellyfinApiController /// The user the authorize. Access to the requested user is required. /// Quick connect result authorized successfully. /// Unknown user id. + /// Request is already authorized, or authorizing it did not complete. + /// Quick connect state is unavailable. /// Boolean indicating if the authorization was successful. [HttpPost("Authorize")] [Authorize] [ProducesResponseType(StatusCodes.Status200OK)] [ProducesResponseType(StatusCodes.Status403Forbidden)] + [ProducesResponseType(StatusCodes.Status409Conflict)] + [ProducesResponseType(StatusCodes.Status503ServiceUnavailable)] public async Task> AuthorizeQuickConnect([FromQuery, Required] string code, [FromQuery] Guid? userId = null) { userId = RequestHelpers.GetUserId(User, userId); diff --git a/Jellyfin.Api/Controllers/UserController.cs b/Jellyfin.Api/Controllers/UserController.cs index 657bda4d15..48e473557b 100644 --- a/Jellyfin.Api/Controllers/UserController.cs +++ b/Jellyfin.Api/Controllers/UserController.cs @@ -241,15 +241,19 @@ public class UserController : BaseJellyfinApiController /// The request. /// User authenticated. /// Missing token. + /// Unknown or unauthorized quick connect secret. + /// Quick connect state is unavailable. /// A containing an with information about the new session. [HttpPost("AuthenticateWithQuickConnect")] [ProducesResponseType(StatusCodes.Status200OK)] + [ProducesResponseType(StatusCodes.Status404NotFound)] + [ProducesResponseType(StatusCodes.Status503ServiceUnavailable)] [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.Api/Middleware/ExceptionMiddleware.cs b/Jellyfin.Api/Middleware/ExceptionMiddleware.cs index acbb4877d4..8f5da04575 100644 --- a/Jellyfin.Api/Middleware/ExceptionMiddleware.cs +++ b/Jellyfin.Api/Middleware/ExceptionMiddleware.cs @@ -131,6 +131,8 @@ public class ExceptionMiddleware FileNotFoundException => StatusCodes.Status404NotFound, ResourceNotFoundException => StatusCodes.Status404NotFound, MethodNotAllowedException => StatusCodes.Status405MethodNotAllowed, + ConflictException => StatusCodes.Status409Conflict, + ServiceUnavailableException => StatusCodes.Status503ServiceUnavailable, _ => StatusCodes.Status500InternalServerError }; } 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..aa13634a9e --- /dev/null +++ b/Jellyfin.Server/Extensions/QuickConnectStoreServiceCollectionExtensions.cs @@ -0,0 +1,57 @@ +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. Set but + /// unreachable is a misconfigured deployment rather than a single-instance one, so it fails rather + /// than quietly handing out a store the other instances cannot see. + /// + /// 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 => new RedisQuickConnectStore( + sp.GetRequiredService(), + sp.GetRequiredService>())); + } +} diff --git a/MediaBrowser.Common/Extensions/ConflictException.cs b/MediaBrowser.Common/Extensions/ConflictException.cs new file mode 100644 index 0000000000..433fc13643 --- /dev/null +++ b/MediaBrowser.Common/Extensions/ConflictException.cs @@ -0,0 +1,36 @@ +using System; + +namespace MediaBrowser.Common.Extensions +{ + /// + /// Thrown when the current state of a resource does not allow the requested operation. + /// + public class ConflictException : Exception + { + /// + /// Initializes a new instance of the class. + /// + public ConflictException() + { + } + + /// + /// Initializes a new instance of the class. + /// + /// The message. + public ConflictException(string message) + : base(message) + { + } + + /// + /// Initializes a new instance of the class. + /// + /// The message. + /// The exception that caused this one. + public ConflictException(string message, Exception innerException) + : base(message, innerException) + { + } + } +} diff --git a/MediaBrowser.Common/Extensions/ServiceUnavailableException.cs b/MediaBrowser.Common/Extensions/ServiceUnavailableException.cs new file mode 100644 index 0000000000..34fc8049d6 --- /dev/null +++ b/MediaBrowser.Common/Extensions/ServiceUnavailableException.cs @@ -0,0 +1,37 @@ +using System; + +namespace MediaBrowser.Common.Extensions +{ + /// + /// Thrown when an operation cannot be answered because a backing service is unreachable, rather than + /// because the thing it was asked about does not exist. + /// + public class ServiceUnavailableException : Exception + { + /// + /// Initializes a new instance of the class. + /// + public ServiceUnavailableException() + { + } + + /// + /// Initializes a new instance of the class. + /// + /// The message. + public ServiceUnavailableException(string message) + : base(message) + { + } + + /// + /// Initializes a new instance of the class. + /// + /// The message. + /// The exception that caused this one. + public ServiceUnavailableException(string message, Exception innerException) + : base(message, innerException) + { + } + } +} diff --git a/MediaBrowser.Controller/QuickConnect/IQuickConnect.cs b/MediaBrowser.Controller/QuickConnect/IQuickConnect.cs index ec3706773a..eba710cd81 100644 --- a/MediaBrowser.Controller/QuickConnect/IQuickConnect.cs +++ b/MediaBrowser.Controller/QuickConnect/IQuickConnect.cs @@ -1,5 +1,6 @@ using System; using System.Threading.Tasks; +using MediaBrowser.Common.Extensions; using MediaBrowser.Controller.Authentication; using MediaBrowser.Controller.Net; using MediaBrowser.Model.QuickConnect; @@ -21,17 +22,19 @@ 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. + /// Authorizes a quick connect request to connect as the calling user. A request can be authorized + /// once: a second attempt, including one following an attempt that failed part way, throws + /// and the user has to start quick connect again for a new code. /// /// User id. /// Identifying code for the request. @@ -39,10 +42,11 @@ namespace MediaBrowser.Controller.QuickConnect Task AuthorizeRequest(Guid userId, string code); /// - /// Gets the authorized request for the secret. + /// Gets the authorized request for the secret. The read does not consume the authorization, so a + /// client that retries the exchange gets the same access token until the authorization expires. /// /// 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..fd2254eb17 --- /dev/null +++ b/MediaBrowser.Controller/QuickConnect/IQuickConnectStore.cs @@ -0,0 +1,77 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using MediaBrowser.Common.Extensions; +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. +/// +/// +/// A shared implementation that cannot reach its backend throws +/// rather than reporting a miss, because a miss tells a polling client its secret is invalid. +/// +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 , resolvable by both its secret + /// and its code or by neither. A request already past is not stored. + /// + /// 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); + + /// + /// Atomically claims the sole right to authorize the request behind , so + /// that two instances racing on one code cannot both mint an access token. The claim is never + /// released: a mint that failed after writing its token would otherwise be retried into a second one, + /// so a request whose authorization failed has to be started again. + /// + /// The request secret. + /// The instant the claim lapses, after which the request can be authorized again. + /// A cancellation token. + /// true when this caller may go on to authorize the request; false when it is unknown, expired, already authorized or claimed elsewhere. + Task TryClaimAuthorizationAsync(string secret, 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); + + /// + /// Reads the authentication for . The read does not consume it, so a client + /// that retries an exchange gets the same access token for as long as the authentication lives. + /// + /// The request secret. + /// A cancellation token. + /// The authentication, or null when the secret is unknown or has expired. + Task GetAuthorizationAsync(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..0eef358730 --- /dev/null +++ b/MediaBrowser.Controller/QuickConnect/InMemoryQuickConnectStore.cs @@ -0,0 +1,113 @@ +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 , and the single-instance default. Nothing it holds +/// is visible to another instance, so a deployment running more than one has to configure a shared +/// store instead. +/// +public sealed class InMemoryQuickConnectStore : IQuickConnectStore +{ + private readonly ConcurrentDictionary> _requests = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary> _authorizations = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary _authorizationClaims = 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(); + if (expiresUtc > DateTime.UtcNow) + { + _requests[request.Secret] = new Entry(expiresUtc, request); + } + + return Task.CompletedTask; + } + + /// + public Task TryClaimAuthorizationAsync(string secret, DateTime expiresUtc, CancellationToken cancellationToken = default) + { + Expire(); + if (!_requests.TryGetValue(secret, out var entry) || entry.Value.Authenticated) + { + return Task.FromResult(false); + } + + return Task.FromResult(_authorizationClaims.TryAdd(secret, expiresUtc)); + } + + /// + public Task SetAuthorizationAsync(string secret, AuthenticationResult authenticationResult, DateTime expiresUtc, CancellationToken cancellationToken = default) + { + Expire(); + if (expiresUtc > DateTime.UtcNow) + { + _authorizations[secret] = new Entry(expiresUtc, authenticationResult); + } + + return Task.CompletedTask; + } + + /// + public Task GetAuthorizationAsync(string secret, CancellationToken cancellationToken = default) + { + Expire(); + return Task.FromResult(_authorizations.TryGetValue(secret, out var entry) ? entry.Value : null); + } + + 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 _); + } + } + + foreach (var (secret, expiresUtc) in _authorizationClaims) + { + if (expiresUtc <= now) + { + _authorizationClaims.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..512a42ed77 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,83 @@ 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 AuthorizeRequest_RacedOnOneCode_SucceedsOnce() + { + _config.QuickConnectAvailable = true; + var res = await _quickConnectManager.TryConnect(_quickConnectAuthInfo); + + var outcomes = await Task.WhenAll( + Task.Run(() => AuthorizeAsync(res.Code)), + Task.Run(() => AuthorizeAsync(res.Code))); + + Assert.Single(outcomes, authorized => authorized); + } + + [Fact] + public async Task GetAuthorizedRequest_SecondExchange_ReturnsTheSameResult() + { + _config.QuickConnectAvailable = true; + var res = await _quickConnectManager.TryConnect(_quickConnectAuthInfo); + await _quickConnectManager.AuthorizeRequest(Guid.Empty, res.Code); + + var first = await _quickConnectManager.GetAuthorizedRequest(res.Secret); + var second = await _quickConnectManager.GetAuthorizedRequest(res.Secret); + + Assert.Same(first, second); + } + + [Fact] + public async Task AuthorizeRequest_OfAnAuthorizedRequest_ThrowsConflictException() + { + _config.QuickConnectAvailable = true; + var res = await _quickConnectManager.TryConnect(_quickConnectAuthInfo); + await _quickConnectManager.AuthorizeRequest(Guid.Empty, res.Code); + + await Assert.ThrowsAsync(() => _quickConnectManager.AuthorizeRequest(Guid.Empty, res.Code)); + } + + private async Task AuthorizeAsync(string code) + { + try + { + return await _quickConnectManager.AuthorizeRequest(Guid.Empty, code).ConfigureAwait(false); + } + catch (ConflictException) + { + return false; + } + } } } diff --git a/tests/Jellyfin.Server.Tests/HighAvailability/JellyfinSectionConfigurationTests.cs b/tests/Jellyfin.Server.Tests/HighAvailability/JellyfinSectionConfigurationTests.cs index 4cfc9518ed..45f6d598aa 100644 --- a/tests/Jellyfin.Server.Tests/HighAvailability/JellyfinSectionConfigurationTests.cs +++ b/tests/Jellyfin.Server.Tests/HighAvailability/JellyfinSectionConfigurationTests.cs @@ -13,6 +13,7 @@ namespace Jellyfin.Server.Tests.HighAvailability; /// startup configuration the host reads them from must therefore accept that form; when it does not, /// a correctly set variable is dropped and the feature it configures stays off without any error. /// +[Collection("JellyfinSectionConfiguration")] public sealed class JellyfinSectionConfigurationTests : IDisposable { private const string RedisKey = "Jellyfin:TranscodeStore:RedisConnectionString"; diff --git a/tests/Jellyfin.Server.Tests/HighAvailability/RedisFaultProxy.cs b/tests/Jellyfin.Server.Tests/HighAvailability/RedisFaultProxy.cs new file mode 100644 index 0000000000..6c4c6cb4a3 --- /dev/null +++ b/tests/Jellyfin.Server.Tests/HighAvailability/RedisFaultProxy.cs @@ -0,0 +1,223 @@ +using System; +using System.Collections.Concurrent; +using System.Globalization; +using System.IO; +using System.Net; +using System.Net.Sockets; +using System.Text; +using System.Threading; +using System.Threading.Tasks; +using StackExchange.Redis; + +namespace Jellyfin.Server.Tests.HighAvailability; + +/// +/// A loopback TCP proxy in front of a Redis server. Cutting it drops every connection through it and +/// refuses new ones, so a test can take Redis away from one instance mid-flow - and give it back - the +/// way a restarted valkey does, and watch what a real StackExchange.Redis client makes of it. +/// +public sealed class RedisFaultProxy : IAsyncDisposable +{ + private readonly ConcurrentDictionary _live = new(); + private readonly CancellationTokenSource _cts = new(); + private readonly TcpListener _listener; + private readonly string _targetHost; + private readonly int _targetPort; + private readonly int _port; + + private volatile bool _cut; + private volatile byte[]? _cutAfterMarker; + + private RedisFaultProxy(TcpListener listener, int port, string targetHost, int targetPort) + { + _listener = listener; + _port = port; + _targetHost = targetHost; + _targetPort = targetPort; + } + + /// + /// Gets a connection string pointing at the proxy. The timeouts are short so a cut surfaces as a + /// failure in seconds rather than in the library's minute-scale defaults. + /// + public string ConnectionString => string.Create( + CultureInfo.InvariantCulture, + $"127.0.0.1:{_port},abortConnect=false,connectTimeout=500,syncTimeout=2000,connectRetry=1"); + + /// + /// Starts a proxy in front of the server named by . + /// + /// The connection string of the server to forward to. + /// The running proxy. + public static RedisFaultProxy Start(string target) + { + var endpoint = ConfigurationOptions.Parse(target).EndPoints[0]; + var (host, port) = endpoint switch + { + DnsEndPoint dns => (dns.Host, dns.Port), + IPEndPoint ip => (ip.Address.ToString(), ip.Port), + _ => throw new NotSupportedException("Unsupported endpoint " + endpoint) + }; + + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + + var proxy = new RedisFaultProxy(listener, ((IPEndPoint)listener.LocalEndpoint).Port, host, port); + _ = Task.Run(proxy.AcceptAsync); + return proxy; + } + + /// + /// Takes Redis away from everything connected through the proxy. + /// + public void Cut() + { + _cut = true; + DropLiveConnections(); + } + + /// + /// Arms a cut for the moment after a command containing has been forwarded + /// and answered, so a test can take Redis away between two round trips of one operation rather than + /// only before or after all of them. + /// + /// Text that identifies the command to cut after. + public void CutAfterForwarding(string marker) => _cutAfterMarker = Encoding.UTF8.GetBytes(marker); + + /// + /// Lets connections through again. Clients reconnect on their own schedule, so callers have to wait + /// for the connection to come back rather than assume it already has. + /// + public void Restore() + { + _cutAfterMarker = null; + _cut = false; + } + + /// + public async ValueTask DisposeAsync() + { + _cut = true; + await _cts.CancelAsync().ConfigureAwait(false); + _listener.Stop(); + DropLiveConnections(); + _cts.Dispose(); + } + + private void DropLiveConnections() + { + foreach (var client in _live.Keys) + { + if (_live.TryRemove(client, out _)) + { + client.Dispose(); + } + } + } + + private async Task AcceptAsync() + { + while (!_cts.IsCancellationRequested) + { + TcpClient client; + try + { + client = await _listener.AcceptTcpClientAsync(_cts.Token).ConfigureAwait(false); + } + catch (Exception exception) when (exception is OperationCanceledException or SocketException or ObjectDisposedException) + { + return; + } + + if (_cut) + { + client.Dispose(); + continue; + } + + _ = Task.Run(() => ForwardAsync(client)); + } + } + + private async Task ForwardAsync(TcpClient client) + { + TcpClient? upstream = null; + try + { + upstream = new TcpClient(); + await upstream.ConnectAsync(_targetHost, _targetPort, _cts.Token).ConfigureAwait(false); + + _live[client] = 0; + _live[upstream] = 0; + + // Registered first, then rechecked: a cut concurrent with this connect would otherwise drop + // the live connections before this pair joined them and leave it running through the outage. + if (_cut) + { + return; + } + + var clientStream = client.GetStream(); + var upstreamStream = upstream.GetStream(); + await Task.WhenAny( + CopyFromClientAsync(clientStream, upstreamStream), + CopyAsync(upstreamStream, clientStream)).ConfigureAwait(false); + } + catch (Exception exception) when (exception is IOException or SocketException or OperationCanceledException or ObjectDisposedException) + { + } + finally + { + _live.TryRemove(client, out _); + client.Dispose(); + if (upstream is not null) + { + _live.TryRemove(upstream, out _); + upstream.Dispose(); + } + } + } + + private async Task CopyFromClientAsync(NetworkStream from, NetworkStream to) + { + var buffer = new byte[16 * 1024]; + try + { + while (true) + { + var read = await from.ReadAsync(buffer, _cts.Token).ConfigureAwait(false); + if (read == 0) + { + return; + } + + await to.WriteAsync(buffer.AsMemory(0, read), _cts.Token).ConfigureAwait(false); + + var marker = _cutAfterMarker; + if (marker is not null && buffer.AsSpan(0, read).IndexOf(marker) >= 0) + { + _cutAfterMarker = null; + + // Long enough for the server to have applied the command that was just forwarded. + await Task.Delay(TimeSpan.FromMilliseconds(250), _cts.Token).ConfigureAwait(false); + Cut(); + return; + } + } + } + catch (Exception exception) when (exception is IOException or SocketException or OperationCanceledException or ObjectDisposedException) + { + } + } + + private async Task CopyAsync(NetworkStream from, NetworkStream to) + { + try + { + await from.CopyToAsync(to, _cts.Token).ConfigureAwait(false); + } + catch (Exception exception) when (exception is IOException or SocketException or OperationCanceledException or ObjectDisposedException) + { + } + } +} 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..31341139cc --- /dev/null +++ b/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectReplicaTests.cs @@ -0,0 +1,374 @@ +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); + } + + /// + /// Exchanging a secret does not spend it: a client that retries, or whose retry lands on another + /// replica, gets the same access token back rather than a 404, and the device is minted once. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Exchange_RepeatedOnTwoReplicas_ReturnsTheSameToken() + { + var cancellationToken = TestContext.Current.CancellationToken; + var connectionString = await _postgres.CreateDatabaseAsync("quickconnect_replica_reexchange", 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); + + for (var attempt = 0; attempt < 10; attempt++) + { + var authorizationInfo = AuthorizationInfoFor(attempt); + var initiated = await replicaA.Manager.TryConnect(authorizationInfo); + await replicaB.Manager.AuthorizeRequest(user.Id, initiated.Code); + + var exchanged = await Task.WhenAll( + Task.Run(() => ExchangeAsync(replicaA.Manager, initiated.Secret), cancellationToken), + Task.Run(() => ExchangeAsync(replicaC.Manager, initiated.Secret), cancellationToken)); + + Assert.All(exchanged, outcome => Assert.NotNull(outcome)); + Assert.Equal(exchanged[0]!.AccessToken, exchanged[1]!.AccessToken); + + // Still there afterwards, on a replica that has not exchanged it yet. + var later = await replicaB.Manager.GetAuthorizedRequest(initiated.Secret); + Assert.Equal(exchanged[0]!.AccessToken, later.AccessToken); + + var devices = await replicaA.Devices.GetDevices(new DeviceQuery { DeviceId = authorizationInfo.DeviceId }); + Assert.Equal(later.AccessToken, Assert.Single(devices.Items).AccessToken); + } + } + + /// + /// Two replicas authorizing one code at the same time mint one access token between them. A second + /// one would be live, attached to the same device and reachable by nobody. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Authorize_RacedOnTwoReplicas_MintsOneAccessToken() + { + var cancellationToken = TestContext.Current.CancellationToken; + var connectionString = await _postgres.CreateDatabaseAsync("quickconnect_replica_authorize_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); + + for (var attempt = 0; attempt < 20; attempt++) + { + var authorizationInfo = AuthorizationInfoFor(attempt); + var initiated = await replicaA.Manager.TryConnect(authorizationInfo); + + var outcomes = await Task.WhenAll( + Task.Run(() => AuthorizeAsync(replicaB.Manager, user.Id, initiated.Code), cancellationToken), + Task.Run(() => AuthorizeAsync(replicaC.Manager, user.Id, initiated.Code), cancellationToken)); + + Assert.Single(outcomes, authorized => authorized); + + var devices = await replicaA.Devices.GetDevices(new DeviceQuery { DeviceId = authorizationInfo.DeviceId }); + var device = Assert.Single(devices.Items); + + var exchanged = await replicaA.Manager.GetAuthorizedRequest(initiated.Secret); + Assert.Equal(device.AccessToken, exchanged.AccessToken); + } + } + + /// + /// 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 AuthorizationInfo AuthorizationInfoFor(int attempt) => new AuthorizationInfo + { + Device = _authorizationInfo.Device, + DeviceId = string.Create(CultureInfo.InvariantCulture, $"device-{attempt}"), + Client = _authorizationInfo.Client, + Version = _authorizationInfo.Version + }; + + private static async Task AuthorizeAsync(IQuickConnect manager, Guid userId, string code) + { + try + { + return await manager.AuthorizeRequest(userId, code).ConfigureAwait(false); + } + catch (ConflictException) + { + return false; + } + } + + 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); + } +} diff --git a/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectStatusCodeTests.cs b/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectStatusCodeTests.cs new file mode 100644 index 0000000000..55d3284de7 --- /dev/null +++ b/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectStatusCodeTests.cs @@ -0,0 +1,261 @@ +using System; +using System.IO; +using System.Threading.Tasks; +using Emby.Server.Implementations.QuickConnect; +using Jellyfin.Api.Controllers; +using Jellyfin.Api.Middleware; +using Jellyfin.Server.Tests.HighAvailability; +using MediaBrowser.Common.Extensions; +using MediaBrowser.Controller; +using MediaBrowser.Controller.Authentication; +using MediaBrowser.Controller.Configuration; +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.AspNetCore.Hosting; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Mvc; +using Microsoft.AspNetCore.Mvc.Infrastructure; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using StackExchange.Redis; +using Xunit; + +namespace Jellyfin.Server.Tests.QuickConnect; + +/// +/// The status a client actually sees. A missing key and an unreachable valkey are different answers and +/// must not collapse into one: telling a polling client its secret is unknown ends its flow, while 503 +/// tells it to keep trying. The exception the store really throws is run through the real exception +/// middleware, so the mapping is exercised rather than assumed. +/// +[Trait("Category", "RequiresDocker")] +public sealed class QuickConnectStatusCodeTests : IAsyncLifetime +{ + private readonly Mock _sessionManager = new(); + + private RedisTestServer _redis = null!; + private RedisFaultProxy _proxy = null!; + private IConnectionMultiplexer _connection = null!; + private QuickConnectManager _manager = null!; + private QuickConnectController _controller = null!; + + /// + public async ValueTask InitializeAsync() + { + _redis = await RedisTestServer.StartAsync().ConfigureAwait(false); + _proxy = RedisFaultProxy.Start(_redis.ConnectionString); + _connection = await ConnectionMultiplexer.ConnectAsync(_proxy.ConnectionString).ConfigureAwait(false); + + var configManager = new Mock(); + configManager.Setup(manager => manager.Configuration).Returns(new ServerConfiguration { QuickConnectAvailable = true }); + + _manager = new QuickConnectManager( + configManager.Object, + NullLogger.Instance, + _sessionManager.Object, + new RedisQuickConnectStore(_connection, NullLogger.Instance)); + + _controller = new QuickConnectController(_manager, Mock.Of()); + } + + /// + public async ValueTask DisposeAsync() + { + await _connection.DisposeAsync().ConfigureAwait(false); + await _proxy.DisposeAsync().ConfigureAwait(false); + await _redis.DisposeAsync().ConfigureAwait(false); + } + + /// + /// A secret valkey has never heard of is a 404, which is what ends a flow the user abandoned. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Poll_UnknownSecret_IsNotFound() + { + Assert.Equal( + StatusCodes.Status404NotFound, + await StatusCodeAsync(async () => StatusOf(await _controller.GetQuickConnectState(NewSecret())))); + } + + /// + /// The same poll while valkey is unreachable is a 503. This is the bug the shared store is here to + /// avoid: a blip must not tell every polling client that its secret is invalid. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Poll_WhileRedisIsUnreachable_IsServiceUnavailable() + { + var secret = await InitiateAsync(); + _proxy.Cut(); + + Assert.Equal( + StatusCodes.Status503ServiceUnavailable, + await StatusCodeAsync(async () => StatusOf(await _controller.GetQuickConnectState(secret)))); + } + + /// + /// The exchange leg tells the two apart the same way. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Exchange_UnknownSecret_IsNotFound() + { + Assert.Equal( + StatusCodes.Status404NotFound, + await StatusCodeAsync(async () => + { + await _manager.GetAuthorizedRequest(NewSecret()).ConfigureAwait(false); + return StatusCodes.Status200OK; + })); + } + + /// + /// The exchange leg while valkey is unreachable. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Exchange_WhileRedisIsUnreachable_IsServiceUnavailable() + { + var secret = await InitiateAsync(); + _proxy.Cut(); + + Assert.Equal( + StatusCodes.Status503ServiceUnavailable, + await StatusCodeAsync(async () => + { + await _manager.GetAuthorizedRequest(secret).ConfigureAwait(false); + return StatusCodes.Status200OK; + })); + } + + /// + /// The authorize leg while valkey is unreachable. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Authorize_WhileRedisIsUnreachable_IsServiceUnavailable() + { + var initiated = await _manager.TryConnect(AuthorizationInfo()); + _proxy.Cut(); + + Assert.Equal( + StatusCodes.Status503ServiceUnavailable, + await StatusCodeAsync(async () => + { + await _manager.AuthorizeRequest(Guid.NewGuid(), initiated.Code).ConfigureAwait(false); + return StatusCodes.Status200OK; + })); + } + + /// + /// A mint that threw leaves its claim taken on purpose, because the write it failed on may have + /// landed. Retrying then has to say so and be a 409 the client can act on, not a 500 and not the + /// untrue claim that the request is already authorized. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Authorize_AfterAMintThatFailed_IsConflictAndSaysToStartAgain() + { + var initiated = await _manager.TryConnect(AuthorizationInfo()); + + _sessionManager + .Setup(manager => manager.AuthenticateDirect(It.IsAny())) + .ThrowsAsync(new InvalidOperationException("mint failed")); + + await Assert.ThrowsAsync(() => _manager.AuthorizeRequest(Guid.NewGuid(), initiated.Code)); + + var retry = await Record.ExceptionAsync(() => _manager.AuthorizeRequest(Guid.NewGuid(), initiated.Code)); + + var conflict = Assert.IsType(retry); + Assert.DoesNotContain("already authorized", conflict.Message, StringComparison.Ordinal); + Assert.Contains("Start quick connect again", conflict.Message, StringComparison.Ordinal); + + Assert.Equal( + StatusCodes.Status409Conflict, + await StatusCodeAsync(async () => + { + await _manager.AuthorizeRequest(Guid.NewGuid(), initiated.Code).ConfigureAwait(false); + return StatusCodes.Status200OK; + })); + } + + /// + /// A request that really was authorized still says so, so the accurate message above is not just a + /// blanket replacement for the old one. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Authorize_OfAnAuthorizedRequest_IsConflictAndSaysAlreadyAuthorized() + { + var initiated = await _manager.TryConnect(AuthorizationInfo()); + var userId = Guid.NewGuid(); + + _sessionManager + .Setup(manager => manager.AuthenticateDirect(It.IsAny())) + .ReturnsAsync(new AuthenticationResult + { + AccessToken = "token-1", + ServerId = "server-1", + User = new UserDto { Id = userId, Name = "user", ServerId = "server-1" } + }); + + Assert.True(await _manager.AuthorizeRequest(userId, initiated.Code)); + + var retry = await Record.ExceptionAsync(() => _manager.AuthorizeRequest(userId, initiated.Code)); + + Assert.Equal("Request is already authorized", Assert.IsType(retry).Message); + } + + private static AuthorizationInfo AuthorizationInfo() => new AuthorizationInfo + { + Device = "Living Room TV", + DeviceId = Guid.NewGuid().ToString("N"), + Client = "Jellyfin Web", + Version = "1.0.0" + }; + + private static string NewSecret() => Guid.NewGuid().ToString("N"); + + private static int StatusOf(ActionResult result) + => result.Result is IStatusCodeActionResult status + ? status.StatusCode ?? StatusCodes.Status200OK + : StatusCodes.Status200OK; + + private static async Task StatusCodeAsync(Func> action) + { + var appPaths = new Mock(); + appPaths.Setup(paths => paths.ProgramSystemPath).Returns("/program"); + appPaths.Setup(paths => paths.ProgramDataPath).Returns("/data"); + + var configManager = new Mock(); + configManager.Setup(manager => manager.ApplicationPaths).Returns(appPaths.Object); + + var hostEnvironment = new Mock(); + hostEnvironment.SetupGet(environment => environment.EnvironmentName).Returns(Environments.Production); + + var context = new DefaultHttpContext(); + context.Response.Body = new MemoryStream(); + + var middleware = new ExceptionMiddleware( + async _ => + { + context.Response.StatusCode = await action().ConfigureAwait(false); + }, + NullLogger.Instance, + configManager.Object, + hostEnvironment.Object); + + await middleware.Invoke(context).ConfigureAwait(false); + + return context.Response.StatusCode; + } + + private async Task InitiateAsync() + => (await _manager.TryConnect(AuthorizationInfo()).ConfigureAwait(false)).Secret; +} diff --git a/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectStoreWiringTests.cs b/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectStoreWiringTests.cs new file mode 100644 index 0000000000..b492f44d98 --- /dev/null +++ b/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectStoreWiringTests.cs @@ -0,0 +1,147 @@ +using System; +using System.IO; +using System.Threading.Tasks; +using Emby.Server.Implementations.QuickConnect; +using Jellyfin.Server.Extensions; +using Jellyfin.Server.Tests.HighAvailability; +using MediaBrowser.Common.Configuration; +using MediaBrowser.Controller.Authentication; +using MediaBrowser.Controller.QuickConnect; +using MediaBrowser.Model.QuickConnect; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using StackExchange.Redis; +using Xunit; + +namespace Jellyfin.Server.Tests.QuickConnect; + +/// +/// Drives the whole configuration path a deployment uses: a bare +/// Jellyfin__TranscodeStore__RedisConnectionString environment variable, the server's own +/// configuration builder, the store registration, and a quick connect flow against a real valkey. +/// +[Trait("Category", "RequiresDocker")] +[Collection("JellyfinSectionConfiguration")] +public sealed class QuickConnectStoreWiringTests : IAsyncLifetime +{ + private const string RedisConnectionStringVariable = "Jellyfin__TranscodeStore__RedisConnectionString"; + + private RedisTestServer _redis = null!; + private string _configDirectory = string.Empty; + + /// + public async ValueTask InitializeAsync() + { + _redis = await RedisTestServer.StartAsync().ConfigureAwait(false); + _configDirectory = Directory.CreateTempSubdirectory("jellyfin-quickconnect-wiring").FullName; + await File.WriteAllTextAsync(Path.Combine(_configDirectory, "logging.default.json"), "{}").ConfigureAwait(false); + } + + /// + public async ValueTask DisposeAsync() + { + Environment.SetEnvironmentVariable(RedisConnectionStringVariable, null); + + if (_configDirectory.Length > 0) + { + Directory.Delete(_configDirectory, true); + } + + await _redis.DisposeAsync().ConfigureAwait(false); + } + + /// + /// The variable form deployments set selects the shared store, and that store really talks to valkey. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task ManifestStyleEnvironmentVariable_SelectsTheSharedStore() + { + Environment.SetEnvironmentVariable(RedisConnectionStringVariable, _redis.ConnectionString); + + await using var provider = BuildProvider(); + + var store = provider.GetRequiredService(); + Assert.IsType(store); + + var request = NewRequest(); + await store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), TestContext.Current.CancellationToken); + + Assert.True(await store.TryClaimAuthorizationAsync(request.Secret, DateTime.UtcNow.AddMinutes(10), TestContext.Current.CancellationToken)); + + var redis = provider.GetRequiredService(); + Assert.True(await redis.GetDatabase().KeyExistsAsync("jellyfin:quickconnect:request:" + request.Secret)); + } + + /// + /// Without the variable the deployment is single-instance and gets the process-local store, which + /// runs a whole flow on its own. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task NoEnvironmentVariable_SelectsTheProcessLocalStore() + { + Environment.SetEnvironmentVariable(RedisConnectionStringVariable, null); + + await using var provider = BuildProvider(); + + var store = provider.GetRequiredService(); + Assert.IsType(store); + + var cancellationToken = TestContext.Current.CancellationToken; + var request = NewRequest(); + await store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), cancellationToken); + + Assert.Equal(request.Secret, (await store.GetRequestByCodeAsync(request.Code, cancellationToken))?.Secret); + Assert.True(await store.TryClaimAuthorizationAsync(request.Secret, DateTime.UtcNow.AddMinutes(10), cancellationToken)); + + await store.SetAuthorizationAsync( + request.Secret, + new AuthenticationResult { AccessToken = "token-1" }, + DateTime.UtcNow.AddMinutes(10), + cancellationToken); + + Assert.Equal("token-1", (await store.GetAuthorizationAsync(request.Secret, cancellationToken))?.AccessToken); + Assert.Equal("token-1", (await store.GetAuthorizationAsync(request.Secret, cancellationToken))?.AccessToken); + } + + /// + /// A connection string that is set but unreachable is a misconfigured multi-instance deployment. It + /// fails rather than handing out a store the other instances cannot see, which would put quick + /// connect back on the cross-instance behaviour this configuration exists to fix. + /// + [Fact] + public void UnreachableRedisAtStartup_FailsClosed() + { + Environment.SetEnvironmentVariable(RedisConnectionStringVariable, "127.0.0.1:1,connectTimeout=250,connectRetry=0"); + + using var provider = BuildProvider(); + + Assert.ThrowsAny(() => provider.GetRequiredService()); + } + + private static QuickConnectResult NewRequest() => new QuickConnectResult( + Guid.NewGuid().ToString("N"), + Guid.NewGuid().ToString("N").Substring(0, 6), + DateTime.UtcNow, + "device-1", + "Living Room TV", + "Jellyfin Web", + "1.0.0"); + + private ServiceProvider BuildProvider() + { + var appPaths = new Mock(); + appPaths.Setup(paths => paths.ConfigurationDirectoryPath).Returns(_configDirectory); + IConfiguration configuration = Jellyfin.Server.Program.CreateAppConfiguration(new StartupOptions(), appPaths.Object); + + var services = new ServiceCollection(); + services.AddLogging(); + services.AddTranscodeSessionStore(configuration, NullLogger.Instance); + services.AddQuickConnectStore(configuration, NullLogger.Instance); + + return services.BuildServiceProvider(); + } +} diff --git a/tests/Jellyfin.Server.Tests/QuickConnect/RedisQuickConnectStoreDegradedTests.cs b/tests/Jellyfin.Server.Tests/QuickConnect/RedisQuickConnectStoreDegradedTests.cs new file mode 100644 index 0000000000..bfd4999705 --- /dev/null +++ b/tests/Jellyfin.Server.Tests/QuickConnect/RedisQuickConnectStoreDegradedTests.cs @@ -0,0 +1,302 @@ +using System; +using System.Collections.Generic; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using Emby.Server.Implementations.QuickConnect; +using Jellyfin.Server.Tests.HighAvailability; +using MediaBrowser.Common.Extensions; +using MediaBrowser.Controller.Authentication; +using MediaBrowser.Controller.QuickConnect; +using MediaBrowser.Model.QuickConnect; +using Microsoft.Extensions.Logging.Abstractions; +using StackExchange.Redis; +using Xunit; + +namespace Jellyfin.Server.Tests.QuickConnect; + +/// +/// What a does while its Redis is unreachable. Each instance talks +/// to the one real server through a proxy of its own, so an outage can be given to one instance and not +/// the others, and then taken back. +/// +[Trait("Category", "RequiresDocker")] +public sealed class RedisQuickConnectStoreDegradedTests : IAsyncLifetime +{ + private readonly List _proxies = new(); + private readonly List _connections = new(); + + private RedisTestServer _redis = null!; + + private static CancellationToken CancellationToken => TestContext.Current.CancellationToken; + + /// + public async ValueTask InitializeAsync() + { + _redis = await RedisTestServer.StartAsync().ConfigureAwait(false); + } + + /// + public async ValueTask DisposeAsync() + { + foreach (var connection in _connections) + { + await connection.DisposeAsync().ConfigureAwait(false); + } + + foreach (var proxy in _proxies) + { + await proxy.DisposeAsync().ConfigureAwait(false); + } + + await _redis.DisposeAsync().ConfigureAwait(false); + } + + /// + /// Every leg of the flow fails closed while Redis is unreachable. None of them may answer as though + /// Redis had said the request is unknown, because that tells a polling client its secret is invalid. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task EveryLeg_WhileRedisIsUnreachable_ReportsUnavailable() + { + var instance = await CreateInstanceAsync(); + var request = NewRequest(); + var expiresUtc = DateTime.UtcNow.AddMinutes(10); + + instance.Proxy.Cut(); + + await AssertUnavailableAsync(() => instance.Store.SetRequestAsync(request, expiresUtc, CancellationToken)); + await AssertUnavailableAsync(() => instance.Store.GetRequestBySecretAsync(request.Secret, CancellationToken)); + await AssertUnavailableAsync(() => instance.Store.GetRequestByCodeAsync(request.Code, CancellationToken)); + await AssertUnavailableAsync(() => instance.Store.TryClaimAuthorizationAsync(request.Secret, expiresUtc, CancellationToken)); + await AssertUnavailableAsync(() => instance.Store.SetAuthorizationAsync( + request.Secret, + new AuthenticationResult { AccessToken = "token-1" }, + expiresUtc, + CancellationToken)); + await AssertUnavailableAsync(() => instance.Store.GetAuthorizationAsync(request.Secret, CancellationToken)); + } + + /// + /// The same three reads against a Redis that is answering report a genuine miss as a miss, which is + /// what makes an outage and an unknown secret tellable apart by the callers above. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task EveryRead_AgainstAHealthyRedis_ReportsAMissAsAMiss() + { + var instance = await CreateInstanceAsync(); + var request = NewRequest(); + + Assert.Null(await instance.Store.GetRequestBySecretAsync(request.Secret, CancellationToken)); + Assert.Null(await instance.Store.GetRequestByCodeAsync(request.Code, CancellationToken)); + Assert.Null(await instance.Store.GetAuthorizationAsync(request.Secret, CancellationToken)); + } + + /// + /// A request is resolvable by its secret and by its code together or not at all: a failure part way + /// through storing it must not leave a code on the user's screen that resolves to nothing for the + /// whole ten minutes the poll keeps succeeding. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task PendingRequest_WhenRedisGoesAwayMidWrite_IsResolvableByBothKeysOrNeither() + { + var instance = await CreateInstanceAsync(); + var request = NewRequest(); + + // Redis is taken away the instant after it has applied the write of the secret key, which is + // where a two round trip write loses the code key. + instance.Proxy.CutAfterForwarding("request:" + request.Secret); + await Record.ExceptionAsync(() => instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken)); + + // Read straight from the server: the instance's own connection is the one that was cut. + var direct = await _redis.ConnectAsync(); + _connections.Add(direct); + var bySecret = await direct.GetDatabase().KeyExistsAsync("jellyfin:quickconnect:request:" + request.Secret); + var byCode = await direct.GetDatabase().KeyExistsAsync("jellyfin:quickconnect:code:" + request.Code); + + Assert.Equal(bySecret, byCode); + } + + /// + /// A malformed stored value is a fault of its own, not Redis being unavailable, so it is not reported + /// as either a miss or an outage. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task PendingRequest_ThatIsMalformedInRedis_SurfacesAsItsOwnFault() + { + var instance = await CreateInstanceAsync(); + var request = NewRequest(); + + await instance.Connection.GetDatabase().StringSetAsync( + "jellyfin:quickconnect:request:" + request.Secret, + "{ not json", + TimeSpan.FromMinutes(10)); + + await Assert.ThrowsAsync(() => instance.Store.GetRequestBySecretAsync(request.Secret, CancellationToken)); + } + + /// + /// A store whose Redis comes back answers from Redis again, with nothing carried over from the outage. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Store_AfterAnOutage_WorksAgain() + { + var instance = await CreateInstanceAsync(); + var request = NewRequest(); + + instance.Proxy.Cut(); + await AssertUnavailableAsync(() => instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken)); + + await RestoreAsync(instance); + + Assert.Null(await instance.Store.GetRequestBySecretAsync(request.Secret, CancellationToken)); + + await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken); + Assert.Equal(request.Secret, (await instance.Store.GetRequestByCodeAsync(request.Code, CancellationToken))?.Secret); + } + + /// + /// Two instances racing to authorize one request: exactly one of them may go on to mint a token. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Claim_RacedOnTwoInstances_SucceedsOnce() + { + var first = await CreateInstanceAsync(); + var second = await CreateInstanceAsync(); + + for (var attempt = 0; attempt < 25; attempt++) + { + var request = NewRequest(); + var expiresUtc = DateTime.UtcNow.AddMinutes(10); + await first.Store.SetRequestAsync(request, expiresUtc, CancellationToken); + + var claims = await Task.WhenAll( + Task.Run(() => first.Store.TryClaimAuthorizationAsync(request.Secret, expiresUtc, CancellationToken), CancellationToken), + Task.Run(() => second.Store.TryClaimAuthorizationAsync(request.Secret, expiresUtc, CancellationToken), CancellationToken)); + + Assert.Single(claims, claimed => claimed); + } + } + + /// + /// A request that is unknown, already claimed or already authorized cannot be claimed. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Claim_IsRefused_ForUnknownClaimedAndAuthorizedRequests() + { + var instance = await CreateInstanceAsync(); + var expiresUtc = DateTime.UtcNow.AddMinutes(10); + + Assert.False(await instance.Store.TryClaimAuthorizationAsync("unknown-secret", expiresUtc, CancellationToken)); + + var request = NewRequest(); + await instance.Store.SetRequestAsync(request, expiresUtc, CancellationToken); + Assert.True(await instance.Store.TryClaimAuthorizationAsync(request.Secret, expiresUtc, CancellationToken)); + Assert.False(await instance.Store.TryClaimAuthorizationAsync(request.Secret, expiresUtc, CancellationToken)); + + var authorized = NewRequest(); + authorized.Authenticated = true; + await instance.Store.SetRequestAsync(authorized, expiresUtc, CancellationToken); + Assert.False(await instance.Store.TryClaimAuthorizationAsync(authorized.Secret, expiresUtc, CancellationToken)); + } + + /// + /// An authorization is read, not spent: the same secret exchanged again on the same instance returns + /// the same access token for as long as the authorization lives. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Authorization_IsReadRepeatedly_WithoutBeingSpent() + { + var instance = await CreateInstanceAsync(); + var request = NewRequest(); + await instance.Store.SetAuthorizationAsync( + request.Secret, + new AuthenticationResult { AccessToken = "token-1" }, + DateTime.UtcNow.AddMinutes(10), + CancellationToken); + + Assert.Equal("token-1", (await instance.Store.GetAuthorizationAsync(request.Secret, CancellationToken))?.AccessToken); + Assert.Equal("token-1", (await instance.Store.GetAuthorizationAsync(request.Secret, CancellationToken))?.AccessToken); + } + + /// + /// A write whose expiry has already passed is ignored, by the shared store and the process-local one + /// alike: a deployment must not get a different answer out of the two. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task ElapsedExpiry_IsIgnoredByBothStores() + { + var shared = (await CreateInstanceAsync()).Store; + var local = new InMemoryQuickConnectStore(); + + foreach (var store in new IQuickConnectStore[] { shared, local }) + { + var request = NewRequest(); + var elapsed = DateTime.UtcNow.AddSeconds(-1); + + await store.SetRequestAsync(request, elapsed, CancellationToken); + await store.SetAuthorizationAsync(request.Secret, new AuthenticationResult { AccessToken = "token-1" }, elapsed, CancellationToken); + + Assert.Null(await store.GetRequestBySecretAsync(request.Secret, CancellationToken)); + Assert.Null(await store.GetRequestByCodeAsync(request.Code, CancellationToken)); + Assert.Null(await store.GetAuthorizationAsync(request.Secret, CancellationToken)); + } + } + + private static QuickConnectResult NewRequest() => new QuickConnectResult( + Guid.NewGuid().ToString("N"), + Guid.NewGuid().ToString("N").Substring(0, 6), + DateTime.UtcNow, + "device-1", + "Living Room TV", + "Jellyfin Web", + "1.0.0"); + + private static async Task AssertUnavailableAsync(Func operation) + { + var exception = await Record.ExceptionAsync(operation); + + Assert.NotNull(exception); + Assert.IsType(exception); + } + + private static async Task RestoreAsync(Instance instance) + { + instance.Proxy.Restore(); + + for (var attempt = 1; ; attempt++) + { + try + { + await instance.Connection.GetDatabase().PingAsync(); + return; + } + catch (Exception exception) when (exception is RedisException or TimeoutException && attempt < 60) + { + await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken); + } + } + } + + private async Task CreateInstanceAsync() + { + var proxy = RedisFaultProxy.Start(_redis.ConnectionString); + _proxies.Add(proxy); + + var connection = await ConnectionMultiplexer.ConnectAsync(proxy.ConnectionString).ConfigureAwait(false); + _connections.Add(connection); + + return new Instance(proxy, connection, new RedisQuickConnectStore(connection, NullLogger.Instance)); + } + + private sealed record Instance(RedisFaultProxy Proxy, IConnectionMultiplexer Connection, RedisQuickConnectStore Store); +}