Rendezvous/tests/FinalFactory.Rendezvous.Tests/Server/NatMediationProcessorTests.cs

using System.Collections.Concurrent;
using System.Net;
using FinalFactory.Rendezvous.Contracts;
using FinalFactory.Rendezvous.Server.JoinAttempts;
using FinalFactory.Rendezvous.Server.State;
using FinalFactory.Rendezvous.Server.Transport;
using FinalFactory.Rendezvous.Tests.JoinAttempts;

namespace FinalFactory.Rendezvous.Tests.Server;

public sealed class NatMediationProcessorTests
{
    [Fact]
    public void AuthenticatedHostPresenceUsesTheObservedGameplaySocket()
    {
        using JoinAttemptFixture fixture = new();
        (RegisterSessionResponse registration, _) = fixture.CreateHost(bindPresence: false);
        NatMediationProcessor processor = CreateProcessor(fixture);
        CaptureIntroductionSink sink = new();
        string token = NatPunchRequestTokenCodec.Encode(
            NatPunchPeerRole.HostPresence,
            registration.HostPresenceHandle,
            registration.HostPresenceCapability);

        Assert.Equal(
            NatMediationResult.HostPresenceAccepted,
            processor.ProcessRequest(
                Endpoint("192.168.1.50", 40_000),
                Endpoint("203.0.113.77", 51_234),
                token,
                sink));
        Assert.Equal(registration.ListingId, Assert.Single(fixture.Sessions.Browse()).Definition.ListingId);
        Assert.Empty(sink.Plans);
    }

    [Fact]
    public void MatchedPeersReceiveOneIntroductionAndSameNatPrivateCandidates()
    {
        using JoinAttemptFixture fixture = new();
        (RegisterSessionResponse registration, _) = fixture.CreateHost();
        AttemptCredentials attempt = CreateAttempt(fixture, registration, "same-nat");
        NatMediationProcessor processor = CreateProcessor(fixture);
        CaptureIntroductionSink sink = new();

        Assert.Equal(
            NatMediationResult.WaitingForPeer,
            Process(processor, sink, attempt, NatPunchPeerRole.Host,
                Endpoint("192.168.1.10", 41_000), Endpoint("203.0.113.20", 51_000)));
        Assert.Equal(
            NatMediationResult.Introduced,
            Process(processor, sink, attempt, NatPunchPeerRole.Client,
                Endpoint("192.168.1.11", 42_000), Endpoint("203.0.113.20", 52_000)));

        NatIntroductionPlan plan = Assert.Single(sink.Plans);
        Assert.Equal(Endpoint("192.168.1.10", 41_000), plan.HostLocal);
        Assert.Equal(Endpoint("192.168.1.11", 42_000), plan.ClientLocal);
        Assert.Equal(Endpoint("203.0.113.20", 51_000), plan.HostPublic);
        Assert.Equal(Endpoint("203.0.113.20", 52_000), plan.ClientPublic);
        Assert.Equal(43, plan.ConnectionTicket.Length);
        Assert.DoesNotContain(plan.ConnectionTicket, plan.ToString(), StringComparison.Ordinal);

        Assert.True(fixture.Sessions.Capabilities.TryFingerprint(
            plan.ConnectionTicket,
            out SecretFingerprint ticketFingerprint));
        Assert.True(fixture.Sessions.Store.ConsumeConnectionTicket(new(
            attempt.AttemptId,
            ticketFingerprint)).Succeeded);

        Assert.Equal(
            NatMediationResult.Duplicate,
            Process(processor, sink, attempt, NatPunchPeerRole.Client,
                Endpoint("192.168.1.11", 42_000), Endpoint("203.0.113.20", 52_000)));
        Assert.Single(sink.Plans);
    }

    [Fact]
    public void DifferentNatsAndInvalidLocalClaimsExposeOnlyObservedPublicEndpoints()
    {
        using JoinAttemptFixture fixture = new();
        (RegisterSessionResponse registration, _) = fixture.CreateHost();
        AttemptCredentials attempt = CreateAttempt(fixture, registration, "different-nats");
        NatMediationProcessor processor = CreateProcessor(fixture);
        CaptureIntroductionSink sink = new();

        _ = Process(processor, sink, attempt, NatPunchPeerRole.Client,
            Endpoint("8.8.8.8", 42_000), Endpoint("198.51.100.40", 52_000));
        Assert.Equal(
            NatMediationResult.Introduced,
            Process(processor, sink, attempt, NatPunchPeerRole.Host,
                Endpoint("192.168.1.10", 41_000), Endpoint("203.0.113.20", 51_000)));

        NatIntroductionPlan plan = Assert.Single(sink.Plans);
        Assert.Equal(plan.HostPublic, plan.HostLocal);
        Assert.Equal(plan.ClientPublic, plan.ClientLocal);
        Assert.NotEqual(IPAddress.Parse("8.8.8.8"), plan.ClientLocal.Address);
    }

