feat(server): add observability and operator controls (#16)

src/FinalFactory.Rendezvous.Server/Abuse/AbuseProtectionOptions.cs

@@ -20,6 +20,8 @@ internal sealed class AbuseProtectionOptions
 
     public string[] TrustedProxyAddresses { get; set; } = [];
 
+    public string[] OperatorAllowedAddresses { get; set; } = [];
+
     [Range(1, 100_000)]
     public int HealthGlobalRequestsPerWindow { get; set; } = 1_000;
 
@@ -32,6 +34,18 @@ internal sealed class AbuseProtectionOptions
     [Range(1, 1_000)]
     public int HealthIpPrefixConcurrency { get; set; } = 8;
 
+    [Range(1, 100_000)]
+    public int OperatorGlobalRequestsPerWindow { get; set; } = 1_000;
+
+    [Range(1, 10_000)]
+    public int OperatorGlobalConcurrency { get; set; } = 32;
+
+    [Range(1, 100_000)]
+    public int OperatorIpPrefixRequestsPerWindow { get; set; } = 120;
+
+    [Range(1, 1_000)]
+    public int OperatorIpPrefixConcurrency { get; set; } = 8;
+
     [Range(1, 1_000_000)]
     public int HttpGlobalRequestsPerWindow { get; set; } = 20_000;
 

src/FinalFactory.Rendezvous.Server/Abuse/AbuseProtectionService.cs

@@ -2,6 +2,7 @@ using System.Buffers;
 using System.Net;
 using System.Security.Cryptography;
 using System.Text;
+using FinalFactory.Rendezvous.Server.Observability;
 using Microsoft.Extensions.Options;
 
 namespace FinalFactory.Rendezvous.Server.Abuse;
@@ -12,13 +13,23 @@ internal sealed class AbuseProtectionService
     private readonly TimeProvider _timeProvider;
     private readonly TrackerState _httpTracker;
     private readonly TrackerState _udpTracker;
+    private readonly RendezvousTelemetry? _telemetry;
+    private readonly HashSet<string> _operatorAllowedAddresses;
 
     public AbuseProtectionService(
         IOptions<AbuseProtectionOptions> options,
-        TimeProvider? timeProvider = null)
+        TimeProvider? timeProvider = null,
+        RendezvousTelemetry? telemetry = null)
     {
         _options = options.Value;
         _timeProvider = timeProvider ?? TimeProvider.System;
+        _telemetry = telemetry;
+        _operatorAllowedAddresses = options.Value.OperatorAllowedAddresses
+            .Select(static value => IPAddress.TryParse(value, out IPAddress? address)
+                ? NormalizeAddress(address).ToString()
+                : string.Empty)
+            .Where(static value => value.Length > 0)
+            .ToHashSet(StringComparer.Ordinal);
         DateTimeOffset now = _timeProvider.GetUtcNow();
         _httpTracker = new(now);
         _udpTracker = new(now);
@@ -87,6 +98,35 @@ internal sealed class AbuseProtectionService
             out retryAfterSeconds);
     }
 
