Arcturus.MassTransit.Security 1.0.0

Arcturus.MassTransit.Security

A security framework for MassTransit message buses. It adds:

• Identity propagation — forwards a service JWT (machine identity) and/or a user JWT (delegated end-user identity) alongside published/sent messages, via dedicated headers.
• Identity validation — consume-side filters that validate those JWTs against pluggable identity providers (Keycloak, Azure AD, etc.) and expose the resulting ClaimsPrincipals to the rest of the pipeline.
• Policy-based authorization — a fluent, compile-time-checked DSL for declaring per-message authorization policies (role/claim/client checks, conditional rules, custom rules), enforced by a consume filter before your consumer ever runs.

Table of contents

• Installation
• Core concepts
• Getting started
• Propagating identity (publisher side)
• Validating identity (consumer side)
• Policy-based authorization
• Full example
• Reference

Installation

dotnet add package Arcturus.MassTransit.Security

Targets net8.0 and depends on MassTransit, Microsoft.AspNetCore.Authorization, Microsoft.AspNetCore.Http.Abstractions, Microsoft.Net.Http.Headers and System.IdentityModel.Tokens.Jwt.

Core concepts

Getting started

1. Register the core services

using Arcturus.MassTransitSecurity;

builder.Services.AddMassTransitSecurity()
    .AddServiceIdentityFilter()
    .AddUserIdentityFilter();

• AddMassTransitSecurity() registers:
  • IPolicyRegistry → PolicyRegistry (singleton)
  • PolicyConfigurator (singleton)
  • IAuthorizationPolicyProvider → AuthorizationPolicyProvider (singleton)
  • IIdentityProviderResolver → IdentityProviderResolver (singleton)
  • PolicyAuthorizationFilter<> (scoped, open generic)
• AddServiceIdentityFilter() registers ServiceIdentityFilter<> (scoped, open generic).
• AddUserIdentityFilter() registers UserIdentityFilter<> (scoped, open generic).

2. Register at least one identity provider

IIdentityProviderResolver picks an IIdentityProvider for an incoming token based on its iss/aud claims. You must register at least one implementation:

using Arcturus.MassTransit.Security.Identity;

builder.Services.AddHttpClient<IIdentityProvider, KeycloakIdentityProvider>();

See Implementing IIdentityProvider for a worked example.

3. Wire the consume filters into the bus

using Arcturus.MassTransit.Security.Identity;

builder.Services.AddMassTransit(x =>
{
    x.SetKebabCaseEndpointNameFormatter();

    x.AddConsumer<CreatePaymentCommandConsumer>();

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("localhost", "/", h =>
        {
            h.Username("guest");
            h.Password("guest");
        });

        cfg.UseConsumeFilter(typeof(ServiceIdentityFilter<>), context);
        cfg.UseConsumeFilter(typeof(UserIdentityFilter<>), context);
        cfg.UseConsumeFilter(typeof(PolicyAuthorizationFilter<>), context);

        cfg.ConfigureEndpoints(context);
    });
});

Order matters. The identity filters must run before PolicyAuthorizationFilter<>, since it relies on the SecurityPayload they populate.

4. Define policies

using Arcturus.MassTransit.Security.Policy;

builder.Services.AddPolicies(cfg =>
{
    cfg.Policy<CreatePaymentCommand, ServiceAndUser>("create-payment")
        .RequireClient("payment-service")
        .RequireRole("payment:initiator")
        .RequireClaim("scope", "initiator");
});

The configure callback passed to AddPolicies runs when the DI container starts the registered IHostedServices (i.e. during host startup), via PolicyInitCompletedService.

If no policy is registered for a message type, PolicyAuthorizationFilter<T> falls back to a default Policy<T> with SecurityMode.ServiceOnly and no rules.

Propagating identity (publisher side)

Token accessors

Two small interfaces in Arcturus.MassTransit.Security.TokenAccessors abstract how to obtain the outgoing tokens:

public interface IServiceTokenAccessor
{
    ValueTask<string?> GetTokenAsync(CancellationToken cancellationToken = default);
}

public interface IUserTokenAccessor
{
    ValueTask<string?> GetTokenAsync(CancellationToken cancellationToken = default);
}

Built-in implementations:

ITokenClient (in Arcturus.MassTransit.Security.TokenClients) is the seam you implement to fetch a service token:

public interface ITokenClient
{
    ValueTask<string?> GetAccessTokenAsync(CancellationToken cancellationToken);
}

Example registration (matches SourceTestApplication):

using Arcturus.MassTransit.Security.TokenAccessors;
using Arcturus.MassTransit.Security.TokenClients;

builder.Services.AddHttpContextAccessor();

builder.Services.AddScoped<ITokenClient, KeycloakTokenClient>();
builder.Services.AddScoped<IServiceTokenAccessor, ClientCredentialsTokenAccessor>();
builder.Services.AddScoped<IUserTokenAccessor, HttpContextTokenAccessor>();

Where KeycloakTokenClient implements ITokenClient by calling Keycloak's token endpoint with a client_credentials grant.

Publish/Send filters

Arcturus.MassTransit.Security.Filters provides four filters that attach the tokens produced by the accessors above to outgoing messages:

Each filter calls GetTokenAsync() and, if a non-empty token is returned, sets the X-Service-Token or X-User-Token header on the outgoing context. If the accessor returns null/empty, the header is simply not set.

Register the publish filters (for IPublishEndpoint.Publish(...)):

using Arcturus.MassTransit.Security.Filters;

builder.Services.AddMassTransit(x =>
{
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("localhost", "/", h =>
        {
            h.Username("guest");
            h.Password("guest");
        });

        cfg.UsePublishFilter(typeof(SecurityPublishServiceFilter<>), context);
        cfg.UsePublishFilter(typeof(SecurityPublishUserFilter<>), context);
    });
});

For point-to-point Send, use cfg.UseSendFilter(typeof(SecuritySendServiceFilter<>), context) and/or SecuritySendUserFilter<> the same way.

Validating identity (consumer side)

ServiceIdentityFilter<T> / UserIdentityFilter<T>

Both filters live in Arcturus.MassTransit.Security.Identity and run as consume filters:

• ServiceIdentityFilter<T> — the X-Service-Token header is optional.
  If the header is not present, the filter simply calls next and skips service identity processing. If present, it strips a Bearer prefix, inspects the token's iss/aud (via TokenInspector), resolves an IIdentityProvider, validates the token, and stores the resulting ClaimsPrincipal as SecurityPayload.Service.

• UserIdentityFilter<T> — the X-User-Token header is optional.
  If the header is not present, the filter simply calls next and skips user identity processing. If present, it strips a Bearer prefix, inspects the token's iss/aud (via TokenInspector), resolves an IIdentityProvider, validates the token, and stores the resulting ClaimsPrincipal as SecurityPayload.User.

Both filters use context.AddOrUpdatePayload<SecurityPayload>(...) so they can run in either order and compose into the same payload.

Implementing IIdentityProvider

namespace Arcturus.MassTransit.Security.Identity
{
    public interface IIdentityProvider
    {
        string Name { get; }
        bool CanHandle(IdentityProviderContext context);
        Task<ClaimsPrincipal> ValidateAsync(IdentityProviderContext context);
    }

    public class IdentityProviderContext
    {
        public string? Issuer { get; set; }
        public string? Audience { get; set; }
        public string? Token { get; set; }
    }
}

Issuer/Audience are read directly off the (unvalidated) JWT by TokenInspector.Inspect and are used purely to route the token to the right provider via CanHandle. Actual signature/ issuer/audience validation happens in ValidateAsync.

A minimal Keycloak-backed provider:

public class KeycloakIdentityProvider : IIdentityProvider
{
    private readonly IConfigurationManager<OpenIdConnectConfiguration> _configManager;

    public string Name => "keycloak";

    public KeycloakIdentityProvider(HttpClient httpClient)
    {
        _configManager = new ConfigurationManager<OpenIdConnectConfiguration>(
            "http://localhost:8080/realms/test-realm/.well-known/openid-configuration",
            new OpenIdConnectConfigurationRetriever(),
            new HttpDocumentRetriever { RequireHttps = false }); // DEV ONLY
    }

    public bool CanHandle(IdentityProviderContext context) =>
        context.Issuer?.Contains(
            "http://localhost:8080/realms/test-realm",
            StringComparison.OrdinalIgnoreCase) == true;

