natsdotnet/tests/NATS.Server.Tests/PermissionIntegrationTests.cs

using System.Net;
using System.Net.Sockets;
using Microsoft.Extensions.Logging.Abstractions;
using NATS.Client.Core;
using NATS.Server.Auth;

namespace NATS.Server.Tests;

public class PermissionIntegrationTests : IAsyncLifetime
{
    private NatsServer _server = null!;
    private int _port;
    private readonly CancellationTokenSource _cts = new();
    private Task _serverTask = null!;

    private static int GetFreePort()
    {
        using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
        sock.Bind(new IPEndPoint(IPAddress.Loopback, 0));
        return ((IPEndPoint)sock.LocalEndPoint!).Port;
    }

    public async Task InitializeAsync()
    {
        _port = GetFreePort();
        _server = new NatsServer(new NatsOptions
        {
            Port = _port,
            Users =
            [
                new User
                {
                    Username = "publisher",
                    Password = "pass",
                    Permissions = new Permissions
                    {
                        Publish = new SubjectPermission { Allow = ["events.>"] },
                        Subscribe = new SubjectPermission { Deny = [">"] },
                    },
                },
                new User
                {
                    Username = "subscriber",
                    Password = "pass",
                    Permissions = new Permissions
                    {
                        Publish = new SubjectPermission { Deny = [">"] },
                        Subscribe = new SubjectPermission { Allow = ["events.>"] },
                    },
                },
                new User
                {
                    Username = "admin",
                    Password = "pass",
                    // No permissions — full access
                },
            ],
        }, NullLoggerFactory.Instance);

        _serverTask = _server.StartAsync(_cts.Token);
        await _server.WaitForReadyAsync();
    }

    public async Task DisposeAsync()
    {
        await _cts.CancelAsync();
        _server.Dispose();
    }

    [Fact]
    public async Task Publisher_can_publish_to_allowed_subject()
    {
        await using var pub = new NatsConnection(new NatsOpts
        {
            Url = $"nats://publisher:pass@127.0.0.1:{_port}",
        });
        await using var admin = new NatsConnection(new NatsOpts
        {
            Url = $"nats://admin:pass@127.0.0.1:{_port}",
        });

        await pub.ConnectAsync();
        await admin.ConnectAsync();

        await using var sub = await admin.SubscribeCoreAsync<string>("events.test");
        await admin.PingAsync();

        await pub.PublishAsync("events.test", "hello");

        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
        var msg = await sub.Msgs.ReadAsync(timeout.Token);
        msg.Data.ShouldBe("hello");
    }

    [Fact]
    public async Task Admin_has_full_access()
    {
        await using var admin1 = new NatsConnection(new NatsOpts
        {
            Url = $"nats://admin:pass@127.0.0.1:{_port}",
        });
        await using var admin2 = new NatsConnection(new NatsOpts
        {
            Url = $"nats://admin:pass@127.0.0.1:{_port}",
        });

        await admin1.ConnectAsync();
        await admin2.ConnectAsync();

        await using var sub = await admin2.SubscribeCoreAsync<string>("anything.at.all");
        await admin2.PingAsync();

        await admin1.PublishAsync("anything.at.all", "data");

        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
        var msg = await sub.Msgs.ReadAsync(timeout.Token);
        msg.Data.ShouldBe("data");
    }
}