+    public bool IsOperatorSourceAllowed(IPAddress? remoteAddress) =>
+        remoteAddress is not null
+        && _operatorAllowedAddresses.Contains(NormalizeAddress(remoteAddress).ToString());
+
+    public bool TryAcquireOperatorIngress(
+        IPAddress? remoteAddress,
+        out AbuseLease? lease,
+        out int retryAfterSeconds)
+    {
+        string prefix = GetNetworkPrefix(remoteAddress);
+        RateDimension[] rates =
+        [
+            new("operator:rate:global", _options.OperatorGlobalRequestsPerWindow),
+            new($"operator:rate:ip:{prefix}", _options.OperatorIpPrefixRequestsPerWindow),
+        ];
+        RateDimension[] concurrency =
+        [
+            new("operator:concurrency:global", _options.OperatorGlobalConcurrency),
+            new($"operator:concurrency:ip:{prefix}", _options.OperatorIpPrefixConcurrency),
+        ];
+        return TryAcquire(
+            rates,
+            concurrency,
+            TrackerDomain.Http,
+            true,
+            out lease,
+            out retryAfterSeconds);
+    }
+
     public bool TryAcquireHttpIdentity(
         string operation,
         string? tenant,
@@ -266,67 +306,96 @@ internal sealed class AbuseProtectionService
         out int retryAfterSeconds)
     {
         TrackerState tracker = domain == TrackerDomain.Udp ? _udpTracker : _httpTracker;
+        bool accepted;
         lock (tracker.Gate)
         {
-            DateTimeOffset now = _timeProvider.GetUtcNow();
-            TimeSpan window = TimeSpan.FromSeconds(_options.WindowSeconds);
-            if (now - tracker.WindowStartedAt >= window || now < tracker.WindowStartedAt)
-            {
-                tracker.WindowCounts.Clear();
-                tracker.WindowStartedAt = now;
-            }
+            accepted = TryAcquireLocked(
+                tracker,
+                rates,
+                concurrency,
+                domain,
+                canUseCriticalReserve,
+                out lease,
+                out retryAfterSeconds);
+        }
 
-            retryAfterSeconds = Math.Max(
-                1,
-                (int)Math.Ceiling((window - (now - tracker.WindowStartedAt)).TotalSeconds));
-            int stagedNewKeys = 0;
-            int partitionLimit = domain == TrackerDomain.Udp
-                ? _options.UdpTrackedKeyLimit
-                : _options.MaxTrackedKeys - _options.UdpTrackedKeyLimit;
-            int maxTrackedKeys = domain == TrackerDomain.Udp || canUseCriticalReserve
-                ? partitionLimit
-                : partitionLimit - _options.CriticalTrackedKeyReserve;
-            if (!CanAcquireAll(
-                    tracker,
-                    tracker.WindowCounts,
-                    rates,
-                    maxTrackedKeys,
-                    ref stagedNewKeys)
-                || !CanAcquireAll(
-                    tracker,
-                    tracker.ConcurrencyCounts,
-                    concurrency,
-                    maxTrackedKeys,
-                    ref stagedNewKeys))
-            {
-                lease = null;
-                return false;
-            }
+        if (!accepted)
+        {
+            _telemetry?.RecordLimiterDrop(
+                domain == TrackerDomain.Udp ? "udp" : "http",
+                "rate-or-concurrency");
+        }
 
-            foreach (RateDimension dimension in rates)
-            {
-                tracker.WindowCounts[dimension.Key] =
-                    tracker.WindowCounts.GetValueOrDefault(dimension.Key) + 1;
-            }
+        return accepted;
+    }
 
-            if (concurrency.IsEmpty)
-            {
-                lease = null;
-                return true;
-            }
+    private bool TryAcquireLocked(
+        TrackerState tracker,
+        ReadOnlySpan<RateDimension> rates,
+        ReadOnlySpan<RateDimension> concurrency,
+        TrackerDomain domain,
+        bool canUseCriticalReserve,
+        out AbuseLease? lease,
+        out int retryAfterSeconds)
+    {
+        DateTimeOffset now = _timeProvider.GetUtcNow();
+        TimeSpan window = TimeSpan.FromSeconds(_options.WindowSeconds);
+        if (now - tracker.WindowStartedAt >= window || now < tracker.WindowStartedAt)
+        {
+            tracker.WindowCounts.Clear();
+            tracker.WindowStartedAt = now;
+        }
 
-            string[] acquiredConcurrency = new string[concurrency.Length];
-            for (int index = 0; index < concurrency.Length; index++)
-            {
-                RateDimension dimension = concurrency[index];
-                tracker.ConcurrencyCounts[dimension.Key] =
-                    tracker.ConcurrencyCounts.GetValueOrDefault(dimension.Key) + 1;
-                acquiredConcurrency[index] = dimension.Key;
-            }
+        retryAfterSeconds = Math.Max(
+            1,
+            (int)Math.Ceiling((window - (now - tracker.WindowStartedAt)).TotalSeconds));
+        int stagedNewKeys = 0;
+        int partitionLimit = domain == TrackerDomain.Udp
+            ? _options.UdpTrackedKeyLimit
+            : _options.MaxTrackedKeys - _options.UdpTrackedKeyLimit;
+        int maxTrackedKeys = domain == TrackerDomain.Udp || canUseCriticalReserve
+            ? partitionLimit
+            : partitionLimit - _options.CriticalTrackedKeyReserve;
+        if (!CanAcquireAll(
+                tracker,
+                tracker.WindowCounts,
+                rates,
+                maxTrackedKeys,
+                ref stagedNewKeys)
+            || !CanAcquireAll(
+                tracker,
+                tracker.ConcurrencyCounts,
+                concurrency,
+                maxTrackedKeys,
+                ref stagedNewKeys))
+        {
+            lease = null;
+            return false;
+        }
 
-            lease = new AbuseLease(this, tracker, acquiredConcurrency);
+        foreach (RateDimension dimension in rates)
+        {
+            tracker.WindowCounts[dimension.Key] =
+                tracker.WindowCounts.GetValueOrDefault(dimension.Key) + 1;
+        }
+
+        if (concurrency.IsEmpty)
+        {
+            lease = null;
             return true;
         }
+
+        string[] acquiredConcurrency = new string[concurrency.Length];
+        for (int index = 0; index < concurrency.Length; index++)
+        {
+            RateDimension dimension = concurrency[index];
+            tracker.ConcurrencyCounts[dimension.Key] =
+                tracker.ConcurrencyCounts.GetValueOrDefault(dimension.Key) + 1;
+            acquiredConcurrency[index] = dimension.Key;
+        }
+
+        lease = new AbuseLease(this, tracker, acquiredConcurrency);
+        return true;
     }
 
     private static bool CanAcquireAll(
@@ -415,6 +484,9 @@ internal sealed class AbuseProtectionService
         return "unknown";
     }
 
