using System.Runtime.CompilerServices;
using Grpc.Core;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.Auth.Abstractions.ApiKeys;
using ZB.MOM.WW.MxGateway.Contracts;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Server.Configuration;
using ZB.MOM.WW.MxGateway.Server.Grpc;
using ZB.MOM.WW.MxGateway.Server.Metrics;
using ZB.MOM.WW.MxGateway.Server.Security.Authentication;
using ZB.MOM.WW.MxGateway.Server.Security.Authorization;
using ZB.MOM.WW.MxGateway.Server.Sessions;
using ZB.MOM.WW.MxGateway.Tests.TestSupport;

// The handler exposes the gateway's constraint-bearing identity; alias the shared library identity
// (returned by the verifier) so the two can be referenced unambiguously.
using ApiKeyIdentity = ZB.MOM.WW.MxGateway.Server.Security.Authentication.ApiKeyIdentity;
using LibApiKeyIdentity = ZB.MOM.WW.Auth.Abstractions.ApiKeys.ApiKeyIdentity;

namespace ZB.MOM.WW.MxGateway.Tests.Security.Authorization;

public sealed class GatewayGrpcAuthorizationInterceptorTests
{
    private const string AttackerPeer = "ipv4:203.0.113.7:5000";
    private const string HolderPeer = "ipv4:198.51.100.4:5000";

    private static readonly TimeSpan FailureWindow = TimeSpan.FromMinutes(1);
    private static readonly TimeSpan ProbeInterval = TimeSpan.FromSeconds(5);

    /// <summary>Verifies that missing API key returns unauthenticated status.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task UnaryServerHandler_MissingApiKey_ReturnsUnauthenticated()
    {
        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
            new FakeApiKeyVerifier(Failure(ApiKeyFailure.MissingOrMalformed)),
            new GatewayRequestIdentityAccessor());

        RpcException exception = await Assert.ThrowsAsync<RpcException>(
            () => interceptor.UnaryServerHandler(
                new OpenSessionRequest(),
                new TestServerCallContext([]),
                (_, _) => Task.FromResult(new OpenSessionReply())));