    public async Task<ClaimsPrincipal> ValidateAsync(IdentityProviderContext context)
    {
        var config = await _configManager.GetConfigurationAsync(CancellationToken.None);

        var parameters = new TokenValidationParameters
        {
            ValidateIssuer = true,
            ValidIssuer = "http://localhost:8080/realms/test-realm",
            ValidateAudience = true,
            ValidAudience = "payment-service",
            IssuerSigningKeys = config.SigningKeys
        };

        return new JwtSecurityTokenHandler().ValidateToken(context.Token, parameters, out _);
    }
}

Register it as IIdentityProvider (e.g. AddHttpClient<IIdentityProvider, KeycloakIdentityProvider>()). You can register multiple providers — IdentityProviderResolver picks the one whose CanHandle returns true, and throws if zero or more than one match:

public class IdentityProviderResolver : IIdentityProviderResolver
{
    public IIdentityProvider Resolve(IdentityProviderContext context)
    {
        var providers = _providers.Where(p => p.CanHandle(context)).ToList();

        return providers.Count switch
        {
            0 => throw new UnauthorizedAccessException("No identity provider found."),
            1 => providers[0],
            _ => throw new InvalidOperationException(
                $"Multiple identity providers matched: {string.Join(", ", providers.Select(x => x.Name))}.")
        };
    }
}

Policy-based authorization

SecurityMode and compile-time-checked modes

SecurityMode (in Arcturus.MassTransit.Security.Policies) is a plain enum:

public enum SecurityMode
{
    ServiceOnly,
    UserOnly,
    ServiceAndUser
}

Rather than passing this enum around at runtime, policies are built with phantom mode types (in Arcturus.MassTransit.Security.Policy):

public interface IPolicyMode { static abstract SecurityMode Value { get; } }
public interface IRequiresService { }
public interface IRequiresUser { }

public sealed class ServiceOnly    : IPolicyMode, IRequiresService { /* Value = SecurityMode.ServiceOnly */ }
public sealed class UserOnly       : IPolicyMode, IRequiresUser    { /* Value = SecurityMode.UserOnly */ }
public sealed class ServiceAndUser : IPolicyMode, IRequiresService, IRequiresUser
                                                                     { /* Value = SecurityMode.ServiceAndUser */ }

Policy<T, TMode> derives Mode from TMode.Value. Each rule-attaching extension method constrains TMode to the marker interface(s) it needs:

This means an incompatible combination is rejected at compile time:

cfg.Policy<CreatePaymentCommand, ServiceOnly>("create-payment")
    .RequireRole("payment:initiator"); // ❌ CS0311: 'ServiceOnly' does not satisfy 'IRequiresUser'

cfg.Policy<CreatePaymentCommand, UserOnly>("create-payment")
    .RequireClient("payment-service"); // ❌ CS0311: 'UserOnly' does not satisfy 'IRequiresService'

Use ServiceAndUser whenever a policy needs both user-only rules (RequireRole, RequireClaim) and service-only rules (RequireClient).

Defining policies

using Arcturus.MassTransit.Security.Policy;

builder.Services.AddPolicies(cfg =>
{
    cfg.Policy<CreatePaymentCommand, ServiceAndUser>("create-payment")
        .RequireClient("payment-service")
        .RequireRole("payment:initiator")
        .RequireClaim("scope", "payer");

    cfg.Policy<ApprovePaymentCommand, UserOnly>("approve-payment")
        .RequireRole("payment:approver");
});

cfg.Policy<T, TMode>(name):

1. Creates new Policy<T, TMode>(name) via PolicyFactory.For<T, TMode>(name).
2. Registers it in the IPolicyRegistry, keyed by typeof(T).
3. Returns the policy so you can chain rule-attaching extension methods.

Built-in rules

All rule extension methods live in Arcturus.MassTransit.Security.Policy.PolicyExtensions and return the policy for chaining (policy.AttachRule(...)).

RequireRole(string role)

Passes if context.User!.IsInRole(role). Requires IRequiresUser (UserOnly / ServiceAndUser).

cfg.Policy<ApprovePaymentCommand, UserOnly>("approve-payment")
    .RequireRole("payment:approver");

RequireClaim(string type, string value)

Passes if context.User!.HasClaim(type, value). Requires IRequiresUser.

cfg.Policy<CreatePaymentCommand, ServiceAndUser>("create-payment")
    .RequireClaim("scope", "payer");

RequireClient(string client)