    [Fact]
    public void RoleAndEndpointSubstitutionAreRejectedWithoutChangingTheFirstBinding()
    {
        using JoinAttemptFixture fixture = new();
        (RegisterSessionResponse registration, _) = fixture.CreateHost();
        AttemptCredentials attempt = CreateAttempt(fixture, registration, "substitution");
        NatMediationProcessor processor = CreateProcessor(fixture);
        CaptureIntroductionSink sink = new();

        string crossRole = NatPunchRequestTokenCodec.Encode(
            NatPunchPeerRole.Host,
            attempt.Handle,
            attempt.ClientCapability);
        Assert.Equal(
            NatMediationResult.Dropped,
            processor.ProcessRequest(
                Endpoint("192.168.1.10", 41_000),
                Endpoint("203.0.113.20", 51_000),
                crossRole,
                sink));

        Assert.Equal(
            NatMediationResult.WaitingForPeer,
            Process(processor, sink, attempt, NatPunchPeerRole.Host,
                Endpoint("192.168.1.10", 41_000), Endpoint("203.0.113.20", 51_000)));
        Assert.Equal(
            NatMediationResult.Rejected,
            Process(processor, sink, attempt, NatPunchPeerRole.Host,
                Endpoint("192.168.1.99", 41_999), Endpoint("203.0.113.99", 51_999)));
        Assert.Equal(
            NatMediationResult.Introduced,
            Process(processor, sink, attempt, NatPunchPeerRole.Client,
                Endpoint("192.168.2.10", 42_000), Endpoint("198.51.100.40", 52_000)));

        Assert.Equal(Endpoint("203.0.113.20", 51_000), Assert.Single(sink.Plans).HostPublic);
    }

    [Fact]
    public void ConcurrentAttemptsForOneSessionNeverCrossWire()
    {
        using JoinAttemptFixture fixture = new();
        (RegisterSessionResponse registration, _) = fixture.CreateHost();
        AttemptCredentials first = CreateAttempt(fixture, registration, "parallel-1");
        AttemptCredentials second = CreateAttempt(fixture, registration, "parallel-2");
        NatMediationProcessor processor = CreateProcessor(fixture);
        CaptureIntroductionSink sink = new();

        _ = Process(processor, sink, first, NatPunchPeerRole.Host,
            Endpoint("10.0.0.10", 41_001), Endpoint("203.0.113.10", 51_001));
        _ = Process(processor, sink, second, NatPunchPeerRole.Host,
            Endpoint("10.0.0.20", 41_002), Endpoint("203.0.113.20", 51_002));
        _ = Process(processor, sink, second, NatPunchPeerRole.Client,
            Endpoint("10.0.0.21", 42_002), Endpoint("198.51.100.20", 52_002));
        _ = Process(processor, sink, first, NatPunchPeerRole.Client,
            Endpoint("10.0.0.11", 42_001), Endpoint("198.51.100.10", 52_001));

        Assert.Equal(2, sink.Plans.Count);
        Assert.Contains(sink.Plans, plan =>
            plan.HostPublic.Equals(Endpoint("203.0.113.10", 51_001))
            && plan.ClientPublic.Equals(Endpoint("198.51.100.10", 52_001)));
        Assert.Contains(sink.Plans, plan =>
            plan.HostPublic.Equals(Endpoint("203.0.113.20", 51_002))
            && plan.ClientPublic.Equals(Endpoint("198.51.100.20", 52_002)));
    }

    [Fact]
    public async Task ConcurrentDuplicateCompletionEmitsExactlyOneIntroduction()
    {
        using JoinAttemptFixture fixture = new();
        (RegisterSessionResponse registration, _) = fixture.CreateHost();
        AttemptCredentials attempt = CreateAttempt(fixture, registration, "completion-race");
        NatMediationProcessor processor = CreateProcessor(fixture);
        ConcurrentIntroductionSink sink = new();
        _ = Process(processor, sink, attempt, NatPunchPeerRole.Host,
            Endpoint("192.168.1.10", 41_000), Endpoint("203.0.113.20", 51_000));
        using Barrier barrier = new(2);

        Task<NatMediationResult>[] completions = Enumerable.Range(0, 2)
            .Select(_ => Task.Run(() =>
            {
                barrier.SignalAndWait();
                return Process(processor, sink, attempt, NatPunchPeerRole.Client,
                    Endpoint("192.168.1.11", 42_000), Endpoint("198.51.100.40", 52_000));
            }))
            .ToArray();
        NatMediationResult[] results = await Task.WhenAll(completions);

        Assert.Single(results, result => result == NatMediationResult.Introduced);
        Assert.Single(sink.Plans);
    }

