fix(ha): share quick connect state between instances #34
+5
-1
@@ -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,21 +132,27 @@ 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");
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
/// <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.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);
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
private async Task<ConflictException> 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);
|
||||
}
|
||||
|
||||
/// <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,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;
|
||||
|
||||
/// <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
|
||||
/// claimed with a Lua check-and-set, so only one instance can ever mint a given secret's access token.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// 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.
|
||||
/// </remarks>
|
||||
public sealed class RedisQuickConnectStore : IQuickConnectStore
|
||||
{
|
||||
private const string KeyPrefix = "jellyfin:quickconnect:";
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
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";
|
||||
|
||||
/// <summary>
|
||||
/// 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 <c>SET NX</c>, so of two
|
||||
/// instances racing on one code exactly one goes on to mint an access token.
|
||||
/// </summary>
|
||||
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<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();
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<QuickConnectResult?> 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<QuickConnectResult>(raw.ToString(), JsonDefaults.Options) : null;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<QuickConnectResult?> 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;
|
||||
}
|
||||
|
||||
/// <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;
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<bool> 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;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
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);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<AuthenticationResult?> GetAuthorizationAsync(string secret, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var raw = await CallAsync(() => _db.StringGetAsync(AuthorizationKey(secret))).ConfigureAwait(false);
|
||||
|
||||
return raw.HasValue
|
||||
? JsonSerializer.Deserialize<AuthenticationResult>(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<T> CallAsync<T>(Func<Task<T>> 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -50,16 +50,18 @@ public class QuickConnectController : BaseJellyfinApiController
|
||||
/// </summary>
|
||||
/// <response code="200">Quick connect request successfully created.</response>
|
||||
/// <response code="401">Quick connect is not active on this server.</response>
|
||||
/// <response code="503">Quick connect state is unavailable.</response>
|
||||
/// <returns>A <see cref="QuickConnectResult"/> with a secret and code for future use or an error message.</returns>
|
||||
[HttpPost("Initiate")]
|
||||
[ProducesResponseType(StatusCodes.Status200OK)]
|
||||
[ProducesResponseType(StatusCodes.Status401Unauthorized)]
|
||||
[ProducesResponseType(StatusCodes.Status503ServiceUnavailable)]
|
||||
public async Task<ActionResult<QuickConnectResult>> 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
|
||||
/// <param name="secret">Secret previously returned from the Initiate endpoint.</param>
|
||||
/// <response code="200">Quick connect result returned.</response>
|
||||
/// <response code="404">Unknown quick connect secret.</response>
|
||||
/// <response code="503">Quick connect state is unavailable.</response>
|
||||
/// <returns>An updated <see cref="QuickConnectResult"/>.</returns>
|
||||
[HttpGet("Connect")]
|
||||
[ProducesResponseType(StatusCodes.Status200OK)]
|
||||
[ProducesResponseType(StatusCodes.Status404NotFound)]
|
||||
public ActionResult<QuickConnectResult> GetQuickConnectState([FromQuery, Required] string secret)
|
||||
[ProducesResponseType(StatusCodes.Status503ServiceUnavailable)]
|
||||
public async Task<ActionResult<QuickConnectResult>> 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
|
||||
/// <param name="userId">The user the authorize. Access to the requested user is required.</param>
|
||||
/// <response code="200">Quick connect result authorized successfully.</response>
|
||||
/// <response code="403">Unknown user id.</response>
|
||||
/// <response code="409">Request is already authorized, or authorizing it did not complete.</response>
|
||||
/// <response code="503">Quick connect state is unavailable.</response>
|
||||
/// <returns>Boolean indicating if the authorization was successful.</returns>
|
||||
[HttpPost("Authorize")]
|
||||
[Authorize]
|
||||
[ProducesResponseType(StatusCodes.Status200OK)]
|
||||
[ProducesResponseType(StatusCodes.Status403Forbidden)]
|
||||
[ProducesResponseType(StatusCodes.Status409Conflict)]
|
||||
[ProducesResponseType(StatusCodes.Status503ServiceUnavailable)]
|
||||
public async Task<ActionResult<bool>> AuthorizeQuickConnect([FromQuery, Required] string code, [FromQuery] Guid? userId = null)
|
||||
{
|
||||
userId = RequestHelpers.GetUserId(User, userId);
|
||||
|
||||
@@ -241,15 +241,19 @@ public class UserController : BaseJellyfinApiController
|
||||
/// <param name="request">The <see cref="QuickConnectDto"/> request.</param>
|
||||
/// <response code="200">User authenticated.</response>
|
||||
/// <response code="400">Missing token.</response>
|
||||
/// <response code="404">Unknown or unauthorized quick connect secret.</response>
|
||||
/// <response code="503">Quick connect state is unavailable.</response>
|
||||
/// <returns>A <see cref="Task"/> containing an <see cref="AuthenticationRequest"/> with information about the new session.</returns>
|
||||
[HttpPost("AuthenticateWithQuickConnect")]
|
||||
[ProducesResponseType(StatusCodes.Status200OK)]
|
||||
[ProducesResponseType(StatusCodes.Status404NotFound)]
|
||||
[ProducesResponseType(StatusCodes.Status503ServiceUnavailable)]
|
||||
[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)
|
||||
{
|
||||
|
||||
@@ -131,6 +131,8 @@ public class ExceptionMiddleware
|
||||
FileNotFoundException => StatusCodes.Status404NotFound,
|
||||
ResourceNotFoundException => StatusCodes.Status404NotFound,
|
||||
MethodNotAllowedException => StatusCodes.Status405MethodNotAllowed,
|
||||
ConflictException => StatusCodes.Status409Conflict,
|
||||
ServiceUnavailableException => StatusCodes.Status503ServiceUnavailable,
|
||||
_ => StatusCodes.Status500InternalServerError
|
||||
};
|
||||
}
|
||||
|
||||
@@ -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,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;
|
||||
|
||||
/// <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. 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.
|
||||
/// </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 => new RedisQuickConnectStore(
|
||||
sp.GetRequiredService<IConnectionMultiplexer>(),
|
||||
sp.GetRequiredService<ILogger<RedisQuickConnectStore>>()));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
using System;
|
||||
|
||||
namespace MediaBrowser.Common.Extensions
|
||||
{
|
||||
/// <summary>
|
||||
/// Thrown when the current state of a resource does not allow the requested operation.
|
||||
/// </summary>
|
||||
public class ConflictException : Exception
|
||||
{
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="ConflictException" /> class.
|
||||
/// </summary>
|
||||
public ConflictException()
|
||||
{
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="ConflictException" /> class.
|
||||
/// </summary>
|
||||
/// <param name="message">The message.</param>
|
||||
public ConflictException(string message)
|
||||
: base(message)
|
||||
{
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="ConflictException" /> class.
|
||||
/// </summary>
|
||||
/// <param name="message">The message.</param>
|
||||
/// <param name="innerException">The exception that caused this one.</param>
|
||||
public ConflictException(string message, Exception innerException)
|
||||
: base(message, innerException)
|
||||
{
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
using System;
|
||||
|
||||
namespace MediaBrowser.Common.Extensions
|
||||
{
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
public class ServiceUnavailableException : Exception
|
||||
{
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="ServiceUnavailableException" /> class.
|
||||
/// </summary>
|
||||
public ServiceUnavailableException()
|
||||
{
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="ServiceUnavailableException" /> class.
|
||||
/// </summary>
|
||||
/// <param name="message">The message.</param>
|
||||
public ServiceUnavailableException(string message)
|
||||
: base(message)
|
||||
{
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="ServiceUnavailableException" /> class.
|
||||
/// </summary>
|
||||
/// <param name="message">The message.</param>
|
||||
/// <param name="innerException">The exception that caused this one.</param>
|
||||
public ServiceUnavailableException(string message, Exception innerException)
|
||||
: base(message, innerException)
|
||||
{
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
/// </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.
|
||||
/// 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
|
||||
/// <see cref="ConflictException"/> and the user has to start quick connect again for a new code.
|
||||
/// </summary>
|
||||
/// <param name="userId">User id.</param>
|
||||
/// <param name="code">Identifying code for the request.</param>
|
||||
@@ -39,10 +42,11 @@ namespace MediaBrowser.Controller.QuickConnect
|
||||
Task<bool> AuthorizeRequest(Guid userId, string code);
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <param name="secret">The secret.</param>
|
||||
/// <returns>The authentication result.</returns>
|
||||
AuthenticationResult GetAuthorizedRequest(string secret);
|
||||
Task<AuthenticationResult> GetAuthorizedRequest(string secret);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
/// <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>
|
||||
/// <remarks>
|
||||
/// A shared implementation that cannot reach its backend throws <see cref="ServiceUnavailableException"/>
|
||||
/// rather than reporting a miss, because a miss tells a polling client its secret is invalid.
|
||||
/// </remarks>
|
||||
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"/>, resolvable by both its secret
|
||||
/// and its code or by neither. A request already past <paramref name="expiresUtc"/> is not stored.
|
||||
/// </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>
|
||||
/// Atomically claims the sole right to authorize the request behind <paramref name="secret"/>, 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.
|
||||
/// </summary>
|
||||
/// <param name="secret">The request secret.</param>
|
||||
/// <param name="expiresUtc">The instant the claim lapses, after which the request can be authorized again.</param>
|
||||
/// <param name="cancellationToken">A cancellation token.</param>
|
||||
/// <returns><c>true</c> when this caller may go on to authorize the request; <c>false</c> when it is unknown, expired, already authorized or claimed elsewhere.</returns>
|
||||
Task<bool> TryClaimAuthorizationAsync(string secret, 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>
|
||||
/// Reads the authentication for <paramref name="secret"/>. 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.
|
||||
/// </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 or has expired.</returns>
|
||||
Task<AuthenticationResult?> GetAuthorizationAsync(string secret, CancellationToken cancellationToken = default);
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// A process-local <see cref="IQuickConnectStore"/>, 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.
|
||||
/// </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);
|
||||
private readonly ConcurrentDictionary<string, DateTime> _authorizationClaims = 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();
|
||||
if (expiresUtc > DateTime.UtcNow)
|
||||
{
|
||||
_requests[request.Secret] = new Entry<QuickConnectResult>(expiresUtc, request);
|
||||
}
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<bool> 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));
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task SetAuthorizationAsync(string secret, AuthenticationResult authenticationResult, DateTime expiresUtc, CancellationToken cancellationToken = default)
|
||||
{
|
||||
Expire();
|
||||
if (expiresUtc > DateTime.UtcNow)
|
||||
{
|
||||
_authorizations[secret] = new Entry<AuthenticationResult>(expiresUtc, authenticationResult);
|
||||
}
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<AuthenticationResult?> 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<T>(DateTime ExpiresUtc, T Value);
|
||||
}
|
||||
+69
-17
@@ -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,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<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 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<ConflictException>(() => _quickConnectManager.AuthorizeRequest(Guid.Empty, res.Code));
|
||||
}
|
||||
|
||||
private async Task<bool> AuthorizeAsync(string code)
|
||||
{
|
||||
try
|
||||
{
|
||||
return await _quickConnectManager.AuthorizeRequest(Guid.Empty, code).ConfigureAwait(false);
|
||||
}
|
||||
catch (ConflictException)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
/// </summary>
|
||||
[Collection("JellyfinSectionConfiguration")]
|
||||
public sealed class JellyfinSectionConfigurationTests : IDisposable
|
||||
{
|
||||
private const string RedisKey = "Jellyfin:TranscodeStore:RedisConnectionString";
|
||||
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
public sealed class RedisFaultProxy : IAsyncDisposable
|
||||
{
|
||||
private readonly ConcurrentDictionary<TcpClient, byte> _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;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
public string ConnectionString => string.Create(
|
||||
CultureInfo.InvariantCulture,
|
||||
$"127.0.0.1:{_port},abortConnect=false,connectTimeout=500,syncTimeout=2000,connectRetry=1");
|
||||
|
||||
/// <summary>
|
||||
/// Starts a proxy in front of the server named by <paramref name="target"/>.
|
||||
/// </summary>
|
||||
/// <param name="target">The connection string of the server to forward to.</param>
|
||||
/// <returns>The running proxy.</returns>
|
||||
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;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Takes Redis away from everything connected through the proxy.
|
||||
/// </summary>
|
||||
public void Cut()
|
||||
{
|
||||
_cut = true;
|
||||
DropLiveConnections();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Arms a cut for the moment after a command containing <paramref name="marker"/> 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.
|
||||
/// </summary>
|
||||
/// <param name="marker">Text that identifies the command to cut after.</param>
|
||||
public void CutAfterForwarding(string marker) => _cutAfterMarker = Encoding.UTF8.GetBytes(marker);
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
public void Restore()
|
||||
{
|
||||
_cutAfterMarker = null;
|
||||
_cut = false;
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
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)
|
||||
{
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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,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;
|
||||
|
||||
/// <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>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[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);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[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);
|
||||
}
|
||||
}
|
||||
|
||||
/// <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 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<bool> AuthorizeAsync(IQuickConnect manager, Guid userId, string code)
|
||||
{
|
||||
try
|
||||
{
|
||||
return await manager.AuthorizeRequest(userId, code).ConfigureAwait(false);
|
||||
}
|
||||
catch (ConflictException)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[Trait("Category", "RequiresDocker")]
|
||||
public sealed class QuickConnectStatusCodeTests : IAsyncLifetime
|
||||
{
|
||||
private readonly Mock<ISessionManager> _sessionManager = new();
|
||||
|
||||
private RedisTestServer _redis = null!;
|
||||
private RedisFaultProxy _proxy = null!;
|
||||
private IConnectionMultiplexer _connection = null!;
|
||||
private QuickConnectManager _manager = null!;
|
||||
private QuickConnectController _controller = null!;
|
||||
|
||||
/// <inheritdoc/>
|
||||
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<IServerConfigurationManager>();
|
||||
configManager.Setup(manager => manager.Configuration).Returns(new ServerConfiguration { QuickConnectAvailable = true });
|
||||
|
||||
_manager = new QuickConnectManager(
|
||||
configManager.Object,
|
||||
NullLogger<QuickConnectManager>.Instance,
|
||||
_sessionManager.Object,
|
||||
new RedisQuickConnectStore(_connection, NullLogger<RedisQuickConnectStore>.Instance));
|
||||
|
||||
_controller = new QuickConnectController(_manager, Mock.Of<IAuthorizationContext>());
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
await _connection.DisposeAsync().ConfigureAwait(false);
|
||||
await _proxy.DisposeAsync().ConfigureAwait(false);
|
||||
await _redis.DisposeAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A secret valkey has never heard of is a 404, which is what ends a flow the user abandoned.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task Poll_UnknownSecret_IsNotFound()
|
||||
{
|
||||
Assert.Equal(
|
||||
StatusCodes.Status404NotFound,
|
||||
await StatusCodeAsync(async () => StatusOf(await _controller.GetQuickConnectState(NewSecret()))));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[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))));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The exchange leg tells the two apart the same way.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task Exchange_UnknownSecret_IsNotFound()
|
||||
{
|
||||
Assert.Equal(
|
||||
StatusCodes.Status404NotFound,
|
||||
await StatusCodeAsync(async () =>
|
||||
{
|
||||
await _manager.GetAuthorizedRequest(NewSecret()).ConfigureAwait(false);
|
||||
return StatusCodes.Status200OK;
|
||||
}));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The exchange leg while valkey is unreachable.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[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;
|
||||
}));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The authorize leg while valkey is unreachable.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[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;
|
||||
}));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task Authorize_AfterAMintThatFailed_IsConflictAndSaysToStartAgain()
|
||||
{
|
||||
var initiated = await _manager.TryConnect(AuthorizationInfo());
|
||||
|
||||
_sessionManager
|
||||
.Setup(manager => manager.AuthenticateDirect(It.IsAny<AuthenticationRequest>()))
|
||||
.ThrowsAsync(new InvalidOperationException("mint failed"));
|
||||
|
||||
await Assert.ThrowsAsync<InvalidOperationException>(() => _manager.AuthorizeRequest(Guid.NewGuid(), initiated.Code));
|
||||
|
||||
var retry = await Record.ExceptionAsync(() => _manager.AuthorizeRequest(Guid.NewGuid(), initiated.Code));
|
||||
|
||||
var conflict = Assert.IsType<ConflictException>(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;
|
||||
}));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A request that really was authorized still says so, so the accurate message above is not just a
|
||||
/// blanket replacement for the old one.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task Authorize_OfAnAuthorizedRequest_IsConflictAndSaysAlreadyAuthorized()
|
||||
{
|
||||
var initiated = await _manager.TryConnect(AuthorizationInfo());
|
||||
var userId = Guid.NewGuid();
|
||||
|
||||
_sessionManager
|
||||
.Setup(manager => manager.AuthenticateDirect(It.IsAny<AuthenticationRequest>()))
|
||||
.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<ConflictException>(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<QuickConnectResult> result)
|
||||
=> result.Result is IStatusCodeActionResult status
|
||||
? status.StatusCode ?? StatusCodes.Status200OK
|
||||
: StatusCodes.Status200OK;
|
||||
|
||||
private static async Task<int> StatusCodeAsync(Func<Task<int>> action)
|
||||
{
|
||||
var appPaths = new Mock<IServerApplicationPaths>();
|
||||
appPaths.Setup(paths => paths.ProgramSystemPath).Returns("/program");
|
||||
appPaths.Setup(paths => paths.ProgramDataPath).Returns("/data");
|
||||
|
||||
var configManager = new Mock<IServerConfigurationManager>();
|
||||
configManager.Setup(manager => manager.ApplicationPaths).Returns(appPaths.Object);
|
||||
|
||||
var hostEnvironment = new Mock<IWebHostEnvironment>();
|
||||
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<ExceptionMiddleware>.Instance,
|
||||
configManager.Object,
|
||||
hostEnvironment.Object);
|
||||
|
||||
await middleware.Invoke(context).ConfigureAwait(false);
|
||||
|
||||
return context.Response.StatusCode;
|
||||
}
|
||||
|
||||
private async Task<string> InitiateAsync()
|
||||
=> (await _manager.TryConnect(AuthorizationInfo()).ConfigureAwait(false)).Secret;
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Drives the whole configuration path a deployment uses: a bare
|
||||
/// <c>Jellyfin__TranscodeStore__RedisConnectionString</c> environment variable, the server's own
|
||||
/// configuration builder, the store registration, and a quick connect flow against a real valkey.
|
||||
/// </summary>
|
||||
[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;
|
||||
|
||||
/// <inheritdoc/>
|
||||
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);
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
Environment.SetEnvironmentVariable(RedisConnectionStringVariable, null);
|
||||
|
||||
if (_configDirectory.Length > 0)
|
||||
{
|
||||
Directory.Delete(_configDirectory, true);
|
||||
}
|
||||
|
||||
await _redis.DisposeAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The variable form deployments set selects the shared store, and that store really talks to valkey.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task ManifestStyleEnvironmentVariable_SelectsTheSharedStore()
|
||||
{
|
||||
Environment.SetEnvironmentVariable(RedisConnectionStringVariable, _redis.ConnectionString);
|
||||
|
||||
await using var provider = BuildProvider();
|
||||
|
||||
var store = provider.GetRequiredService<IQuickConnectStore>();
|
||||
Assert.IsType<RedisQuickConnectStore>(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<IConnectionMultiplexer>();
|
||||
Assert.True(await redis.GetDatabase().KeyExistsAsync("jellyfin:quickconnect:request:" + request.Secret));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Without the variable the deployment is single-instance and gets the process-local store, which
|
||||
/// runs a whole flow on its own.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task NoEnvironmentVariable_SelectsTheProcessLocalStore()
|
||||
{
|
||||
Environment.SetEnvironmentVariable(RedisConnectionStringVariable, null);
|
||||
|
||||
await using var provider = BuildProvider();
|
||||
|
||||
var store = provider.GetRequiredService<IQuickConnectStore>();
|
||||
Assert.IsType<InMemoryQuickConnectStore>(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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void UnreachableRedisAtStartup_FailsClosed()
|
||||
{
|
||||
Environment.SetEnvironmentVariable(RedisConnectionStringVariable, "127.0.0.1:1,connectTimeout=250,connectRetry=0");
|
||||
|
||||
using var provider = BuildProvider();
|
||||
|
||||
Assert.ThrowsAny<RedisConnectionException>(() => provider.GetRequiredService<IQuickConnectStore>());
|
||||
}
|
||||
|
||||
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<IApplicationPaths>();
|
||||
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();
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// What a <see cref="RedisQuickConnectStore"/> 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.
|
||||
/// </summary>
|
||||
[Trait("Category", "RequiresDocker")]
|
||||
public sealed class RedisQuickConnectStoreDegradedTests : IAsyncLifetime
|
||||
{
|
||||
private readonly List<RedisFaultProxy> _proxies = new();
|
||||
private readonly List<IConnectionMultiplexer> _connections = new();
|
||||
|
||||
private RedisTestServer _redis = null!;
|
||||
|
||||
private static CancellationToken CancellationToken => TestContext.Current.CancellationToken;
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async ValueTask InitializeAsync()
|
||||
{
|
||||
_redis = await RedisTestServer.StartAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[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));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[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));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[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<JsonException>(() => instance.Store.GetRequestBySecretAsync(request.Secret, CancellationToken));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A store whose Redis comes back answers from Redis again, with nothing carried over from the outage.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Two instances racing to authorize one request: exactly one of them may go on to mint a token.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[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);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A request that is unknown, already claimed or already authorized cannot be claimed.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[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));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[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<Task> operation)
|
||||
{
|
||||
var exception = await Record.ExceptionAsync(operation);
|
||||
|
||||
Assert.NotNull(exception);
|
||||
Assert.IsType<ServiceUnavailableException>(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<Instance> 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<RedisQuickConnectStore>.Instance));
|
||||
}
|
||||
|
||||
private sealed record Instance(RedisFaultProxy Proxy, IConnectionMultiplexer Connection, RedisQuickConnectStore Store);
|
||||
}
|
||||
Reference in New Issue
Block a user