        Assert.Equal(StatusCode.Unauthenticated, exception.StatusCode);
        Assert.DoesNotContain("secret", exception.Status.Detail, StringComparison.OrdinalIgnoreCase);
    }

    /// <summary>Verifies that invalid API key error does not expose raw credentials.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task UnaryServerHandler_InvalidApiKey_DoesNotExposeRawCredentialInStatus()
    {
        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
            new FakeApiKeyVerifier(Failure(ApiKeyFailure.SecretMismatch)),
            new GatewayRequestIdentityAccessor());

        RpcException exception = await Assert.ThrowsAsync<RpcException>(
            () => interceptor.UnaryServerHandler(
                new OpenSessionRequest(),
                ContextWithAuthorization("Bearer mxgw_operator01_super-secret"),
                (_, _) => Task.FromResult(new OpenSessionReply())));

        Assert.Equal(StatusCode.Unauthenticated, exception.StatusCode);
        Assert.DoesNotContain("super-secret", exception.Status.Detail, StringComparison.Ordinal);
    }

    /// <summary>Verifies that valid key without required scope returns permission denied.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task UnaryServerHandler_ValidApiKeyMissingScope_ReturnsPermissionDenied()
    {
        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
            new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.EventsRead)),
            new GatewayRequestIdentityAccessor());

        RpcException exception = await Assert.ThrowsAsync<RpcException>(
            () => interceptor.UnaryServerHandler(
                new OpenSessionRequest(),
                ContextWithAuthorization("Bearer mxgw_operator01_secret"),
                (_, _) => Task.FromResult(new OpenSessionReply())));

        Assert.Equal(StatusCode.PermissionDenied, exception.StatusCode);
        Assert.Contains(GatewayScopes.SessionOpen, exception.Status.Detail, StringComparison.Ordinal);
    }

    /// <summary>Verifies that valid key with scope sets request identity for the handler.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task UnaryServerHandler_ValidApiKeyWithScope_SetsRequestIdentity()
    {
        GatewayRequestIdentityAccessor identityAccessor = new();
        ApiKeyIdentity? identitySeenByHandler = null;
        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
            new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.SessionOpen)),
            identityAccessor);

        OpenSessionReply reply = await interceptor.UnaryServerHandler(
            new OpenSessionRequest(),
            ContextWithAuthorization("Bearer mxgw_operator01_secret"),
            (_, _) =>
            {
                identitySeenByHandler = identityAccessor.Current;

                return Task.FromResult(new OpenSessionReply { SessionId = "session-1" });
            });

        Assert.Equal("session-1", reply.SessionId);
        Assert.NotNull(identitySeenByHandler);
        Assert.Equal("operator01", identitySeenByHandler.KeyId);
        Assert.Null(identityAccessor.Current);
    }

    /// <summary>Verifies that server stream handler requires proper scope.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ServerStreamingServerHandler_ValidApiKeyMissingScope_ReturnsPermissionDenied()
    {
        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
            new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.SessionOpen)),
            new GatewayRequestIdentityAccessor());

        RpcException exception = await Assert.ThrowsAsync<RpcException>(
            () => interceptor.ServerStreamingServerHandler(
                new StreamEventsRequest(),
                new RecordingServerStreamWriter<MxEvent>(),
                ContextWithAuthorization("Bearer mxgw_operator01_secret"),
                (_, _, _) => Task.CompletedTask));

        Assert.Equal(StatusCode.PermissionDenied, exception.StatusCode);
        Assert.Contains(GatewayScopes.EventsRead, exception.Status.Detail, StringComparison.Ordinal);
    }

    /// <summary>Verifies that server stream handler allows streams with proper scope.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ServerStreamingServerHandler_ValidApiKeyWithScope_AllowsStream()
    {
        GatewayRequestIdentityAccessor identityAccessor = new();
        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
            new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.EventsRead)),
            identityAccessor);
        RecordingServerStreamWriter<MxEvent> streamWriter = new();

        await interceptor.ServerStreamingServerHandler(
            new StreamEventsRequest(),
            streamWriter,
            ContextWithAuthorization("Bearer mxgw_operator01_secret"),
            async (_, writer, _) =>
            {
                Assert.Equal("operator01", identityAccessor.Current?.KeyId);
                await writer.WriteAsync(new MxEvent { SessionId = "session-1" });
            });

        MxEvent eventMessage = Assert.Single(streamWriter.Messages);
        Assert.Equal("session-1", eventMessage.SessionId);
        Assert.Null(identityAccessor.Current);
    }

    /// <summary>Verifies that disabled authentication skips API key verification.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task UnaryServerHandler_AuthenticationDisabled_SkipsApiKeyVerification()
    {
        GatewayRequestIdentityAccessor identityAccessor = new();
        FakeApiKeyVerifier verifier = new(Failure(ApiKeyFailure.MissingOrMalformed));
        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
            verifier,
            identityAccessor,
            AuthenticationMode.Disabled);

        OpenSessionReply reply = await interceptor.UnaryServerHandler(
            new OpenSessionRequest(),
            new TestServerCallContext([]),
            (_, _) => Task.FromResult(new OpenSessionReply { SessionId = "session-1" }));

        Assert.Equal("session-1", reply.SessionId);
        Assert.False(verifier.WasCalled);
        Assert.Null(identityAccessor.Current);
    }

    /// <summary>
    /// End-to-end composition test: runs an <c>OpenSession</c> call through the real
    /// interceptor in front of the real <see cref="MxAccessGatewayService"/> with a key
    /// that lacks the <c>session:open</c> scope, and asserts the interceptor denies the
    /// call with <see cref="StatusCode.PermissionDenied"/> before the service runs.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task InterceptorComposedWithService_OpenSessionMissingScope_DeniesBeforeServiceRuns()
    {
        GatewayRequestIdentityAccessor identityAccessor = new();
        RecordingSessionManager sessionManager = new();
        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
            new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.EventsRead)),
            identityAccessor);
        MxAccessGatewayService service = CreateService(sessionManager, identityAccessor);

        RpcException exception = await Assert.ThrowsAsync<RpcException>(
            () => interceptor.UnaryServerHandler(
                new OpenSessionRequest { ClientSessionName = "operator-session" },
                ContextWithAuthorization("Bearer mxgw_operator01_secret"),
                (request, context) => service.OpenSession(request, context)));

        Assert.Equal(StatusCode.PermissionDenied, exception.StatusCode);
        Assert.Contains(GatewayScopes.SessionOpen, exception.Status.Detail, StringComparison.Ordinal);
        Assert.Equal(0, sessionManager.OpenSessionCount);
    }

    /// <summary>
    /// End-to-end composition test: runs an <c>OpenSession</c> call through the real
    /// interceptor in front of the real <see cref="MxAccessGatewayService"/> with a key
    /// that holds <c>session:open</c>, and asserts the service runs and observes the
    /// interceptor-supplied identity.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task InterceptorComposedWithService_OpenSessionWithScope_RunsServiceWithIdentity()
    {
        GatewayRequestIdentityAccessor identityAccessor = new();
        RecordingSessionManager sessionManager = new();
        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
            new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.SessionOpen)),
            identityAccessor);
        MxAccessGatewayService service = CreateService(sessionManager, identityAccessor);

        OpenSessionReply reply = await interceptor.UnaryServerHandler(
            new OpenSessionRequest { ClientSessionName = "operator-session" },
            ContextWithAuthorization("Bearer mxgw_operator01_secret"),
            (request, context) => service.OpenSession(request, context));

        Assert.Equal("session-1", reply.SessionId);
        Assert.Equal(1, sessionManager.OpenSessionCount);
        Assert.Equal("Operator Key", sessionManager.LastClientIdentity);
    }

    /// <summary>
    /// End-to-end composition test: an <c>Invoke</c> call through the real interceptor in
    /// front of the real service with a key holding only <c>invoke:read</c> is denied
    /// because the wrapped command is a write, confirming command-scope mapping is
    /// enforced through the full composition.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task InterceptorComposedWithService_InvokeWriteCommandWithReadScope_DeniesBeforeServiceRuns()
    {
        GatewayRequestIdentityAccessor identityAccessor = new();
        RecordingSessionManager sessionManager = new();
        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
            new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.InvokeRead)),
            identityAccessor);
        MxAccessGatewayService service = CreateService(sessionManager, identityAccessor);
        MxCommandRequest request = new()
        {
            SessionId = "session-1",
            Command = new MxCommand
            {
                Kind = MxCommandKind.Write,
                Write = new WriteCommand { ServerHandle = 1, ItemHandle = 2 },
            },
        };

        RpcException exception = await Assert.ThrowsAsync<RpcException>(
            () => interceptor.UnaryServerHandler(
                request,
                ContextWithAuthorization("Bearer mxgw_operator01_secret"),
                (req, context) => service.Invoke(req, context)));

        Assert.Equal(StatusCode.PermissionDenied, exception.StatusCode);
        Assert.Contains(GatewayScopes.InvokeWrite, exception.Status.Detail, StringComparison.Ordinal);
        Assert.Equal(0, sessionManager.InvokeCount);
    }

    /// <summary>
    /// Verifies the interceptor denies <c>AcknowledgeAlarm</c> calls that lack
    /// <see cref="GatewayScopes.InvokeWrite"/>. Ack is a write-shaped mutation against
    /// alarm state, so it carries the same scope as <c>MxCommandKind.Write</c>.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task UnaryServerHandler_AcknowledgeAlarmMissingScope_ReturnsPermissionDenied()
    {
        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
            new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.InvokeRead)),
            new GatewayRequestIdentityAccessor());

        RpcException exception = await Assert.ThrowsAsync<RpcException>(
            () => interceptor.UnaryServerHandler(
                new AcknowledgeAlarmRequest { AlarmFullReference = "ref" },
                ContextWithAuthorization("Bearer mxgw_operator01_secret"),
                (_, _) => Task.FromResult(new AcknowledgeAlarmReply())));

        Assert.Equal(StatusCode.PermissionDenied, exception.StatusCode);
        Assert.Contains(GatewayScopes.InvokeWrite, exception.Status.Detail, StringComparison.Ordinal);
    }

    /// <summary>Verifies that an API key holding <c>invoke:write</c> may call <c>AcknowledgeAlarm</c>.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task UnaryServerHandler_AcknowledgeAlarmWithScope_RunsHandler()
    {
        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
            new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.InvokeWrite)),
            new GatewayRequestIdentityAccessor());
        bool handlerRan = false;

        AcknowledgeAlarmReply reply = await interceptor.UnaryServerHandler(
            new AcknowledgeAlarmRequest { AlarmFullReference = "ref" },
            ContextWithAuthorization("Bearer mxgw_operator01_secret"),
            (_, _) =>
            {
                handlerRan = true;
                return Task.FromResult(new AcknowledgeAlarmReply());
            });

        Assert.NotNull(reply);
        Assert.True(handlerRan);
    }

    /// <summary>
    /// Verifies the interceptor denies <c>QueryActiveAlarms</c> server-streaming calls that
    /// lack <see cref="GatewayScopes.EventsRead"/>. Active-alarm snapshots are part of the
    /// alarm/event surface and share the same scope as <c>StreamEvents</c>.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ServerStreamingServerHandler_QueryActiveAlarmsMissingScope_ReturnsPermissionDenied()
    {
        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
            new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.InvokeRead)),
            new GatewayRequestIdentityAccessor());

        RpcException exception = await Assert.ThrowsAsync<RpcException>(
            () => interceptor.ServerStreamingServerHandler(
                new QueryActiveAlarmsRequest(),
                new RecordingServerStreamWriter<ActiveAlarmSnapshot>(),
                ContextWithAuthorization("Bearer mxgw_operator01_secret"),
                (_, _, _) => Task.CompletedTask));

        Assert.Equal(StatusCode.PermissionDenied, exception.StatusCode);
        Assert.Contains(GatewayScopes.EventsRead, exception.Status.Detail, StringComparison.Ordinal);
    }

    /// <summary>Verifies that an API key holding <c>events:read</c> may call <c>QueryActiveAlarms</c>.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ServerStreamingServerHandler_QueryActiveAlarmsWithScope_RunsHandler()
    {
        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
            new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.EventsRead)),
            new GatewayRequestIdentityAccessor());
        RecordingServerStreamWriter<ActiveAlarmSnapshot> streamWriter = new();

        await interceptor.ServerStreamingServerHandler(
            new QueryActiveAlarmsRequest(),
            streamWriter,
            ContextWithAuthorization("Bearer mxgw_operator01_secret"),
            async (_, writer, _) =>
            {
                await writer.WriteAsync(new ActiveAlarmSnapshot());
            });

        Assert.Single(streamWriter.Messages);
    }

    /// <summary>
    /// SEC-31: once an attacking peer has exceeded the failure limit for a key id, the interceptor
    /// short-circuits with <see cref="StatusCode.ResourceExhausted"/> BEFORE calling the verifier, so
    /// an online guessing loop stops spending a store read per attempt. The composite
    /// <c>(peer, key id)</c> partition keeps that bound per attacking address.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task BruteForceBound_StillEnforcedPerAttackingPeer()
    {
        CountingFailureVerifier verifier = new(Failure(ApiKeyFailure.SecretMismatch));
        ManualTimeProvider clock = new(DateTimeOffset.UnixEpoch);
        ApiKeyFailureLimiter limiter = CreateLimiter(clock, limit: 3);
        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
            verifier,
            new GatewayRequestIdentityAccessor(),
            failureLimiter: limiter);

        // The first three attempts reach the verifier and fail (recording a failure each time).
        for (int attempt = 0; attempt < 3; attempt++)
        {
            RpcException failure = await Assert.ThrowsAsync<RpcException>(
                () => interceptor.UnaryServerHandler(
                    new OpenSessionRequest(),
                    ContextWithAuthorization("Bearer mxgw_operator01_bad-secret", AttackerPeer),
                    (_, _) => Task.FromResult(new OpenSessionReply())));
            Assert.Equal(StatusCode.Unauthenticated, failure.StatusCode);
        }

        Assert.Equal(3, verifier.CallCount);

        // The fourth attempt is short-circuited: ResourceExhausted, and the verifier is NOT called.
        RpcException throttled = await Assert.ThrowsAsync<RpcException>(
            () => interceptor.UnaryServerHandler(
                new OpenSessionRequest(),
                ContextWithAuthorization("Bearer mxgw_operator01_bad-secret", AttackerPeer),
                (_, _) => Task.FromResult(new OpenSessionReply())));

        Assert.Equal(StatusCode.ResourceExhausted, throttled.StatusCode);
        Assert.Equal(3, verifier.CallCount);
    }

    /// <summary>
    /// SEC-31 (the lockout inversion): an attacker who floods failures for a victim's key id from its
    /// own address must not deny that key to the legitimate holder. The holder presents the correct
    /// secret from a different transport peer and authenticates on the first attempt — the verifier
    /// is reached and the RPC succeeds.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task AttackerSpamOnVictimKeyId_FromDifferentPeer_DoesNotBlockLegitimateHolderPresentingCorrectSecret()
    {
        ManualTimeProvider clock = new(DateTimeOffset.UnixEpoch);
        ApiKeyFailureLimiter limiter = CreateLimiter(clock, limit: 3, aggregateLimit: 30);
        GatewayGrpcAuthorizationInterceptor attacked = CreateInterceptor(
            new CountingFailureVerifier(Failure(ApiKeyFailure.SecretMismatch)),
            new GatewayRequestIdentityAccessor(),
            failureLimiter: limiter);

        // Flood the victim's key id from the attacker's address until that partition is throttled.
        StatusCode lastAttackerStatus = StatusCode.OK;
        for (int attempt = 0; attempt < 6; attempt++)
        {
            RpcException failure = await Assert.ThrowsAsync<RpcException>(
                () => attacked.UnaryServerHandler(
                    new OpenSessionRequest(),
                    ContextWithAuthorization("Bearer mxgw_operator01_guess", AttackerPeer),
                    (_, _) => Task.FromResult(new OpenSessionReply())));
            lastAttackerStatus = failure.StatusCode;
        }

        Assert.Equal(StatusCode.ResourceExhausted, lastAttackerStatus);

        // The legitimate holder, on a different address, is verified and admitted immediately.
        FakeApiKeyVerifier holderVerifier = new(SuccessWithScopes(GatewayScopes.SessionOpen));
        GatewayGrpcAuthorizationInterceptor holder = CreateInterceptor(
            holderVerifier,
            new GatewayRequestIdentityAccessor(),
            failureLimiter: limiter);

        OpenSessionReply reply = await holder.UnaryServerHandler(
            new OpenSessionRequest(),
            ContextWithAuthorization("Bearer mxgw_operator01_correct", HolderPeer),
            (_, _) => Task.FromResult(new OpenSessionReply { SessionId = "session-1" }));

        Assert.True(holderVerifier.WasCalled);
        Assert.Equal("session-1", reply.SessionId);
    }

    /// <summary>
    /// SEC-31 layer 2: failures for one key id sprayed across more distinct peers than
    /// <c>ApiKeyFailureAggregateLimit</c> put that key id into probe mode globally, so a
    /// rotating-source attacker gets at most one verifier call per probe interval.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task AggregateSpray_AcrossManyPeers_TripsPerKeyProbeMode()
    {
        CountingFailureVerifier verifier = new(Failure(ApiKeyFailure.SecretMismatch));
        ManualTimeProvider clock = new(DateTimeOffset.UnixEpoch);
        ApiKeyFailureLimiter limiter = CreateLimiter(clock, limit: 100, aggregateLimit: 5);
        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
            verifier,
            new GatewayRequestIdentityAccessor(),
            failureLimiter: limiter);

        for (int peer = 0; peer < 5; peer++)
        {
            await Assert.ThrowsAsync<RpcException>(
                () => interceptor.UnaryServerHandler(
                    new OpenSessionRequest(),
                    ContextWithAuthorization("Bearer mxgw_operator01_guess", $"ipv4:10.9.0.{peer}:5000"),
                    (_, _) => Task.FromResult(new OpenSessionReply())));
        }

        Assert.Equal(5, verifier.CallCount);

        // A never-seen peer is now probe-limited: no composite failures of its own, but the key id's
        // aggregate is tripped, so the request never reaches the verifier.
        RpcException throttled = await Assert.ThrowsAsync<RpcException>(
            () => interceptor.UnaryServerHandler(
                new OpenSessionRequest(),
                ContextWithAuthorization("Bearer mxgw_operator01_guess", "ipv4:10.9.1.1:5000"),
                (_, _) => Task.FromResult(new OpenSessionReply())));

        Assert.Equal(StatusCode.ResourceExhausted, throttled.StatusCode);
        Assert.Equal(5, verifier.CallCount);

        // One probe slot opens per interval, and it is consumed by the first arrival.
        clock.Advance(ProbeInterval);

        RpcException probed = await Assert.ThrowsAsync<RpcException>(
            () => interceptor.UnaryServerHandler(
                new OpenSessionRequest(),
                ContextWithAuthorization("Bearer mxgw_operator01_guess", "ipv4:10.9.1.2:5000"),
                (_, _) => Task.FromResult(new OpenSessionReply())));

        Assert.Equal(StatusCode.Unauthenticated, probed.StatusCode);
        Assert.Equal(6, verifier.CallCount);

        RpcException throttledAgain = await Assert.ThrowsAsync<RpcException>(
            () => interceptor.UnaryServerHandler(
                new OpenSessionRequest(),
                ContextWithAuthorization("Bearer mxgw_operator01_guess", "ipv4:10.9.1.3:5000"),
                (_, _) => Task.FromResult(new OpenSessionReply())));

        Assert.Equal(StatusCode.ResourceExhausted, throttledAgain.StatusCode);
        Assert.Equal(6, verifier.CallCount);
    }

    /// <summary>
    /// SEC-31: the success-reset path stays reachable while throttled. A throttled partition admits
    /// one probe per interval; the correct secret rides that slot, authenticates, and fully clears
    /// both limiter layers, so the next wrong attempt is <see cref="StatusCode.Unauthenticated"/>
    /// rather than <see cref="StatusCode.ResourceExhausted"/>.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task CorrectSecret_DuringProbeMode_AuthenticatesViaProbeSlotAndResets()
    {
        ManualTimeProvider clock = new(DateTimeOffset.UnixEpoch);
        ApiKeyFailureLimiter limiter = CreateLimiter(clock, limit: 3, aggregateLimit: 30);
        GatewayGrpcAuthorizationInterceptor failing = CreateInterceptor(
            new FakeApiKeyVerifier(Failure(ApiKeyFailure.SecretMismatch)),
            new GatewayRequestIdentityAccessor(),
            failureLimiter: limiter);
        FakeApiKeyVerifier holderVerifier = new(SuccessWithScopes(GatewayScopes.SessionOpen));
        GatewayGrpcAuthorizationInterceptor succeeding = CreateInterceptor(
            holderVerifier,
            new GatewayRequestIdentityAccessor(),
            failureLimiter: limiter);

        for (int attempt = 0; attempt < 3; attempt++)
        {
            await Assert.ThrowsAsync<RpcException>(
                () => failing.UnaryServerHandler(
                    new OpenSessionRequest(),
                    ContextWithAuthorization("Bearer mxgw_operator01_bad", HolderPeer),
                    (_, _) => Task.FromResult(new OpenSessionReply())));
        }

        RpcException throttled = await Assert.ThrowsAsync<RpcException>(
            () => failing.UnaryServerHandler(
                new OpenSessionRequest(),
                ContextWithAuthorization("Bearer mxgw_operator01_bad", HolderPeer),
                (_, _) => Task.FromResult(new OpenSessionReply())));
        Assert.Equal(StatusCode.ResourceExhausted, throttled.StatusCode);

        clock.Advance(ProbeInterval);

        OpenSessionReply reply = await succeeding.UnaryServerHandler(
            new OpenSessionRequest(),
            ContextWithAuthorization("Bearer mxgw_operator01_correct", HolderPeer),
            (_, _) => Task.FromResult(new OpenSessionReply { SessionId = "session-1" }));

        Assert.True(holderVerifier.WasCalled);
        Assert.Equal("session-1", reply.SessionId);

        RpcException afterReset = await Assert.ThrowsAsync<RpcException>(
            () => failing.UnaryServerHandler(
                new OpenSessionRequest(),
                ContextWithAuthorization("Bearer mxgw_operator01_bad", HolderPeer),
                (_, _) => Task.FromResult(new OpenSessionReply())));

        Assert.Equal(StatusCode.Unauthenticated, afterReset.StatusCode);
    }

    /// <summary>
    /// SEC-32: only a validly shaped <c>mxgw_&lt;keyId&gt;_&lt;secret&gt;</c> token mints a key-id
    /// partition. Junk tokens of varied shapes all collapse onto the sender's transport-peer fallback
    /// partition, so a spray cannot mint one tracked entry per invented token.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task NonMxgwToken_FallsBackToTransportPeerPartition()
    {
        ManualTimeProvider clock = new(DateTimeOffset.UnixEpoch);
        ApiKeyFailureLimiter limiter = CreateLimiter(clock, limit: 1000);
        GatewayGrpcAuthorizationInterceptor interceptor = CreateInterceptor(
            new FakeApiKeyVerifier(Failure(ApiKeyFailure.MissingOrMalformed)),
            new GatewayRequestIdentityAccessor(),
            failureLimiter: limiter);

        string[] junkTokens =
        [
            "Bearer garbage",
            "Bearer a_b_c",
            "Bearer notmxgw_operator01_secret",
            "Bearer MXGW_operator01_secret",
            "Bearer mxgw__secret",
            "Bearer mxgw_operator01_",
            "Bearer mxgw_" + new string('k', 65) + "_secret",
            "Bearer mxgw_onlytwo",
        ];

        foreach (string token in junkTokens)
        {
            await Assert.ThrowsAsync<RpcException>(
                () => interceptor.UnaryServerHandler(
                    new OpenSessionRequest(),
                    ContextWithAuthorization(token, AttackerPeer),
                    (_, _) => Task.FromResult(new OpenSessionReply())));
        }

        Assert.Equal(1, limiter.TrackedPartitionCount);
        Assert.True(limiter.IsTracked(new ApiKeyThrottlePartition(AttackerPeer, KeyId: null)));
    }

    /// <summary>
    /// A successful verification resets the peer's failure counter, so accumulated failures
    /// from a fat-fingered secret do not lock out a client that subsequently authenticates.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task UnaryServerHandler_SuccessResetsFailureCounter()
    {
        ApiKeyFailureLimiter limiter = CreateLimiter(new ManualTimeProvider(DateTimeOffset.UnixEpoch), limit: 3);

        // Two failures against the same key id, then a success (which resets), then two more
        // failures — without the reset the fifth attempt would be blocked at the limit of 3.
        GatewayGrpcAuthorizationInterceptor failing = CreateInterceptor(
            new FakeApiKeyVerifier(Failure(ApiKeyFailure.SecretMismatch)),
            new GatewayRequestIdentityAccessor(),
            failureLimiter: limiter);
        GatewayGrpcAuthorizationInterceptor succeeding = CreateInterceptor(
            new FakeApiKeyVerifier(SuccessWithScopes(GatewayScopes.SessionOpen)),
            new GatewayRequestIdentityAccessor(),
            failureLimiter: limiter);

        for (int i = 0; i < 2; i++)
        {
            await Assert.ThrowsAsync<RpcException>(
                () => failing.UnaryServerHandler(
                    new OpenSessionRequest(),
                    ContextWithAuthorization("Bearer mxgw_operator01_bad"),
                    (_, _) => Task.FromResult(new OpenSessionReply())));
        }

        await succeeding.UnaryServerHandler(
            new OpenSessionRequest(),
            ContextWithAuthorization("Bearer mxgw_operator01_good"),
            (_, _) => Task.FromResult(new OpenSessionReply { SessionId = "s" }));

        // Post-reset: two more failures still map to Unauthenticated (not ResourceExhausted).
        for (int i = 0; i < 2; i++)
        {
            RpcException ex = await Assert.ThrowsAsync<RpcException>(
                () => failing.UnaryServerHandler(
                    new OpenSessionRequest(),
                    ContextWithAuthorization("Bearer mxgw_operator01_bad"),
                    (_, _) => Task.FromResult(new OpenSessionReply())));
            Assert.Equal(StatusCode.Unauthenticated, ex.StatusCode);
        }
    }

    private static MxAccessGatewayService CreateService(
        ISessionManager sessionManager,
        IGatewayRequestIdentityAccessor identityAccessor)
    {
        return new MxAccessGatewayService(
            sessionManager,
            identityAccessor,
            new AllowAllConstraintEnforcer(),
            new MxAccessGrpcRequestValidator(),
            new MxAccessGrpcMapper(),
            new NoOpEventStreamService(),
            new GatewayMetrics(),
            NullLogger<MxAccessGatewayService>.Instance,
            new FakeGatewayAlarmService());
    }

    private static GatewayGrpcAuthorizationInterceptor CreateInterceptor(
        IApiKeyVerifier apiKeyVerifier,
        IGatewayRequestIdentityAccessor identityAccessor,
        AuthenticationMode authenticationMode = AuthenticationMode.ApiKey,
        ApiKeyFailureLimiter? failureLimiter = null)
    {
        return new GatewayGrpcAuthorizationInterceptor(
            apiKeyVerifier,
            new GatewayGrpcScopeResolver(),
            identityAccessor,
            Options.Create(new GatewayOptions
            {
                Authentication = new AuthenticationOptions
                {
                    Mode = authenticationMode
                }
            }),
            failureLimiter ?? CreateLimiter(TimeProvider.System, limit: 1000),
            new GatewayMetrics());
    }

    private static ApiKeyFailureLimiter CreateLimiter(
        TimeProvider clock,
        int limit,
        int aggregateLimit = 0,
        int maxPartitions = 1024)
    {
        return new ApiKeyFailureLimiter(
            limit,
            FailureWindow,
            maxPartitions,
            aggregateLimit,
            ProbeInterval,
            clock);
    }

    private static ApiKeyVerification SuccessWithScopes(params string[] scopes)
    {
        return new ApiKeyVerification(
            Succeeded: true,
            Identity: new LibApiKeyIdentity(
                KeyId: "operator01",
                DisplayName: "Operator Key",
                Scopes: new HashSet<string>(scopes, StringComparer.Ordinal),
                Constraints: null),
            Failure: null);
    }

    private static ApiKeyVerification Failure(ApiKeyFailure failure)
    {
        return new ApiKeyVerification(Succeeded: false, Identity: null, Failure: failure);
    }

    private static TestServerCallContext ContextWithAuthorization(string authorizationHeader, string? peer = null)
    {
        return new TestServerCallContext(
            [new Metadata.Entry("authorization", authorizationHeader)],
            peer: peer);
    }

    /// <summary>Records whether the gateway service ran past the interceptor for composition tests.</summary>
    private sealed class RecordingSessionManager : ISessionManager
    {
        /// <summary>Gets the number of times OpenSessionAsync was invoked.</summary>
        public int OpenSessionCount { get; private set; }

        /// <summary>Gets the number of times InvokeAsync was invoked.</summary>
        public int InvokeCount { get; private set; }

        /// <summary>Gets the last client identity passed to OpenSessionAsync.</summary>
        public string? LastClientIdentity { get; private set; }

        /// <inheritdoc />
        public Task<GatewaySession> OpenSessionAsync(
            SessionOpenRequest request,
            string? clientIdentity,
            string? ownerKeyId,
            CancellationToken cancellationToken)
        {
            OpenSessionCount++;
            LastClientIdentity = clientIdentity;

            GatewaySession session = new(
                "session-1",
                GatewayContractInfo.DefaultBackendName,
                "pipe",
                "nonce",
                clientIdentity ?? "client",
                "client-session",
                "client-correlation",
                TimeSpan.FromSeconds(7),
                TimeSpan.FromSeconds(30),
                TimeSpan.FromSeconds(10),
                DateTimeOffset.UtcNow);

            return Task.FromResult(session);
        }

        /// <inheritdoc />
        public bool TryGetSession(string sessionId, out GatewaySession session)
        {
            session = null!;
            return false;
        }

        /// <inheritdoc />
        public Task<WorkerCommandReply> InvokeAsync(
            string sessionId,
            WorkerCommand command,
            CancellationToken cancellationToken)
        {
            InvokeCount++;
            return Task.FromResult(new WorkerCommandReply());
        }

        /// <inheritdoc />
        public IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
            string sessionId,
            CancellationToken cancellationToken)
        {
            return AsyncEnumerable.Empty<WorkerEvent>();
        }

        /// <inheritdoc />
        public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
            string sessionId,
            CancellationToken cancellationToken)
        {
            return AsyncEnumerable.Empty<MxEvent>();
        }

        /// <inheritdoc />
        public Task<SessionCloseResult> CloseSessionAsync(
            string sessionId,
            CancellationToken cancellationToken)
        {
            return Task.FromResult(new SessionCloseResult(sessionId, SessionState.Closed, AlreadyClosed: false));
        }

        /// <inheritdoc />
        public Task<SessionCloseResult> KillWorkerAsync(
            string sessionId,
            string reason,
            CancellationToken cancellationToken)
        {
            return Task.FromResult(new SessionCloseResult(sessionId, SessionState.Closed, AlreadyClosed: false));
        }

        /// <inheritdoc />
        public Task<int> CloseExpiredLeasesAsync(
            DateTimeOffset now,
            CancellationToken cancellationToken)
        {
            return Task.FromResult(0);
        }

        /// <inheritdoc />
        public Task ShutdownAsync(CancellationToken cancellationToken)
        {
            return Task.CompletedTask;
        }
    }

    /// <summary>Event stream service that yields nothing; alarm/event RPCs are not under test here.</summary>
    private sealed class NoOpEventStreamService : IEventStreamService
    {
        /// <inheritdoc />
        public async IAsyncEnumerable<MxEvent> StreamEventsAsync(
            StreamEventsRequest request,
            string? callerKeyId,
            [EnumeratorCancellation] CancellationToken cancellationToken)
        {
            await Task.CompletedTask;
            yield break;
        }
    }

    private sealed class CountingFailureVerifier(ApiKeyVerification result) : IApiKeyVerifier
    {
        /// <summary>Gets the number of times the verifier was invoked.</summary>
        public int CallCount { get; private set; }

        /// <summary>Returns the configured result and counts the invocation.</summary>
        /// <param name="authorizationHeader">The authorization header to verify.</param>
        /// <param name="ct">Cancellation token.</param>
        /// <returns>The configured verification result.</returns>
        public Task<ApiKeyVerification> VerifyAsync(string authorizationHeader, CancellationToken ct)
        {
            CallCount++;
            return Task.FromResult(result);
        }
    }

    private sealed class FakeApiKeyVerifier(ApiKeyVerification result) : IApiKeyVerifier
    {
        /// <summary>Gets whether the verifier was called.</summary>
        public bool WasCalled { get; private set; }

        /// <summary>Gets the last authorization header seen by the verifier.</summary>
        public string? LastAuthorizationHeader { get; private set; }

        /// <summary>Verifies the authorization header against stored result.</summary>
        /// <param name="authorizationHeader">The authorization header to verify.</param>
        /// <param name="ct">Cancellation token.</param>
        /// <returns>Configured verification result.</returns>
        public Task<ApiKeyVerification> VerifyAsync(
            string authorizationHeader,
            CancellationToken ct)
        {
            WasCalled = true;
            LastAuthorizationHeader = authorizationHeader;

            return Task.FromResult(result);
        }
    }

}