    [Fact]
    public async Task CancellationCannotReportSuccessAfterIntroductionIsConsumed()
    {
        using JoinAttemptFixture fixture = new();
        (RegisterSessionResponse registration, _) = fixture.CreateHost();
        AttemptCredentials attempt = CreateAttempt(fixture, registration, "cancel-race");
        NatMediationProcessor processor = CreateProcessor(fixture);
        using BlockingIntroductionSink sink = new();
        _ = Process(processor, sink, attempt, NatPunchPeerRole.Host,
            Endpoint("192.168.1.10", 41_000), Endpoint("203.0.113.20", 51_000));
        Task<NatMediationResult> completion = Task.Run(() => Process(
            processor,
            sink,
            attempt,
            NatPunchPeerRole.Client,
            Endpoint("192.168.1.11", 42_000),
            Endpoint("198.51.100.40", 52_000)));
        Assert.True(sink.WaitUntilEntered(TimeSpan.FromSeconds(2)));

        JoinAttemptServiceResult<bool> cancelled = fixture.Service.Cancel(
            attempt.AttemptId,
            attempt.ClientCapability);

        Assert.Equal(RendezvousErrorCode.Conflict, cancelled.Error);
        sink.Release();
        Assert.Equal(NatMediationResult.Introduced, await completion);
    }

    [Fact]
    public void DuplicateFloodAmortizesGlobalExpiryMaintenance()
    {
        using JoinAttemptFixture fixture = new();
        (RegisterSessionResponse registration, _) = fixture.CreateHost();
        AttemptCredentials attempt = CreateAttempt(fixture, registration, "maintenance-budget");
        NatMediationProcessor processor = CreateProcessor(fixture);
        CaptureIntroductionSink sink = new();
        long before = fixture.Sessions.Store.MaintenanceSweepCount;

        for (int index = 0; index < 256; index++)
        {
            Assert.Equal(
                NatMediationResult.WaitingForPeer,
                Process(processor, sink, attempt, NatPunchPeerRole.Host,
                    Endpoint("192.168.1.10", 41_000), Endpoint("203.0.113.20", 51_000)));
        }

        Assert.InRange(fixture.Sessions.Store.MaintenanceSweepCount - before, 0, 1);
        Assert.Empty(sink.Plans);
    }

    [Fact]
    public void MissingStaleCancelledAndMalformedRequestsNeverIntroduce()
    {
        using JoinAttemptFixture fixture = new();
        (RegisterSessionResponse registration, _) = fixture.CreateHost();
        AttemptCredentials stale = CreateAttempt(fixture, registration, "stale");
        AttemptCredentials cancelled = CreateAttempt(fixture, registration, "cancelled");
        NatMediationProcessor processor = CreateProcessor(fixture);
        CaptureIntroductionSink sink = new();

        Assert.Equal(
            NatMediationResult.WaitingForPeer,
            Process(processor, sink, stale, NatPunchPeerRole.Client,
                Endpoint("192.168.1.11", 42_000), Endpoint("198.51.100.40", 52_000)));
        Assert.True(fixture.Service.Cancel(cancelled.AttemptId, cancelled.ClientCapability).Succeeded);
        Assert.Equal(
            NatMediationResult.Dropped,
            Process(processor, sink, cancelled, NatPunchPeerRole.Client,
                Endpoint("192.168.1.12", 42_001), Endpoint("198.51.100.41", 52_001)));

        fixture.Sessions.Clock.Advance(TimeSpan.FromSeconds(21));
        Assert.Equal(
            NatMediationResult.Dropped,
            Process(processor, sink, stale, NatPunchPeerRole.Host,
                Endpoint("192.168.1.10", 41_000), Endpoint("203.0.113.20", 51_000)));
        Assert.Equal(
            NatMediationResult.Dropped,
            processor.ProcessRequest(
                Endpoint("192.168.1.10", 41_000),
                Endpoint("203.0.113.20", 51_000),
                "malformed",
                sink));
        Assert.Empty(sink.Plans);
    }