Passes if context.Service?.FindFirst("client_id")?.Value == client. Requires IRequiresService (ServiceOnly / ServiceAndUser).

cfg.Policy<CreatePaymentCommand, ServiceAndUser>("create-payment")
    .RequireClient("payment-service");

DenyIf(Func<PolicyContext<T>, Task<bool>> predicate)

Fails the policy when predicate(context) returns true (and passes otherwise). Available on every mode.

cfg.Policy<ApprovePaymentCommand, UserOnly>("approve-payment")
    .RequireRole("payment:senior-approver")
    .DenyIf(async ctx => ctx.Message.Amount > 50_000);

IfElse(condition, truePredicate, falsePredicate)

public static Policy<T, TMode> IfElse<T, TMode>(
    this Policy<T, TMode> policy,
    Func<PolicyContext<T>, Task<bool>> condition,
    Action<Policy<T, TMode>> truePredicate = null!,
    Action<Policy<T, TMode>> falsePredicate = null!)
    where T : class where TMode : IPolicyMode

condition is evaluated against the live PolicyContext<T> (so it can inspect context.Message, context.User, context.Service, etc.). Whichever branch matches (truePredicate or falsePredicate) is invoked with a fresh nested Policy<T, TMode>; any rules it attaches must all pass for the IfElse rule to pass. A null branch (or a branch that attaches no rules) means "allow" for that branch.

// Amounts over 1000 require an "senior-approver" role; smaller amounts only need "approver".
cfg.Policy<ApprovePaymentCommand, UserOnly>("approve-payment")
    .IfElse(
        condition: async ctx => ctx.Message.Amount > 1000,
        truePredicate: p => p.RequireRole("payment:senior-approver"),
        falsePredicate: p => p.RequireRole("payment:approver"));

IfElse branches can be nested, e.g. to layer extra conditions on top of the "true" branch:

cfg.Policy<ApprovePaymentCommand, UserOnly>("approve-payment")
    .IfElse(
        condition: async ctx => ctx.Message.Amount > 1000,
        truePredicate: p => p.IfElse(
            condition: async ctx => ctx.Message.Amount > 10_000,
            truePredicate: inner => inner.RequireRole("payment:senior-approver"),
            falsePredicate: inner => inner.RequireRole("payment:approver")),
        falsePredicate: p => p.RequireRole("payment:junior-approver"));

Writing custom rules

A rule is any IPolicyRule<T>:

namespace Arcturus.MassTransit.Security.Policy
{
    public interface IPolicyRule<T> where T : class
    {
        Task<bool> Evaluate(PolicyContext<T> context);
    }
}

PolicyContext<T> exposes everything a rule typically needs:

public class PolicyContext<T> where T : class
{
    public ConsumeContext<T> MessageContext { get; }
    public ClaimsPrincipal? Service => Payload.Service;
    public ClaimsPrincipal? User => Payload.User;
    public SecurityPayload Payload { get; }
    public IServiceProvider ServiceProvider { get; } // resolve scoped app services
    public T Message => MessageContext.Message;
}

Pair a rule with an extension method constrained to the modes where it makes sense — this keeps the same compile-time safety for your own rules:

using Arcturus.MassTransit.Security.Policy;

public static class CustomRules
{
    public static Policy<T, TMode> RequireActiveUser<T, TMode>(this Policy<T, TMode> policy)
        where T : class
        where TMode : IPolicyMode, IRequiresUser
    {
        policy.AttachRule(new RequireActiveUserRule<T>());
        return policy;
    }
}

public class RequireActiveUserRule<T> : IPolicyRule<T> where T : class
{
    public async Task<bool> Evaluate(PolicyContext<T> context)
    {
        var userService = context.ServiceProvider.GetRequiredService<IUserService>();
        var userId = context.User!.FindFirst(ClaimTypes.NameIdentifier)?.Value;

        return userId is not null && await userService.IsUserActive(userId);
    }
}

Usage:

cfg.Policy<CreatePaymentCommand, ServiceAndUser>("create-payment")
    .RequireClient("payment-service")
    .RequireRole("payment:initiator")
    .RequireActiveUser();

Enforcement: PolicyAuthorizationFilter<T>

For every consumed message of type T, this consume filter:

1. Reads the SecurityPayload from the consume context (throws UnauthorizedAccessException("Missing security context") if the identity filters never ran).
2. Looks up Policy<T> from IPolicyRegistry (or the ServiceOnly/no-rules default).
3. Checks policy.Mode against the payload:
   • ServiceOnly → payload.Service must be non-null, else "Service required".
   • UserOnly → payload.User must be non-null, else "User required".
   • ServiceAndUser → both must be non-null, else "Service and user required".
4. Builds a PolicyContext<T> and evaluates policy.Rules in order, throwing UnauthorizedAccessException($"Policy '{policy.Name}' failed") on the first rule that returns false.
5. Calls next only if every check passed.

Full example

Consumer (DestinationTestApplication)

using Arcturus.MassTransit.Security.Identity;
using Arcturus.MassTransit.Security.Policy;
using Arcturus.MassTransitSecurity;

var builder = WebApplication.CreateBuilder(args);

// 1. Core services + identity filters
builder.Services.AddMassTransitSecurity()
    .AddServiceIdentityFilter()
    .AddUserIdentityFilter();

// 2. Identity provider(s)
builder.Services.AddHttpClient<IIdentityProvider, KeycloakIdentityProvider>();

// Application services used by custom rules
builder.Services.AddScoped<IUserService, UserService>();

// 3. MassTransit + consume filter pipeline
builder.Services.AddMassTransit(x =>
{
    x.SetKebabCaseEndpointNameFormatter();

    x.AddConsumer<CreatePaymentCommandConsumer>();
    x.AddConsumer<ApprovePaymentCommandConsumer>();

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("localhost", "/", h =>
        {
            h.Username("guest");
            h.Password("guest");
        });

        cfg.UseConsumeFilter(typeof(ServiceIdentityFilter<>), context);
        cfg.UseConsumeFilter(typeof(UserIdentityFilter<>), context);
        cfg.UseConsumeFilter(typeof(PolicyAuthorizationFilter<>), context);

        cfg.ConfigureEndpoints(context);

        cfg.UseMessageRetry(r => r.Interval(5, 2));
    });
});

// 4. Policies
builder.Services.AddPolicies(cfg =>
{
    cfg.Policy<CreatePaymentCommand, ServiceAndUser>("create-payment")
        .RequireClient("payment-service")
        .RequireRole("payment:initiator")
        .RequireClaim("scope", "payer")
        .RequireActiveUser();

    cfg.Policy<ApprovePaymentCommand, UserOnly>("approve-payment")
        .IfElse(
            condition: async ctx => ctx.Message.Amount > 10000,
            truePredicate: p => p.RequireRole("payment:senior-approver"),
            falsePredicate: p => p.RequireRole("payment:approver"));
});

var app = builder.Build();
app.Run();

Publisher (SourceTestApplication)

using Arcturus.MassTransit.Security.Filters;
using Arcturus.MassTransit.Security.TokenAccessors;
using Arcturus.MassTransit.Security.TokenClients;

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddHttpContextAccessor();

// Token accessors: service token via client-credentials, user token from the inbound request
builder.Services.AddScoped<ITokenClient, KeycloakTokenClient>();
builder.Services.AddScoped<IServiceTokenAccessor, ClientCredentialsTokenAccessor>();
builder.Services.AddScoped<IUserTokenAccessor, HttpContextTokenAccessor>();

builder.Services.AddMassTransit(x =>
{
    x.SetKebabCaseEndpointNameFormatter();

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("localhost", "/", h =>
        {
            h.Username("guest");
            h.Password("guest");
        });

        cfg.UsePublishFilter(typeof(SecurityPublishServiceFilter<>), context);
        cfg.UsePublishFilter(typeof(SecurityPublishUserFilter<>), context);
    });
});

var app = builder.Build();
app.Run();

// Controller action: publishing attaches X-Service-Token and X-User-Token automatically.
[HttpPost]
[Authorize]
public async Task<IActionResult> CreatePayment([FromServices] IPublishEndpoint publishEndpoint)
{
    await publishEndpoint.Publish(new CreatePaymentCommand
    {
        Id = Guid.NewGuid(),
        Amount = 100_000
    }, HttpContext.RequestAborted);

    return Ok();
}

Reference

Namespace map

Headers

public static class SecurityHeaders
{
    public const string ServiceToken = "X-Service-Token";
    public const string UserToken = "X-User-Token";
}