+    private static IPAddress NormalizeAddress(IPAddress address) =>
+        address.IsIPv4MappedToIPv6 ? address.MapToIPv4() : address;
+
     private readonly record struct RateDimension(string Key, int Limit);
 
     private enum TrackerDomain

src/FinalFactory.Rendezvous.Server/Abuse/HttpAbuseProtectionMiddleware.cs

@@ -19,16 +19,71 @@ internal sealed class HttpAbuseProtectionMiddleware(
         string operation = context.GetEndpoint()?.Metadata.GetMetadata<IEndpointNameMetadata>()
             ?.EndpointName ?? "Unmatched";
         bool healthEndpoint = operation is "GetLiveness" or "GetReadiness";
-        bool acquired = healthEndpoint
-            ? protection.TryAcquireHealthIngress(
+        bool operatorEndpoint = operation is
+            "GetOperatorStatus"
+            or "RevokeOperatorListing"
+            or "RevokeOperatorPrincipal"
+            or "RevokeOperatorSigningKey"
+            or "BeginOperatorDrain";
+        if (operatorEndpoint
+            && !protection.IsOperatorSourceAllowed(context.Connection.RemoteIpAddress))
+        {
+            bool deniedSourceAdmitted = protection.TryAcquireHttpIngress(
                 context.Connection.RemoteIpAddress,
-                out AbuseProtectionService.AbuseLease? lease,
-                out int retryAfterSeconds)
-            : protection.TryAcquireHttpIngress(
+                "Unmatched",
+                out AbuseProtectionService.AbuseLease? deniedSourceLease,
+                out int deniedRetryAfterSeconds);
+            using (deniedSourceLease)
+            {
+                if (!deniedSourceAdmitted)
+                {
+                    context.Response.Headers.RetryAfter = deniedRetryAfterSeconds.ToString(
+                        System.Globalization.CultureInfo.InvariantCulture);
+                    await WriteErrorAsync(
+                        context,
+                        StatusCodes.Status429TooManyRequests,
+                        RendezvousErrorCode.RateLimited,
+                        "The request rate limit was exceeded.",
+                        deniedRetryAfterSeconds).ConfigureAwait(false);
+                    return;
+                }
+
+                await WriteErrorAsync(
+                    context,
+                    StatusCodes.Status404NotFound,
+                    RendezvousErrorCode.NotFound,
+                    "The requested resource was not found.").ConfigureAwait(false);
+            }
+
+            return;
+        }
+
+        AbuseProtectionService.AbuseLease? lease;
+        int retryAfterSeconds;
+        bool acquired;
+        if (healthEndpoint)
+        {
+            acquired = protection.TryAcquireHealthIngress(
+                context.Connection.RemoteIpAddress,
+                out lease,
+                out retryAfterSeconds);
+        }
+        else if (operatorEndpoint)
+        {
+            acquired = protection.TryAcquireOperatorIngress(
+                context.Connection.RemoteIpAddress,
+                out lease,
+                out retryAfterSeconds);
+        }
+        else
+        {
+            acquired = protection.TryAcquireHttpIngress(
                 context.Connection.RemoteIpAddress,
                 operation,
                 out lease,
                 out retryAfterSeconds);
+        }
+
         if (!acquired)
         {
             context.Response.Headers.RetryAfter = retryAfterSeconds.ToString(