Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 6c54a02240 | |||
| ba4d487c65 |
+5
-1
@@ -47,6 +47,7 @@ steps:
|
|||||||
# Its data directory lives on the step's ephemeral storage, not on the workspace volume.
|
# 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
|
# 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.
|
# 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
|
- name: postgres-migration-chain
|
||||||
image: mcr.microsoft.com/dotnet/sdk:10.0
|
image: mcr.microsoft.com/dotnet/sdk:10.0
|
||||||
depends_on:
|
depends_on:
|
||||||
@@ -55,13 +56,16 @@ steps:
|
|||||||
DOTNET_CLI_TELEMETRY_OPTOUT: "1"
|
DOTNET_CLI_TELEMETRY_OPTOUT: "1"
|
||||||
DOTNET_NOLOGO: "1"
|
DOTNET_NOLOGO: "1"
|
||||||
JELLYFIN_TEST_POSTGRES: "Host=127.0.0.1;Port=5432;Database=postgres;Username=postgres"
|
JELLYFIN_TEST_POSTGRES: "Host=127.0.0.1;Port=5432;Database=postgres;Username=postgres"
|
||||||
|
JELLYFIN_TEST_REDIS: "127.0.0.1:6379"
|
||||||
commands:
|
commands:
|
||||||
- apt-get -o Acquire::Retries=3 update || apt-get -o Acquire::Retries=3 update
|
- 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
|
- install -d -o postgres -g postgres /tmp/pgdata /tmp/pgrun
|
||||||
- PGBIN=$(ls -d /usr/lib/postgresql/*/bin | tail -1)
|
- 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/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"
|
- 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.Server.Tests/Jellyfin.Server.Tests.csproj -c Release
|
||||||
- dotnet build tests/Jellyfin.Database.Tests.PostgreSQL/Jellyfin.Database.Tests.PostgreSQL.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"
|
- 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;
|
||||||
using System.Collections.Concurrent;
|
|
||||||
using System.Globalization;
|
using System.Globalization;
|
||||||
using System.Linq;
|
|
||||||
using System.Security.Cryptography;
|
using System.Security.Cryptography;
|
||||||
using System.Threading.Tasks;
|
using System.Threading.Tasks;
|
||||||
using MediaBrowser.Common.Extensions;
|
using MediaBrowser.Common.Extensions;
|
||||||
@@ -30,12 +28,10 @@ namespace Emby.Server.Implementations.QuickConnect
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
private const int Timeout = 10;
|
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 IServerConfigurationManager _config;
|
||||||
private readonly ILogger<QuickConnectManager> _logger;
|
private readonly ILogger<QuickConnectManager> _logger;
|
||||||
private readonly ISessionManager _sessionManager;
|
private readonly ISessionManager _sessionManager;
|
||||||
|
private readonly IQuickConnectStore _store;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Initializes a new instance of the <see cref="QuickConnectManager"/> class.
|
/// 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="config">Configuration.</param>
|
||||||
/// <param name="logger">Logger.</param>
|
/// <param name="logger">Logger.</param>
|
||||||
/// <param name="sessionManager">Session Manager.</param>
|
/// <param name="sessionManager">Session Manager.</param>
|
||||||
|
/// <param name="store">Quick connect store.</param>
|
||||||
public QuickConnectManager(
|
public QuickConnectManager(
|
||||||
IServerConfigurationManager config,
|
IServerConfigurationManager config,
|
||||||
ILogger<QuickConnectManager> logger,
|
ILogger<QuickConnectManager> logger,
|
||||||
ISessionManager sessionManager)
|
ISessionManager sessionManager,
|
||||||
|
IQuickConnectStore store)
|
||||||
{
|
{
|
||||||
_config = config;
|
_config = config;
|
||||||
_logger = logger;
|
_logger = logger;
|
||||||
_sessionManager = sessionManager;
|
_sessionManager = sessionManager;
|
||||||
|
_store = store;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
@@ -69,7 +68,7 @@ namespace Emby.Server.Implementations.QuickConnect
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc/>
|
/// <inheritdoc/>
|
||||||
public QuickConnectResult TryConnect(AuthorizationInfo authorizationInfo)
|
public async Task<QuickConnectResult> TryConnect(AuthorizationInfo authorizationInfo)
|
||||||
{
|
{
|
||||||
ArgumentException.ThrowIfNullOrEmpty(authorizationInfo.DeviceId);
|
ArgumentException.ThrowIfNullOrEmpty(authorizationInfo.DeviceId);
|
||||||
ArgumentException.ThrowIfNullOrEmpty(authorizationInfo.Device);
|
ArgumentException.ThrowIfNullOrEmpty(authorizationInfo.Device);
|
||||||
@@ -77,7 +76,6 @@ namespace Emby.Server.Implementations.QuickConnect
|
|||||||
ArgumentException.ThrowIfNullOrEmpty(authorizationInfo.Version);
|
ArgumentException.ThrowIfNullOrEmpty(authorizationInfo.Version);
|
||||||
|
|
||||||
AssertActive();
|
AssertActive();
|
||||||
ExpireRequests();
|
|
||||||
|
|
||||||
var secret = GenerateSecureRandom();
|
var secret = GenerateSecureRandom();
|
||||||
var code = GenerateCode();
|
var code = GenerateCode();
|
||||||
@@ -90,19 +88,17 @@ namespace Emby.Server.Implementations.QuickConnect
|
|||||||
authorizationInfo.Client,
|
authorizationInfo.Client,
|
||||||
authorizationInfo.Version);
|
authorizationInfo.Version);
|
||||||
|
|
||||||
_currentRequests[code] = result;
|
await _store.SetRequestAsync(result, ExpiryOf(result)).ConfigureAwait(false);
|
||||||
return result;
|
return result;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc/>
|
/// <inheritdoc/>
|
||||||
public QuickConnectResult CheckRequestStatus(string secret)
|
public async Task<QuickConnectResult> CheckRequestStatus(string secret)
|
||||||
{
|
{
|
||||||
AssertActive();
|
AssertActive();
|
||||||
ExpireRequests();
|
|
||||||
|
|
||||||
string code = _currentRequests.Where(x => x.Value.Secret == secret).Select(x => x.Value.Code).DefaultIfEmpty(string.Empty).First();
|
var result = await _store.GetRequestBySecretAsync(secret).ConfigureAwait(false);
|
||||||
|
if (result is null)
|
||||||
if (!_currentRequests.TryGetValue(code, out QuickConnectResult? result))
|
|
||||||
{
|
{
|
||||||
throw new ResourceNotFoundException("Unable to find request with provided secret");
|
throw new ResourceNotFoundException("Unable to find request with provided secret");
|
||||||
}
|
}
|
||||||
@@ -136,9 +132,9 @@ namespace Emby.Server.Implementations.QuickConnect
|
|||||||
public async Task<bool> AuthorizeRequest(Guid userId, string code)
|
public async Task<bool> AuthorizeRequest(Guid userId, string code)
|
||||||
{
|
{
|
||||||
AssertActive();
|
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");
|
throw new ResourceNotFoundException("Unable to find request");
|
||||||
}
|
}
|
||||||
@@ -151,6 +147,12 @@ namespace Emby.Server.Implementations.QuickConnect
|
|||||||
// 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.
|
// 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));
|
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 new InvalidOperationException("Request is already authorized");
|
||||||
|
}
|
||||||
|
|
||||||
var authenticationResult = await _sessionManager.AuthenticateDirect(new AuthenticationRequest
|
var authenticationResult = await _sessionManager.AuthenticateDirect(new AuthenticationRequest
|
||||||
{
|
{
|
||||||
UserId = userId,
|
UserId = userId,
|
||||||
@@ -160,9 +162,10 @@ namespace Emby.Server.Implementations.QuickConnect
|
|||||||
AppVersion = result.AppVersion
|
AppVersion = result.AppVersion
|
||||||
}).ConfigureAwait(false);
|
}).ConfigureAwait(false);
|
||||||
|
|
||||||
_authorizedSecrets[result.Secret] = (DateTime.UtcNow, authenticationResult);
|
|
||||||
result.Authenticated = true;
|
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);
|
_logger.LogDebug("Authorizing device with code {Code} to login as user {UserId}", code, userId);
|
||||||
|
|
||||||
@@ -170,19 +173,21 @@ namespace Emby.Server.Implementations.QuickConnect
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc/>
|
/// <inheritdoc/>
|
||||||
public AuthenticationResult GetAuthorizedRequest(string secret)
|
public async Task<AuthenticationResult> GetAuthorizedRequest(string secret)
|
||||||
{
|
{
|
||||||
AssertActive();
|
AssertActive();
|
||||||
ExpireRequests();
|
|
||||||
|
|
||||||
if (!_authorizedSecrets.TryGetValue(secret, out var result))
|
var result = await _store.TryConsumeAuthorizationAsync(secret).ConfigureAwait(false);
|
||||||
|
if (result is null)
|
||||||
{
|
{
|
||||||
throw new ResourceNotFoundException("Unable to find request");
|
throw new ResourceNotFoundException("Unable to find request");
|
||||||
}
|
}
|
||||||
|
|
||||||
return result.AuthenticationResult;
|
return result;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private static DateTime ExpiryOf(QuickConnectResult request) => request.DateAdded.AddMinutes(Timeout);
|
||||||
|
|
||||||
private string GenerateSecureRandom(int length = 32)
|
private string GenerateSecureRandom(int length = 32)
|
||||||
{
|
{
|
||||||
Span<byte> bytes = stackalloc byte[length];
|
Span<byte> bytes = stackalloc byte[length];
|
||||||
@@ -190,42 +195,5 @@ namespace Emby.Server.Implementations.QuickConnect
|
|||||||
|
|
||||||
return Convert.ToHexString(bytes);
|
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,175 @@
|
|||||||
|
using System;
|
||||||
|
using System.Text.Json;
|
||||||
|
using System.Threading;
|
||||||
|
using System.Threading.Tasks;
|
||||||
|
using Jellyfin.Extensions.Json;
|
||||||
|
using MediaBrowser.Controller.Authentication;
|
||||||
|
using MediaBrowser.Controller.QuickConnect;
|
||||||
|
using MediaBrowser.Model.QuickConnect;
|
||||||
|
using Microsoft.Extensions.Logging;
|
||||||
|
using StackExchange.Redis;
|
||||||
|
|
||||||
|
namespace Emby.Server.Implementations.QuickConnect;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// A Redis-backed <see cref="IQuickConnectStore"/> that lets the initiate, authorize and exchange legs
|
||||||
|
/// of a quick connect flow land on different instances. Expiry is the key TTL, an authorization is
|
||||||
|
/// claimed with a Lua check-and-set and consumed with <c>GETDEL</c>, so only one instance can ever mint
|
||||||
|
/// a given secret's access token and only one can ever hand it out.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// A pending request survives an unreachable Redis through a process-local fallback, because a second
|
||||||
|
/// copy of it is harmless. An authorization has none: a second copy of it is a second access token, and
|
||||||
|
/// a write whose response timed out may well have been applied, so a transport failure on that path is
|
||||||
|
/// surfaced rather than degraded.
|
||||||
|
/// </remarks>
|
||||||
|
public sealed class RedisQuickConnectStore : IQuickConnectStore
|
||||||
|
{
|
||||||
|
private const string KeyPrefix = "jellyfin:quickconnect:";
|
||||||
|
|
||||||
|
/// <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 InMemoryQuickConnectStore _fallback;
|
||||||
|
private readonly ILogger<RedisQuickConnectStore> _logger;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Initializes a new instance of the <see cref="RedisQuickConnectStore"/> class.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="redis">The Redis connection multiplexer.</param>
|
||||||
|
/// <param name="logger">The logger.</param>
|
||||||
|
public RedisQuickConnectStore(IConnectionMultiplexer redis, ILogger<RedisQuickConnectStore> logger)
|
||||||
|
{
|
||||||
|
ArgumentNullException.ThrowIfNull(redis);
|
||||||
|
|
||||||
|
_db = redis.GetDatabase();
|
||||||
|
_fallback = new InMemoryQuickConnectStore();
|
||||||
|
_logger = logger;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public async Task<QuickConnectResult?> GetRequestBySecretAsync(string secret, CancellationToken cancellationToken = default)
|
||||||
|
{
|
||||||
|
RedisValue raw;
|
||||||
|
try
|
||||||
|
{
|
||||||
|
raw = await _db.StringGetAsync(RequestKey(secret)).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
catch (Exception ex) when (IsTransportFailure(ex))
|
||||||
|
{
|
||||||
|
LogDegraded(ex);
|
||||||
|
return await _fallback.GetRequestBySecretAsync(secret, cancellationToken).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
|
||||||
|
// A miss is an answer rather than a transport failure, so the fallback is not consulted for it.
|
||||||
|
return raw.HasValue ? JsonSerializer.Deserialize<QuickConnectResult>(raw.ToString(), JsonDefaults.Options) : null;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public async Task<QuickConnectResult?> GetRequestByCodeAsync(string code, CancellationToken cancellationToken = default)
|
||||||
|
{
|
||||||
|
RedisValue secret;
|
||||||
|
try
|
||||||
|
{
|
||||||
|
secret = await _db.StringGetAsync(CodeKey(code)).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
catch (Exception ex) when (IsTransportFailure(ex))
|
||||||
|
{
|
||||||
|
LogDegraded(ex);
|
||||||
|
return await _fallback.GetRequestByCodeAsync(code, cancellationToken).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;
|
||||||
|
}
|
||||||
|
|
||||||
|
try
|
||||||
|
{
|
||||||
|
var json = JsonSerializer.Serialize(request, JsonDefaults.Options);
|
||||||
|
await _db.StringSetAsync(RequestKey(request.Secret), json, ttl).ConfigureAwait(false);
|
||||||
|
await _db.StringSetAsync(CodeKey(request.Code), request.Secret, ttl).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
catch (Exception ex) when (IsTransportFailure(ex))
|
||||||
|
{
|
||||||
|
LogDegraded(ex);
|
||||||
|
await _fallback.SetRequestAsync(request, expiresUtc, cancellationToken).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 _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 _db.StringSetAsync(AuthorizationKey(secret), json, ttl).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public async Task<AuthenticationResult?> TryConsumeAuthorizationAsync(string secret, CancellationToken cancellationToken = default)
|
||||||
|
{
|
||||||
|
var raw = await _db.StringGetDeleteAsync(AuthorizationKey(secret)).ConfigureAwait(false);
|
||||||
|
|
||||||
|
return raw.HasValue
|
||||||
|
? JsonSerializer.Deserialize<AuthenticationResult>(raw.ToString(), JsonDefaults.Options)
|
||||||
|
: null;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Deliberately excludes a malformed stored value, which is a fault of its own rather than a reason
|
||||||
|
// to answer from this instance.
|
||||||
|
private static bool IsTransportFailure(Exception exception) => exception is RedisException or TimeoutException;
|
||||||
|
|
||||||
|
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 void LogDegraded(Exception exception)
|
||||||
|
=> _logger.LogWarning(exception, "Quick connect request state could not be shared through Redis; falling back to this instance only.");
|
||||||
|
}
|
||||||
@@ -59,7 +59,7 @@ public class QuickConnectController : BaseJellyfinApiController
|
|||||||
try
|
try
|
||||||
{
|
{
|
||||||
var auth = await _authContext.GetAuthorizationInfo(Request).ConfigureAwait(false);
|
var auth = await _authContext.GetAuthorizationInfo(Request).ConfigureAwait(false);
|
||||||
return _quickConnect.TryConnect(auth);
|
return await _quickConnect.TryConnect(auth).ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
catch (AuthenticationException)
|
catch (AuthenticationException)
|
||||||
{
|
{
|
||||||
@@ -77,11 +77,11 @@ public class QuickConnectController : BaseJellyfinApiController
|
|||||||
[HttpGet("Connect")]
|
[HttpGet("Connect")]
|
||||||
[ProducesResponseType(StatusCodes.Status200OK)]
|
[ProducesResponseType(StatusCodes.Status200OK)]
|
||||||
[ProducesResponseType(StatusCodes.Status404NotFound)]
|
[ProducesResponseType(StatusCodes.Status404NotFound)]
|
||||||
public ActionResult<QuickConnectResult> GetQuickConnectState([FromQuery, Required] string secret)
|
public async Task<ActionResult<QuickConnectResult>> GetQuickConnectState([FromQuery, Required] string secret)
|
||||||
{
|
{
|
||||||
try
|
try
|
||||||
{
|
{
|
||||||
return _quickConnect.CheckRequestStatus(secret);
|
return await _quickConnect.CheckRequestStatus(secret).ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
catch (ResourceNotFoundException)
|
catch (ResourceNotFoundException)
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -245,11 +245,11 @@ public class UserController : BaseJellyfinApiController
|
|||||||
[HttpPost("AuthenticateWithQuickConnect")]
|
[HttpPost("AuthenticateWithQuickConnect")]
|
||||||
[ProducesResponseType(StatusCodes.Status200OK)]
|
[ProducesResponseType(StatusCodes.Status200OK)]
|
||||||
[Tags("Authentication")]
|
[Tags("Authentication")]
|
||||||
public ActionResult<AuthenticationResult> AuthenticateWithQuickConnect([FromBody, Required] QuickConnectDto request)
|
public async Task<ActionResult<AuthenticationResult>> AuthenticateWithQuickConnect([FromBody, Required] QuickConnectDto request)
|
||||||
{
|
{
|
||||||
try
|
try
|
||||||
{
|
{
|
||||||
return _quickConnectManager.GetAuthorizedRequest(request.Secret);
|
return await _quickConnectManager.GetAuthorizedRequest(request.Secret).ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
catch (SecurityException e)
|
catch (SecurityException e)
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -116,6 +116,10 @@ namespace Jellyfin.Server
|
|||||||
// to the other instances. Redis-backed when configured, no-op otherwise.
|
// to the other instances. Redis-backed when configured, no-op otherwise.
|
||||||
serviceCollection.AddConfigurationInvalidationBus(_startupConfig, Logger);
|
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>())
|
foreach (var type in GetExportTypes<ILyricProvider>())
|
||||||
{
|
{
|
||||||
serviceCollection.AddSingleton(typeof(ILyricProvider), type);
|
serviceCollection.AddSingleton(typeof(ILyricProvider), type);
|
||||||
|
|||||||
@@ -0,0 +1,72 @@
|
|||||||
|
using System;
|
||||||
|
using Emby.Server.Implementations.QuickConnect;
|
||||||
|
using MediaBrowser.Controller.MediaEncoding;
|
||||||
|
using MediaBrowser.Controller.QuickConnect;
|
||||||
|
using Microsoft.Extensions.Configuration;
|
||||||
|
using Microsoft.Extensions.DependencyInjection;
|
||||||
|
using Microsoft.Extensions.Logging;
|
||||||
|
using StackExchange.Redis;
|
||||||
|
|
||||||
|
namespace Jellyfin.Server.Extensions;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Extensions for registering the quick connect store.
|
||||||
|
/// </summary>
|
||||||
|
public static class QuickConnectStoreServiceCollectionExtensions
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Registers the quick connect store, Redis-backed when a connection string is configured and
|
||||||
|
/// process-local otherwise, and reports the selected store at <see cref="LogLevel.Information"/>.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// The connection string is only set for a multi-instance deployment, which is the only shape where
|
||||||
|
/// the initiate, authorize and exchange legs of one flow can land on different instances.
|
||||||
|
/// </remarks>
|
||||||
|
/// <param name="serviceCollection">The service collection.</param>
|
||||||
|
/// <param name="configuration">The configuration to read the Redis connection string from.</param>
|
||||||
|
/// <param name="logger">The logger to report the selected store on.</param>
|
||||||
|
/// <returns>The updated service collection.</returns>
|
||||||
|
public static IServiceCollection AddQuickConnectStore(
|
||||||
|
this IServiceCollection serviceCollection,
|
||||||
|
IConfiguration configuration,
|
||||||
|
ILogger logger)
|
||||||
|
{
|
||||||
|
ArgumentNullException.ThrowIfNull(configuration);
|
||||||
|
ArgumentNullException.ThrowIfNull(logger);
|
||||||
|
|
||||||
|
if (string.IsNullOrEmpty(configuration[TranscodeStoreOptions.RedisConnectionStringKey]))
|
||||||
|
{
|
||||||
|
logger.LogInformation(
|
||||||
|
"Quick connect store: {Store}. A quick connect flow has to complete against one instance; set {Key} to share it.",
|
||||||
|
nameof(InMemoryQuickConnectStore),
|
||||||
|
TranscodeStoreOptions.RedisConnectionStringKey);
|
||||||
|
|
||||||
|
return serviceCollection.AddSingleton<IQuickConnectStore, InMemoryQuickConnectStore>();
|
||||||
|
}
|
||||||
|
|
||||||
|
logger.LogInformation(
|
||||||
|
"Quick connect store: {Store}. Quick connect flows complete across any instance.",
|
||||||
|
nameof(RedisQuickConnectStore));
|
||||||
|
|
||||||
|
return serviceCollection.AddSingleton<IQuickConnectStore>(sp =>
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
return new RedisQuickConnectStore(
|
||||||
|
sp.GetRequiredService<IConnectionMultiplexer>(),
|
||||||
|
sp.GetRequiredService<ILogger<RedisQuickConnectStore>>());
|
||||||
|
}
|
||||||
|
catch (Exception ex)
|
||||||
|
{
|
||||||
|
// Fail open: an unreachable Redis degrades to the single-instance behaviour of a flow
|
||||||
|
// having to complete against one instance, rather than taking quick connect down.
|
||||||
|
sp.GetRequiredService<ILogger<CoreAppHost>>().LogError(
|
||||||
|
ex,
|
||||||
|
"Redis is configured but unavailable, so quick connect flows will not complete across instances. Check {Key}.",
|
||||||
|
TranscodeStoreOptions.RedisConnectionStringKey);
|
||||||
|
|
||||||
|
return new InMemoryQuickConnectStore();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -21,14 +21,14 @@ namespace MediaBrowser.Controller.QuickConnect
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
/// <param name="authorizationInfo">The initiator authorization info.</param>
|
/// <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>
|
/// <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>
|
/// <summary>
|
||||||
/// Checks the status of an individual request.
|
/// Checks the status of an individual request.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
/// <param name="secret">Unique secret identifier of the request.</param>
|
/// <param name="secret">Unique secret identifier of the request.</param>
|
||||||
/// <returns>Quick connect result.</returns>
|
/// <returns>Quick connect result.</returns>
|
||||||
QuickConnectResult CheckRequestStatus(string secret);
|
Task<QuickConnectResult> CheckRequestStatus(string secret);
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Authorizes a quick connect request to connect as the calling user.
|
/// Authorizes a quick connect request to connect as the calling user.
|
||||||
@@ -43,6 +43,6 @@ namespace MediaBrowser.Controller.QuickConnect
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
/// <param name="secret">The secret.</param>
|
/// <param name="secret">The secret.</param>
|
||||||
/// <returns>The authentication result.</returns>
|
/// <returns>The authentication result.</returns>
|
||||||
AuthenticationResult GetAuthorizedRequest(string secret);
|
Task<AuthenticationResult> GetAuthorizedRequest(string secret);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,70 @@
|
|||||||
|
using System;
|
||||||
|
using System.Threading;
|
||||||
|
using System.Threading.Tasks;
|
||||||
|
using MediaBrowser.Controller.Authentication;
|
||||||
|
using MediaBrowser.Model.QuickConnect;
|
||||||
|
|
||||||
|
namespace MediaBrowser.Controller.QuickConnect;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Holds the state of in-flight quick connect requests. The three legs of a quick connect flow -
|
||||||
|
/// initiate, authorize and exchange - can each land on a different instance, so the state has to be
|
||||||
|
/// reachable from all of them.
|
||||||
|
/// </summary>
|
||||||
|
public interface IQuickConnectStore
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Looks up a pending request by the secret handed to the initiating client.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="secret">The request secret.</param>
|
||||||
|
/// <param name="cancellationToken">A cancellation token.</param>
|
||||||
|
/// <returns>The request, or <c>null</c> when it is unknown or has expired.</returns>
|
||||||
|
Task<QuickConnectResult?> GetRequestBySecretAsync(string secret, CancellationToken cancellationToken = default);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Looks up a pending request by the code shown to the user.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="code">The user facing code.</param>
|
||||||
|
/// <param name="cancellationToken">A cancellation token.</param>
|
||||||
|
/// <returns>The request, or <c>null</c> when it is unknown or has expired.</returns>
|
||||||
|
Task<QuickConnectResult?> GetRequestByCodeAsync(string code, CancellationToken cancellationToken = default);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Stores a new or updated request until <paramref name="expiresUtc"/>.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="request">The request to store.</param>
|
||||||
|
/// <param name="expiresUtc">The instant the request stops being resolvable.</param>
|
||||||
|
/// <param name="cancellationToken">A cancellation token.</param>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
Task SetRequestAsync(QuickConnectResult request, DateTime expiresUtc, CancellationToken cancellationToken = default);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// 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.
|
||||||
|
/// </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 being authorized 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>
|
||||||
|
/// Atomically takes the authentication for <paramref name="secret"/> and removes it, so that two
|
||||||
|
/// instances racing on the same secret cannot both hand out an access token.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="secret">The request secret.</param>
|
||||||
|
/// <param name="cancellationToken">A cancellation token.</param>
|
||||||
|
/// <returns>The authentication, or <c>null</c> when the secret is unknown, expired or already exchanged.</returns>
|
||||||
|
Task<AuthenticationResult?> TryConsumeAuthorizationAsync(string secret, CancellationToken cancellationToken = default);
|
||||||
|
}
|
||||||
@@ -0,0 +1,110 @@
|
|||||||
|
using System;
|
||||||
|
using System.Collections.Concurrent;
|
||||||
|
using System.Linq;
|
||||||
|
using System.Threading;
|
||||||
|
using System.Threading.Tasks;
|
||||||
|
using MediaBrowser.Controller.Authentication;
|
||||||
|
using MediaBrowser.Model.QuickConnect;
|
||||||
|
|
||||||
|
namespace MediaBrowser.Controller.QuickConnect;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// A process-local <see cref="IQuickConnectStore"/>. It is the single-instance default, and the
|
||||||
|
/// fallback a shared store degrades to while its backend is unreachable, so quick connect keeps
|
||||||
|
/// working for clients whose three legs happen to land on one instance.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class InMemoryQuickConnectStore : IQuickConnectStore
|
||||||
|
{
|
||||||
|
private readonly ConcurrentDictionary<string, Entry<QuickConnectResult>> _requests = new(StringComparer.Ordinal);
|
||||||
|
private readonly ConcurrentDictionary<string, Entry<AuthenticationResult>> _authorizations = new(StringComparer.Ordinal);
|
||||||
|
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();
|
||||||
|
_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();
|
||||||
|
_authorizations[secret] = new Entry<AuthenticationResult>(expiresUtc, authenticationResult);
|
||||||
|
return Task.CompletedTask;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public Task<AuthenticationResult?> TryConsumeAuthorizationAsync(string secret, CancellationToken cancellationToken = default)
|
||||||
|
{
|
||||||
|
Expire();
|
||||||
|
if (!_authorizations.TryRemove(secret, out var entry) || entry.ExpiresUtc <= DateTime.UtcNow)
|
||||||
|
{
|
||||||
|
return Task.FromResult<AuthenticationResult?>(null);
|
||||||
|
}
|
||||||
|
|
||||||
|
return Task.FromResult<AuthenticationResult?>(entry.Value);
|
||||||
|
}
|
||||||
|
|
||||||
|
private void Expire()
|
||||||
|
{
|
||||||
|
var now = DateTime.UtcNow;
|
||||||
|
|
||||||
|
foreach (var (secret, entry) in _requests)
|
||||||
|
{
|
||||||
|
if (entry.ExpiresUtc <= now)
|
||||||
|
{
|
||||||
|
_requests.TryRemove(secret, out _);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
foreach (var (secret, entry) in _authorizations)
|
||||||
|
{
|
||||||
|
if (entry.ExpiresUtc <= now)
|
||||||
|
{
|
||||||
|
_authorizations.TryRemove(secret, out _);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
foreach (var (secret, expiresUtc) in _authorizationClaims)
|
||||||
|
{
|
||||||
|
if (expiresUtc <= now)
|
||||||
|
{
|
||||||
|
_authorizationClaims.TryRemove(secret, out _);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private sealed record Entry<T>(DateTime ExpiresUtc, T Value);
|
||||||
|
}
|
||||||
+57
-17
@@ -8,6 +8,7 @@ using MediaBrowser.Common.Extensions;
|
|||||||
using MediaBrowser.Controller.Authentication;
|
using MediaBrowser.Controller.Authentication;
|
||||||
using MediaBrowser.Controller.Configuration;
|
using MediaBrowser.Controller.Configuration;
|
||||||
using MediaBrowser.Controller.Net;
|
using MediaBrowser.Controller.Net;
|
||||||
|
using MediaBrowser.Controller.QuickConnect;
|
||||||
using MediaBrowser.Model.Configuration;
|
using MediaBrowser.Model.Configuration;
|
||||||
using Moq;
|
using Moq;
|
||||||
using Xunit;
|
using Xunit;
|
||||||
@@ -40,6 +41,8 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect
|
|||||||
ConfigureMembers = true
|
ConfigureMembers = true
|
||||||
}).Inject(configManager.Object);
|
}).Inject(configManager.Object);
|
||||||
|
|
||||||
|
_fixture.Inject<IQuickConnectStore>(new InMemoryQuickConnectStore());
|
||||||
|
|
||||||
// User object contains circular references.
|
// User object contains circular references.
|
||||||
_fixture.Behaviors.OfType<ThrowingRecursionBehavior>().ToList()
|
_fixture.Behaviors.OfType<ThrowingRecursionBehavior>().ToList()
|
||||||
.ForEach(b => _fixture.Behaviors.Remove(b));
|
.ForEach(b => _fixture.Behaviors.Remove(b));
|
||||||
@@ -60,8 +63,8 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect
|
|||||||
[InlineData("Device", "", "Client", "1.0.0")]
|
[InlineData("Device", "", "Client", "1.0.0")]
|
||||||
[InlineData("Device", "DeviceId", "", "1.0.0")]
|
[InlineData("Device", "DeviceId", "", "1.0.0")]
|
||||||
[InlineData("Device", "DeviceId", "Client", "")]
|
[InlineData("Device", "DeviceId", "Client", "")]
|
||||||
public void TryConnect_InvalidAuthorizationInfo_ThrowsArgumentException(string device, string deviceId, string client, string version)
|
public async Task TryConnect_InvalidAuthorizationInfo_ThrowsArgumentException(string device, string deviceId, string client, string version)
|
||||||
=> Assert.Throws<ArgumentException>(() => _quickConnectManager.TryConnect(
|
=> await Assert.ThrowsAsync<ArgumentException>(() => _quickConnectManager.TryConnect(
|
||||||
new AuthorizationInfo
|
new AuthorizationInfo
|
||||||
{
|
{
|
||||||
Device = device,
|
Device = device,
|
||||||
@@ -71,17 +74,17 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect
|
|||||||
}));
|
}));
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void TryConnect_QuickConnectUnavailable_ThrowsAuthenticationException()
|
public async Task TryConnect_QuickConnectUnavailable_ThrowsAuthenticationException()
|
||||||
{
|
{
|
||||||
_config.QuickConnectAvailable = false;
|
_config.QuickConnectAvailable = false;
|
||||||
Assert.Throws<AuthenticationException>(() => _quickConnectManager.TryConnect(_quickConnectAuthInfo));
|
await Assert.ThrowsAsync<AuthenticationException>(() => _quickConnectManager.TryConnect(_quickConnectAuthInfo));
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void CheckRequestStatus_QuickConnectUnavailable_ThrowsAuthenticationException()
|
public async Task CheckRequestStatus_QuickConnectUnavailable_ThrowsAuthenticationException()
|
||||||
{
|
{
|
||||||
_config.QuickConnectAvailable = false;
|
_config.QuickConnectAvailable = false;
|
||||||
Assert.Throws<AuthenticationException>(() => _quickConnectManager.CheckRequestStatus(string.Empty));
|
await Assert.ThrowsAsync<AuthenticationException>(() => _quickConnectManager.CheckRequestStatus(string.Empty));
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
@@ -92,10 +95,10 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect
|
|||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void GetAuthorizedRequest_QuickConnectUnavailable_ThrowsAuthenticationException()
|
public async Task GetAuthorizedRequest_QuickConnectUnavailable_ThrowsAuthenticationException()
|
||||||
{
|
{
|
||||||
_config.QuickConnectAvailable = false;
|
_config.QuickConnectAvailable = false;
|
||||||
Assert.Throws<AuthenticationException>(() => _quickConnectManager.GetAuthorizedRequest(string.Empty));
|
await Assert.ThrowsAsync<AuthenticationException>(() => _quickConnectManager.GetAuthorizedRequest(string.Empty));
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
@@ -106,34 +109,71 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect
|
|||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void CheckRequestStatus_QuickConnectAvailable_Success()
|
public async Task CheckRequestStatus_QuickConnectAvailable_Success()
|
||||||
{
|
{
|
||||||
_config.QuickConnectAvailable = true;
|
_config.QuickConnectAvailable = true;
|
||||||
var res1 = _quickConnectManager.TryConnect(_quickConnectAuthInfo);
|
var res1 = await _quickConnectManager.TryConnect(_quickConnectAuthInfo);
|
||||||
var res2 = _quickConnectManager.CheckRequestStatus(res1.Secret);
|
var res2 = await _quickConnectManager.CheckRequestStatus(res1.Secret);
|
||||||
Assert.Equal(res1, res2);
|
Assert.Equal(res1.Secret, res2.Secret);
|
||||||
|
Assert.Equal(res1.Code, res2.Code);
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void CheckRequestStatus_UnknownSecret_ThrowsResourceNotFoundException()
|
public async Task CheckRequestStatus_UnknownSecret_ThrowsResourceNotFoundException()
|
||||||
{
|
{
|
||||||
_config.QuickConnectAvailable = true;
|
_config.QuickConnectAvailable = true;
|
||||||
Assert.Throws<ResourceNotFoundException>(() => _quickConnectManager.CheckRequestStatus("Unknown secret"));
|
await Assert.ThrowsAsync<ResourceNotFoundException>(() => _quickConnectManager.CheckRequestStatus("Unknown secret"));
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void GetAuthorizedRequest_UnknownSecret_ThrowsResourceNotFoundException()
|
public async Task GetAuthorizedRequest_UnknownSecret_ThrowsResourceNotFoundException()
|
||||||
{
|
{
|
||||||
_config.QuickConnectAvailable = true;
|
_config.QuickConnectAvailable = true;
|
||||||
Assert.Throws<ResourceNotFoundException>(() => _quickConnectManager.GetAuthorizedRequest("Unknown secret"));
|
await Assert.ThrowsAsync<ResourceNotFoundException>(() => _quickConnectManager.GetAuthorizedRequest("Unknown secret"));
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public async Task AuthorizeRequest_QuickConnectAvailable_Success()
|
public async Task AuthorizeRequest_QuickConnectAvailable_Success()
|
||||||
{
|
{
|
||||||
_config.QuickConnectAvailable = true;
|
_config.QuickConnectAvailable = true;
|
||||||
var res = _quickConnectManager.TryConnect(_quickConnectAuthInfo);
|
var res = await _quickConnectManager.TryConnect(_quickConnectAuthInfo);
|
||||||
Assert.True(await _quickConnectManager.AuthorizeRequest(Guid.Empty, res.Code));
|
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_ThrowsResourceNotFoundException()
|
||||||
|
{
|
||||||
|
_config.QuickConnectAvailable = true;
|
||||||
|
var res = await _quickConnectManager.TryConnect(_quickConnectAuthInfo);
|
||||||
|
await _quickConnectManager.AuthorizeRequest(Guid.Empty, res.Code);
|
||||||
|
|
||||||
|
Assert.NotNull(await _quickConnectManager.GetAuthorizedRequest(res.Secret));
|
||||||
|
await Assert.ThrowsAsync<ResourceNotFoundException>(() => _quickConnectManager.GetAuthorizedRequest(res.Secret));
|
||||||
|
}
|
||||||
|
|
||||||
|
private async Task<bool> AuthorizeAsync(string code)
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
return await _quickConnectManager.AuthorizeRequest(Guid.Empty, code).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
catch (InvalidOperationException)
|
||||||
|
{
|
||||||
|
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,
|
/// 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.
|
/// a correctly set variable is dropped and the feature it configures stays off without any error.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
|
[Collection("JellyfinSectionConfiguration")]
|
||||||
public sealed class JellyfinSectionConfigurationTests : IDisposable
|
public sealed class JellyfinSectionConfigurationTests : IDisposable
|
||||||
{
|
{
|
||||||
private const string RedisKey = "Jellyfin:TranscodeStore:RedisConnectionString";
|
private const string RedisKey = "Jellyfin:TranscodeStore:RedisConnectionString";
|
||||||
|
|||||||
@@ -0,0 +1,170 @@
|
|||||||
|
using System;
|
||||||
|
using System.Collections.Concurrent;
|
||||||
|
using System.Globalization;
|
||||||
|
using System.IO;
|
||||||
|
using System.Net;
|
||||||
|
using System.Net.Sockets;
|
||||||
|
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 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>
|
||||||
|
/// 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() => _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;
|
||||||
|
|
||||||
|
var clientStream = client.GetStream();
|
||||||
|
var upstreamStream = upstream.GetStream();
|
||||||
|
await Task.WhenAny(
|
||||||
|
CopyAsync(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 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="Microsoft.NET.Test.Sdk" />
|
||||||
<PackageReference Include="Npgsql" />
|
<PackageReference Include="Npgsql" />
|
||||||
<PackageReference Include="Testcontainers.PostgreSql" />
|
<PackageReference Include="Testcontainers.PostgreSql" />
|
||||||
|
<PackageReference Include="Testcontainers.Redis" />
|
||||||
<PackageReference Include="xunit.v3" />
|
<PackageReference Include="xunit.v3" />
|
||||||
<PackageReference Include="xunit.runner.visualstudio">
|
<PackageReference Include="xunit.runner.visualstudio">
|
||||||
<PrivateAssets>all</PrivateAssets>
|
<PrivateAssets>all</PrivateAssets>
|
||||||
|
|||||||
@@ -0,0 +1,369 @@
|
|||||||
|
using System;
|
||||||
|
using System.Collections.Generic;
|
||||||
|
using System.Globalization;
|
||||||
|
using System.Threading;
|
||||||
|
using System.Threading.Tasks;
|
||||||
|
using Emby.Server.Implementations.QuickConnect;
|
||||||
|
using Jellyfin.Data.Queries;
|
||||||
|
using Jellyfin.Database.Implementations;
|
||||||
|
using Jellyfin.Database.Implementations.DbConfiguration;
|
||||||
|
using Jellyfin.Database.Implementations.Entities;
|
||||||
|
using Jellyfin.Database.Implementations.Entities.Security;
|
||||||
|
using Jellyfin.Database.Implementations.Locking;
|
||||||
|
using Jellyfin.Database.Providers.PostgreSQL;
|
||||||
|
using Jellyfin.Server.Implementations.Devices;
|
||||||
|
using Jellyfin.Server.Tests.HighAvailability;
|
||||||
|
using Jellyfin.Server.Tests.Migrations;
|
||||||
|
using MediaBrowser.Common.Extensions;
|
||||||
|
using MediaBrowser.Controller.Authentication;
|
||||||
|
using MediaBrowser.Controller.Configuration;
|
||||||
|
using MediaBrowser.Controller.Devices;
|
||||||
|
using MediaBrowser.Controller.Library;
|
||||||
|
using MediaBrowser.Controller.Net;
|
||||||
|
using MediaBrowser.Controller.QuickConnect;
|
||||||
|
using MediaBrowser.Controller.Session;
|
||||||
|
using MediaBrowser.Model.Configuration;
|
||||||
|
using MediaBrowser.Model.Dto;
|
||||||
|
using MediaBrowser.Model.QuickConnect;
|
||||||
|
using Microsoft.EntityFrameworkCore;
|
||||||
|
using Microsoft.Extensions.Logging.Abstractions;
|
||||||
|
using Moq;
|
||||||
|
using Npgsql;
|
||||||
|
using StackExchange.Redis;
|
||||||
|
using Xunit;
|
||||||
|
|
||||||
|
namespace Jellyfin.Server.Tests.QuickConnect;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Three independently constructed <see cref="QuickConnectManager"/> instances over one PostgreSQL
|
||||||
|
/// database and one Redis are the in-process stand-in for three replicas without sticky sessions: the
|
||||||
|
/// initiate, authorize and exchange legs of one flow each land on a different one.
|
||||||
|
/// </summary>
|
||||||
|
[Trait("Category", "RequiresDocker")]
|
||||||
|
public sealed class QuickConnectReplicaTests : IAsyncLifetime
|
||||||
|
{
|
||||||
|
private static readonly AuthorizationInfo _authorizationInfo = new AuthorizationInfo
|
||||||
|
{
|
||||||
|
Device = "Living Room TV",
|
||||||
|
DeviceId = "device-1",
|
||||||
|
Client = "Jellyfin Web",
|
||||||
|
Version = "1.0.0"
|
||||||
|
};
|
||||||
|
|
||||||
|
private readonly List<IConnectionMultiplexer> _connections = new();
|
||||||
|
|
||||||
|
private PostgreSqlTestServer _postgres = null!;
|
||||||
|
private RedisTestServer _redis = null!;
|
||||||
|
|
||||||
|
/// <inheritdoc/>
|
||||||
|
public async ValueTask InitializeAsync()
|
||||||
|
{
|
||||||
|
_postgres = await PostgreSqlTestServer.StartAsync().ConfigureAwait(false);
|
||||||
|
_redis = await RedisTestServer.StartAsync().ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc/>
|
||||||
|
public async ValueTask DisposeAsync()
|
||||||
|
{
|
||||||
|
foreach (var connection in _connections)
|
||||||
|
{
|
||||||
|
await connection.DisposeAsync().ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
|
||||||
|
await _redis.DisposeAsync().ConfigureAwait(false);
|
||||||
|
await _postgres.DisposeAsync().ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The three legs of a quick connect flow land on three different replicas, and the token the third
|
||||||
|
/// one hands out is the one the second one minted into the shared database.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task InitiateAuthorizeExchange_AcrossThreeReplicas_Succeeds()
|
||||||
|
{
|
||||||
|
var cancellationToken = TestContext.Current.CancellationToken;
|
||||||
|
var connectionString = await _postgres.CreateDatabaseAsync("quickconnect_replica_flow", cancellationToken);
|
||||||
|
|
||||||
|
await using var dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
|
||||||
|
var user = await CreateSchemaWithUserAsync(dataSource, cancellationToken);
|
||||||
|
|
||||||
|
var replicaA = await CreateReplicaAsync(dataSource, user);
|
||||||
|
var replicaB = await CreateReplicaAsync(dataSource, user);
|
||||||
|
var replicaC = await CreateReplicaAsync(dataSource, user);
|
||||||
|
|
||||||
|
var initiated = await replicaA.Manager.TryConnect(_authorizationInfo);
|
||||||
|
|
||||||
|
// The code is shown to the user on whichever replica serves the dashboard.
|
||||||
|
Assert.True(await replicaB.Manager.AuthorizeRequest(user.Id, initiated.Code));
|
||||||
|
|
||||||
|
var polled = await replicaC.Manager.CheckRequestStatus(initiated.Secret);
|
||||||
|
Assert.True(polled.Authenticated);
|
||||||
|
Assert.Equal(initiated.Code, polled.Code);
|
||||||
|
Assert.Equal(_authorizationInfo.DeviceId, polled.DeviceId);
|
||||||
|
|
||||||
|
var exchanged = await replicaC.Manager.GetAuthorizedRequest(initiated.Secret);
|
||||||
|
|
||||||
|
Assert.False(string.IsNullOrEmpty(exchanged.AccessToken));
|
||||||
|
Assert.Equal(user.Id, exchanged.User.Id);
|
||||||
|
|
||||||
|
var devices = await replicaA.Devices.GetDevices(new DeviceQuery { AccessToken = exchanged.AccessToken });
|
||||||
|
Assert.Equal(user.Id, Assert.Single(devices.Items).UserId);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// A secret is single use across the whole deployment: two replicas racing to exchange it must not
|
||||||
|
/// both hand out an access token. One scheduling of one race settles nothing either way, so the race
|
||||||
|
/// is run repeatedly.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task Exchange_RacedOnTwoReplicas_SucceedsOnce()
|
||||||
|
{
|
||||||
|
var cancellationToken = TestContext.Current.CancellationToken;
|
||||||
|
var connectionString = await _postgres.CreateDatabaseAsync("quickconnect_replica_race", cancellationToken);
|
||||||
|
|
||||||
|
await using var dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
|
||||||
|
var user = await CreateSchemaWithUserAsync(dataSource, cancellationToken);
|
||||||
|
|
||||||
|
var replicaA = await CreateReplicaAsync(dataSource, user);
|
||||||
|
var replicaB = await CreateReplicaAsync(dataSource, user);
|
||||||
|
var replicaC = await CreateReplicaAsync(dataSource, user);
|
||||||
|
|
||||||
|
for (var attempt = 0; attempt < 25; attempt++)
|
||||||
|
{
|
||||||
|
var initiated = await replicaA.Manager.TryConnect(AuthorizationInfoFor(attempt));
|
||||||
|
await replicaB.Manager.AuthorizeRequest(user.Id, initiated.Code);
|
||||||
|
|
||||||
|
var outcomes = await Task.WhenAll(
|
||||||
|
Task.Run(() => ExchangeAsync(replicaA.Manager, initiated.Secret), cancellationToken),
|
||||||
|
Task.Run(() => ExchangeAsync(replicaC.Manager, initiated.Secret), cancellationToken));
|
||||||
|
|
||||||
|
Assert.Single(outcomes, outcome => outcome is not null);
|
||||||
|
|
||||||
|
// And it stays consumed for every later attempt, on any replica.
|
||||||
|
await Assert.ThrowsAsync<ResourceNotFoundException>(() => replicaB.Manager.GetAuthorizedRequest(initiated.Secret));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// 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 (InvalidOperationException)
|
||||||
|
{
|
||||||
|
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,142 @@
|
|||||||
|
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 + ",abortConnect=false");
|
||||||
|
|
||||||
|
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.
|
||||||
|
/// </summary>
|
||||||
|
[Fact]
|
||||||
|
public void NoEnvironmentVariable_SelectsTheProcessLocalStore()
|
||||||
|
{
|
||||||
|
Environment.SetEnvironmentVariable(RedisConnectionStringVariable, null);
|
||||||
|
|
||||||
|
using var provider = BuildProvider();
|
||||||
|
|
||||||
|
Assert.IsType<InMemoryQuickConnectStore>(provider.GetRequiredService<IQuickConnectStore>());
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// A configured but unreachable Redis degrades to the single-instance behaviour of a flow having to
|
||||||
|
/// complete against one instance, rather than taking quick connect down at startup.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task UnreachableRedisAtStartup_DegradesToTheProcessLocalStore()
|
||||||
|
{
|
||||||
|
Environment.SetEnvironmentVariable(RedisConnectionStringVariable, "127.0.0.1:1,connectTimeout=250,connectRetry=0");
|
||||||
|
|
||||||
|
await using var provider = BuildProvider();
|
||||||
|
|
||||||
|
var store = provider.GetRequiredService<IQuickConnectStore>();
|
||||||
|
Assert.IsType<InMemoryQuickConnectStore>(store);
|
||||||
|
|
||||||
|
// Quick connect still works, it just cannot span instances.
|
||||||
|
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));
|
||||||
|
await store.SetAuthorizationAsync(
|
||||||
|
request.Secret,
|
||||||
|
new AuthenticationResult { AccessToken = "token-1" },
|
||||||
|
DateTime.UtcNow.AddMinutes(10),
|
||||||
|
TestContext.Current.CancellationToken);
|
||||||
|
|
||||||
|
Assert.Equal("token-1", (await store.TryConsumeAuthorizationAsync(request.Secret, TestContext.Current.CancellationToken))?.AccessToken);
|
||||||
|
Assert.Null(await store.TryConsumeAuthorizationAsync(request.Secret, TestContext.Current.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 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,308 @@
|
|||||||
|
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.Controller.Authentication;
|
||||||
|
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>
|
||||||
|
/// A request stored while Redis is unreachable is still resolvable on the instance that stored it,
|
||||||
|
/// so a flow whose three legs happen to land on one instance keeps working through the outage.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task PendingRequest_SurvivesAnOutage_OnTheInstanceThatStoredIt()
|
||||||
|
{
|
||||||
|
var instance = await CreateInstanceAsync();
|
||||||
|
var request = NewRequest();
|
||||||
|
|
||||||
|
instance.Proxy.Cut();
|
||||||
|
await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken);
|
||||||
|
|
||||||
|
Assert.Equal(request.Secret, (await instance.Store.GetRequestBySecretAsync(request.Secret, CancellationToken))?.Secret);
|
||||||
|
Assert.Equal(request.Secret, (await instance.Store.GetRequestByCodeAsync(request.Code, CancellationToken))?.Secret);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Once Redis answers again it is the only authority: a miss is a miss, not a reason to serve the
|
||||||
|
/// copy this instance kept while it was unreachable.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task PendingRequest_StoredDuringAnOutage_IsNotServedOnceRedisAnswersAgain()
|
||||||
|
{
|
||||||
|
var instance = await CreateInstanceAsync();
|
||||||
|
var request = NewRequest();
|
||||||
|
|
||||||
|
instance.Proxy.Cut();
|
||||||
|
await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken);
|
||||||
|
Assert.NotNull(await instance.Store.GetRequestBySecretAsync(request.Secret, CancellationToken));
|
||||||
|
|
||||||
|
await RestoreAsync(instance);
|
||||||
|
|
||||||
|
Assert.Null(await instance.Store.GetRequestBySecretAsync(request.Secret, CancellationToken));
|
||||||
|
Assert.Null(await instance.Store.GetRequestByCodeAsync(request.Code, CancellationToken));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// A malformed stored value is a fault of its own, not a transport failure, so it is surfaced rather
|
||||||
|
/// than answered from the copy this instance happens to hold.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task PendingRequest_ThatIsMalformedInRedis_SurfacesInsteadOfDegrading()
|
||||||
|
{
|
||||||
|
var instance = await CreateInstanceAsync();
|
||||||
|
var request = NewRequest();
|
||||||
|
|
||||||
|
instance.Proxy.Cut();
|
||||||
|
await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken);
|
||||||
|
await RestoreAsync(instance);
|
||||||
|
|
||||||
|
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>
|
||||||
|
/// An authorization write that failed leaves nothing behind on the instance, because the response
|
||||||
|
/// that never arrived may still have been applied and a second copy of an authorization is a second
|
||||||
|
/// access token.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task Authorization_ThatFailedToStore_LeavesNothingOnTheInstance()
|
||||||
|
{
|
||||||
|
var instance = await CreateInstanceAsync();
|
||||||
|
var request = NewRequest();
|
||||||
|
await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken);
|
||||||
|
|
||||||
|
instance.Proxy.Cut();
|
||||||
|
await AssertTransportFailureAsync(() => instance.Store.SetAuthorizationAsync(
|
||||||
|
request.Secret,
|
||||||
|
new AuthenticationResult { AccessToken = "token-1" },
|
||||||
|
DateTime.UtcNow.AddMinutes(10),
|
||||||
|
CancellationToken));
|
||||||
|
|
||||||
|
await RestoreAsync(instance);
|
||||||
|
|
||||||
|
Assert.Null(await instance.Store.TryConsumeAuthorizationAsync(request.Secret, CancellationToken));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The instance whose authorization write failed while the write landed anyway still hands the token
|
||||||
|
/// out exactly once, rather than once from Redis and again from a copy of its own.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task Authorization_IsHandedOutOnce_EvenAfterAFailedWriteOnTheSameInstance()
|
||||||
|
{
|
||||||
|
var instance = await CreateInstanceAsync();
|
||||||
|
var request = NewRequest();
|
||||||
|
var authentication = new AuthenticationResult { AccessToken = "token-1" };
|
||||||
|
await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken);
|
||||||
|
|
||||||
|
instance.Proxy.Cut();
|
||||||
|
await AssertTransportFailureAsync(() => instance.Store.SetAuthorizationAsync(
|
||||||
|
request.Secret,
|
||||||
|
authentication,
|
||||||
|
DateTime.UtcNow.AddMinutes(10),
|
||||||
|
CancellationToken));
|
||||||
|
|
||||||
|
await RestoreAsync(instance);
|
||||||
|
|
||||||
|
// Stands in for that write having been applied before the response was lost.
|
||||||
|
await instance.Store.SetAuthorizationAsync(request.Secret, authentication, DateTime.UtcNow.AddMinutes(10), CancellationToken);
|
||||||
|
|
||||||
|
Assert.Equal("token-1", (await instance.Store.TryConsumeAuthorizationAsync(request.Secret, CancellationToken))?.AccessToken);
|
||||||
|
Assert.Null(await instance.Store.TryConsumeAuthorizationAsync(request.Secret, CancellationToken));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Exchanging during an outage fails loudly and spends nothing, so the token is still there to be
|
||||||
|
/// handed out once when Redis comes back.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task Exchange_DuringAnOutage_SurfacesTheFailureAndLeavesTheTokenUnspent()
|
||||||
|
{
|
||||||
|
var instance = await CreateInstanceAsync();
|
||||||
|
var request = NewRequest();
|
||||||
|
await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken);
|
||||||
|
await instance.Store.SetAuthorizationAsync(
|
||||||
|
request.Secret,
|
||||||
|
new AuthenticationResult { AccessToken = "token-1" },
|
||||||
|
DateTime.UtcNow.AddMinutes(10),
|
||||||
|
CancellationToken);
|
||||||
|
|
||||||
|
instance.Proxy.Cut();
|
||||||
|
await AssertTransportFailureAsync(() => instance.Store.TryConsumeAuthorizationAsync(request.Secret, CancellationToken));
|
||||||
|
|
||||||
|
await RestoreAsync(instance);
|
||||||
|
|
||||||
|
Assert.Equal("token-1", (await instance.Store.TryConsumeAuthorizationAsync(request.Secret, CancellationToken))?.AccessToken);
|
||||||
|
Assert.Null(await instance.Store.TryConsumeAuthorizationAsync(request.Secret, CancellationToken));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Authorizing during an outage fails loudly rather than claiming locally, because a claim only this
|
||||||
|
/// instance knows about does not stop another one minting a second access token.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task Claim_DuringAnOutage_SurfacesTheFailure()
|
||||||
|
{
|
||||||
|
var instance = await CreateInstanceAsync();
|
||||||
|
var request = NewRequest();
|
||||||
|
await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken);
|
||||||
|
|
||||||
|
instance.Proxy.Cut();
|
||||||
|
await AssertTransportFailureAsync(() => instance.Store.TryClaimAuthorizationAsync(
|
||||||
|
request.Secret,
|
||||||
|
DateTime.UtcNow.AddMinutes(10),
|
||||||
|
CancellationToken));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <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));
|
||||||
|
}
|
||||||
|
|
||||||
|
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 AssertTransportFailureAsync(Func<Task> operation)
|
||||||
|
{
|
||||||
|
var exception = await Record.ExceptionAsync(operation);
|
||||||
|
|
||||||
|
Assert.NotNull(exception);
|
||||||
|
Assert.True(exception is RedisException or TimeoutException, exception.ToString());
|
||||||
|
}
|
||||||
|
|
||||||
|
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