    [Fact]
    public void AddressFamiliesMustMatchAndOnlyGlobalIpv6SourcesAreAccepted()
    {
        using JoinAttemptFixture fixture = new();
        (RegisterSessionResponse registration, _) = fixture.CreateHost();
        AttemptCredentials mismatch = CreateAttempt(fixture, registration, "family-mismatch");
        AttemptCredentials ipv6 = CreateAttempt(fixture, registration, "ipv6");
        NatMediationProcessor processor = CreateProcessor(fixture);
        CaptureIntroductionSink sink = new();

        byte[] shortFrozenIpv6 = RendezvousUdpCodec.Encode(new PresenceDatagram
        {
            MessageType = UdpPresenceMessageType.ClientPresence,
            MediationHandle = ipv6.Handle,
            AddressFamily = AddressFamilyKind.Ipv6,
            LocalAddress = "fd00::11",
            LocalPort = 42_000,
            Capability = ipv6.ClientCapability,
        });
        Assert.Equal(
            NatMediationResult.Dropped,
            processor.ProcessDatagram(
                shortFrozenIpv6,
                Endpoint("2606:4700:4700::1001", 52_000),
                sink));

        _ = Process(processor, sink, mismatch, NatPunchPeerRole.Host,
            Endpoint("192.168.1.10", 41_000), Endpoint("203.0.113.20", 51_000));
        Assert.Equal(
            NatMediationResult.Rejected,
            Process(processor, sink, mismatch, NatPunchPeerRole.Client,
                Endpoint("fd00::11", 42_000), Endpoint("2606:4700:4700::1111", 52_000)));

        Assert.Equal(
            NatMediationResult.Dropped,
            Process(processor, sink, ipv6, NatPunchPeerRole.Host,
                Endpoint("fd00::10", 41_000), Endpoint("2001:db8::10", 51_000)));
        _ = Process(processor, sink, ipv6, NatPunchPeerRole.Host,
            Endpoint("fd00::10", 41_000), Endpoint("2606:4700:4700::1000", 51_000));
        Assert.Equal(
            NatMediationResult.Introduced,
            Process(processor, sink, ipv6, NatPunchPeerRole.Client,
                Endpoint("fd00::11", 42_000), Endpoint("2606:4700:4700::1001", 52_000)));
        Assert.Single(sink.Plans);
    }

    private static NatMediationProcessor CreateProcessor(JoinAttemptFixture fixture) => new(
        fixture.Sessions.Store,
        fixture.Sessions.Capabilities,
        fixture.Service);

    private static AttemptCredentials CreateAttempt(
        JoinAttemptFixture fixture,
        RegisterSessionResponse registration,
        string idempotencyKey)
    {
        CreateJoinAttemptResponse created = fixture.Create(registration.ListingId, idempotencyKey);
        HostJoinAttempt host = fixture.Service.BrowseForHost(
            registration.ListingId,
            ContractLimits.ContractVersion,
            registration.LeaseToken,
            ContractLimits.BrowserPageMaxItems,
            null).Value!.Items.Single(item => item.AttemptId == created.AttemptId);
        return new(
            created.AttemptId,
            created.MediationHandle,
            host.HostPunchCapability,
            created.ClientPunchCapability);
    }

    private static NatMediationResult Process(
        NatMediationProcessor processor,
        INatIntroductionSink sink,
        AttemptCredentials attempt,
        NatPunchPeerRole role,
        IPEndPoint local,
        IPEndPoint observed) => processor.ProcessRequest(
            local,
            observed,
            NatPunchRequestTokenCodec.Encode(
                role,
                attempt.Handle,
                role == NatPunchPeerRole.Client
                    ? attempt.ClientCapability
                    : attempt.HostCapability),
            sink);

    private static IPEndPoint Endpoint(string address, int port) =>
        new(IPAddress.Parse(address), port);

    private sealed record AttemptCredentials(
        JoinAttemptId AttemptId,
        MediationHandle Handle,
        string HostCapability,
        string ClientCapability);

    private sealed class CaptureIntroductionSink : INatIntroductionSink
    {
        public List<NatIntroductionPlan> Plans { get; } = [];

        public void Introduce(NatIntroductionPlan plan) => Plans.Add(plan);
    }

    private sealed class ConcurrentIntroductionSink : INatIntroductionSink
    {
        public ConcurrentBag<NatIntroductionPlan> Plans { get; } = [];

        public void Introduce(NatIntroductionPlan plan) => Plans.Add(plan);
    }

    private sealed class BlockingIntroductionSink : INatIntroductionSink, IDisposable
    {
        private readonly ManualResetEventSlim _entered = new();
        private readonly ManualResetEventSlim _release = new();

        public void Introduce(NatIntroductionPlan plan)
        {
            _entered.Set();
            _release.Wait(TimeSpan.FromSeconds(2));
        }

        public bool WaitUntilEntered(TimeSpan timeout) => _entered.Wait(timeout);
        public void Release() => _release.Set();

        public void Dispose()
        {
            _entered.Dispose();
            _release.Dispose();
        }
    }
}