natsmqtt5

README

natsmqtt5

An MQTT v5.0 broker that uses an existing NATS server as its messaging core.

nats-server speaks MQTT 3.1.1 and will not be adding v5: "we don't want to own the ongoing maintenance cost for the massive API surface increase that comes with MQTT 5 at this time." This is the other option raised in that thread — a separate broker that keeps NATS as the messaging engine, so you point it at the NATS cluster you already run rather than deploying a second, unrelated message bus.

Built against: MQTT 5.0 (OASIS, 2019-03-07 — still the current revision), nats.go v1.53.1 and nats-server v2.14.5, checked 2026-09-24. Both NATS pins are deliberate holds: v1.54.0 and v2.15.0 declare go 1.26.0, and taking either would raise this module's floor from Go 1.25 and drop anyone building on it (#11). nats.go is in the shipped build, so that hold costs you a NATS client one minor version behind — nothing is broken, but it is a version you did not choose. As of 2026-09-21 it also costs you a security patch: the hold keeps golang.org/x/crypto at v0.55.0, which carries two x/crypto/ssh DoS advisories (GO-2026-6354, GO-2026-6355) fixed in v0.56.0 — and v0.56.0 declares go 1.26.0 too, so there is no patched version reachable from a Go 1.25 floor. govulncheck reports them as not called by this module, so it is not an exploitable path here; it is an unpatchable one. See UPSTREAM.md, CHANGELOG.md for what each release changed under you, and ROADMAP.md for where the broker knowingly does not yet conform.

MQTT v5 clients ──TCP/TLS──▶ natsmqtt5 ──▶ your existing NATS server
                             (a NATS client, not a fork)

What it does

• Speaks MQTT v5 to clients. All fifteen control packets, all 27 properties. Verified against the Eclipse Paho v5 client over a real socket.
• Uses NATS for everything underneath. Topics become NATS subjects, subscriptions become NATS subscriptions, shared subscriptions become NATS queue groups, retained messages live in a JetStream stream.
• Interoperates on the subject level. The topic-to-subject mapping is byte-for-byte the one nats-server uses for its own MQTT support, so sensors/7/temp lands on mqtt.sensors.7.temp either way. A NATS-native publisher reaches MQTT subscribers and vice versa, with no bridge.
• Scales horizontally by adding brokers. Two brokers against one NATS server are one logical broker: a client on either reaches subscribers on both.
• Keeps sessions across brokers, if you ask it to. With PersistentSessions, a client that reconnects with Clean Start 0 resumes its subscriptions on a broker that has never served it, and after a restart. Off by default; see Persistent sessions.
• Embeds as a library. natsmqtt5.New returns a *Broker you run inside your own process, sharing your *nats.Conn if you want.

Install

As a container, against a NATS server you already run:

docker run --rm -p 1883:1883 \
  -e NATSMQTT5_NATS=nats://your-nats-server:4222 \
  ghcr.io/taumatix/natsmqtt5:v0.6.0

Images are published for linux/amd64 and linux/arm64 on every release, as :v0.6.0, :0.6.0, :0.6 and :latest. They are built from distroless/static, so there is no shell and no package manager in them, and the broker runs as a non-root user.

If you have no NATS server yet, compose.yaml starts one with JetStream enabled and the broker in front of it:

curl -O https://raw.githubusercontent.com/taumatix/natsmqtt5/v0.6.0/compose.yaml
docker compose up -d
mosquitto_pub -V 5 -h localhost -p 1883 -t sensors/7/temp -m 21.5

Every flag has an environment variable twin — upper-case, - becomes _, behind a NATSMQTT5_ prefix — so -subject-prefix is NATSMQTT5_SUBJECT_PREFIX. An explicit flag beats the environment. Run docker run --rm ghcr.io/taumatix/natsmqtt5:v0.6.0 -h for the full list.

As a binary:

go install github.com/taumatix/natsmqtt5/cmd/natsmqtt5@v0.6.0
natsmqtt5 -nats nats://localhost:4222 -listen :1883

As a library:

go get github.com/taumatix/natsmqtt5@v0.6.0

Requires Go 1.25 or newer, and a NATS server with JetStream enabled (or -no-retained / Options.DisableRetained to run without it).

Use

package main

import (
	"context"
	"log"

	"github.com/taumatix/natsmqtt5"
)

func main() {
	broker, err := natsmqtt5.New(natsmqtt5.Options{
		NATSURL: "nats://localhost:4222",
		Listen:  ":1883",
	})
	if err != nil {
		log.Fatal(err)
	}
	defer broker.Close()

	if err := broker.Serve(context.Background()); err != nil {
		log.Fatal(err)
	}
}

Authentication and authorization are two small interfaces:

opts := natsmqtt5.Options{
	Authenticator: natsmqtt5.AuthenticatorFunc(
		func(ctx context.Context, req *natsmqtt5.AuthRequest) (*natsmqtt5.AuthResult, error) {
			if !credentialsOK(req.Username, req.Password) {
				// Choose the CONNACK Reason Code the client sees.
				return nil, &natsmqtt5.ConnectError{Code: packet.BadUserNameOrPassword}
			}
			// A Session is keyed by its Client Identifier alone, so an
			// authenticated user must not be free to choose any. Equality on
			// the first segment, not a prefix test: user "a" passes
			// HasPrefix("a/b/sensor", "a/") and would inherit "a/b"'s session.
			if owner, _, found := strings.Cut(req.ClientID, "/"); !found || owner != req.Username {
				return nil, &natsmqtt5.ConnectError{Code: packet.ClientIdentifierNotValid}
			}
			return &natsmqtt5.AuthResult{Identity: req.Username}, nil
		}),

	Authorizer: natsmqtt5.AuthorizerFunc(
		func(ctx context.Context, req *natsmqtt5.AuthzRequest) error {
			if req.Action == natsmqtt5.ActionPublish && !mayWrite(req.Identity, req.Topic) {
				return errors.New("denied") // becomes 0x87 Not authorized in the PUBACK
			}
			return nil
		}),
}

AuthRequest carries the TLS ConnectionState, so client-certificate authentication is a few lines.

The Client Identifier check above is not decoration. A Session is identified by its Client Identifier alone (MQTT-5.0 §4.1), so an Authenticator that verifies a principal but accepts any identifier lets that principal inherit another's session — its subscriptions, its unacknowledged messages and its QoS 2 receive state. Bind the identifier to the principal, or — for a client that sends a zero-length identifier to have one assigned [MQTT-3.1.3-6] — derive one with AuthResult.AssignClientID, as Example_authentication shows.

The broker re-runs the Authorizer over a resumed session's filters, with AuthzRequest.Resume set, so a permission narrowed between two connections takes effect on the second one rather than when the session expires. That bounds the damage; it does not remove the need to bind the identifier, and it has two costs worth knowing about. A slow Authorizer delays the CONNACK in proportion to the session's filter count. And a denial there is silent — there is no per-filter Reason Code in a CONNACK, so the filter is torn down and the client is still told SessionPresent: true.

A Will Message goes past the Authorizer too, at CONNECT, as an ActionPublish with AuthzRequest.Will set. The broker publishes a Will on the client's behalf after the client has gone, so CONNECT is the only moment it can be refused: a denied Will refuses the connection with 0x87, and the Will is not checked again when it fires.

A client that stays connected is not re-checked by itself, since asking on every delivery would put your Authorizer on the hot path of every message. When a principal's permissions change, call Broker.Reauthorize(ctx, clientID), from whatever tells you about revocations:

• each subscription goes back past the Authorizer (with Resume set), and a denied one stops delivering at once, including what was already queued;
• its unacknowledged messages are withdrawn, so a later resume does not resend them;
• the Will is checked again, and a denied one is discarded, even one already waiting out its Will Delay Interval.

The connection stays up, because a DISCONNECT would publish the Will being revoked. As at a resume, the client is not told which filters it lost, and any error from the Authorizer counts as a denial.

Any broker will do. Brokers sharing a NATS cluster and a SubjectPrefix answer one another on _NATSMQTT5.reauthorize.<prefix>, and the one holding the session runs the sweep. When none holds it, Reauthorize returns nil after ReauthorizeForwardWait (one second), since an offline session is re-checked when it next connects. Anything that can publish on that subject can make a broker re-ask your Authorizer, which can only narrow a session; restrict _NATSMQTT5.> to the brokers' NATS user if your other NATS clients are not trusted.

How MQTT maps onto NATS

MQTT	NATS
Topic sensors/7/temp	Subject mqtt.sensors.7.temp
Topic level /	Subject token separator .
A literal . in a topic	//
An empty topic level	./ or /. depending on position
Filter wildcard +	*
Filter wildcard #	>, plus a second subscription on the parent
Shared subscription $share/g/f	Queue group on the subject for f
Retained message	JetStream stream, MaxMsgsPerSubject: 1
Session state (opt-in)	JetStream key-value bucket, one key per Client Identifier
v5 properties	NATS headers (Mqtt5-Content-Type, Mqtt5-User, …)

# gets two NATS subscriptions because MQTT's # matches the parent level (sport/# matches sport) and NATS's > does not.

The mqtt subject prefix is configurable with SubjectPrefix; brokers that should share traffic must agree on it, and brokers that must not share traffic must differ.

A NATS message with no Mqtt5-* headers is delivered as a QoS 0 message with no properties, which is what makes plain nats pub reach MQTT subscribers.

Persistent sessions

By default a session lives in the broker process: a client reconnecting to the same broker within its Session Expiry Interval resumes its subscriptions, and one reconnecting to a different broker, or after a restart, has to subscribe again.

Turn that off with one setting:

natsmqtt5.Options{PersistentSessions: true}

or -persistent-sessions / NATSMQTT5_PERSISTENT_SESSIONS=true for the binary and the image. Every broker that should share sessions needs it set, and needs the same StreamPrefix, since that names the bucket.

The subscription set is mirrored into a JetStream key-value bucket (MQTT5_sessions by default), keyed by Client Identifier. A CONNECT claims the record with a compare-and-swap on its revision, so of two brokers racing for one Client Identifier exactly one wins — and the loser disconnects its client with 0x8E Session taken over rather than serving it in parallel [MQTT-3.1.4-3]. That claim is also what makes the Session Present flag mean something across a restart.

compose.yaml runs two brokers this way:

NATSMQTT5_PERSISTENT_SESSIONS=true docker compose --profile cluster up -d

What it costs and what it does not cover:

• One JetStream write per CONNECT, SUBSCRIBE, UNSUBSCRIBE and DISCONNECT. None of those are the throughput path; PUBLISH is untouched.
• The Client Identifier is capped at 128 bytes (MaxPersistentClientIDLen), because it becomes part of a NATS subject. A longer one is refused with 0x85 Client Identifier not valid. Without persistence there is no limit.
• In-flight QoS 1 and QoS 2 messages are not persisted. A client that reconnects to the broker still holding its session gets its unacknowledged messages back; one whose session is restored from the store — after a restart, or on another broker — does not, because the record carries the subscription set and not the in-flight set. Storing Packet Identifiers without the payloads to go with them would promise a resend the broker could not perform.
• A filter the Authorizer no longer permits is dropped without telling the client. Every filter of a resumed session goes back past the Authorizer before the CONNACK, on both branches, so a permission narrowed while the client was away takes effect immediately. A CONNACK has no per-filter Reason Code to carry the refusal, so the dropped filters are logged and the client still sees SessionPresent: true; it finds out by not receiving. The unacknowledged messages a dropped filter earned go with it.
• The Will Message is not persisted. A Will belongs to the network connection (MQTT-5.0 §3.1.2.5), and a broker reading a record cannot tell a connection that has ended from an owner that is merely busy, so publishing from there would announce live clients as dead. A broker that shuts down cleanly still publishes its clients' Wills; one killed outright does not.
• A session whose broker was killed outright is resumed whenever its client comes back, however long that takes. Expiry is measured from the moment a session is released, and a kill -9 records no release. A background sweep reclaims such a record once it is older than MaxSessionExpiry and its broker has stopped answering.

What it does not do

Stated plainly, because a broker you cannot trust the limits of is worse than one with fewer features:

• No offline message queue. Messages published while a session is disconnected are not stored for it, whether or not the session is persisted.
• Retransmission on reconnect reaches as far as the broker's memory. A client that reconnects with Clean Start 0 to a broker still holding its session is resent the PUBLISH and PUBREL packets that connection left unacknowledged, with their original Packet Identifiers [MQTT-4.4.0-1]. A session restored from the store instead — after a restart, or on another broker — has nothing to resend.
• Shared subscriptions are granted QoS 1 at most. A QoS 2 request on a $share/ filter is answered with Granted QoS 1 in the SUBACK, which [MQTT-3.8.4-7] permits. Exactly-once delivery to one member of a group needs per-member ownership of a half-finished PUBREC/PUBREL exchange, and JetStream's unit of ownership is a consumer, not a queue-group member.
• No enhanced authentication (AUTH packets). A CONNECT carrying an Authentication Method is refused with 0x8C Bad authentication method, which is the conforming answer from a server that does not support it.
• The broker never sends topic aliases, though it accepts them from clients.
• Message Expiry Interval is forwarded but not enforced on stored retained messages.
• A PUBACK does not mean the message reached NATS. The broker hands the message to its NATS connection and acknowledges; the write is flushed microseconds later. A broker killed in that window loses a QoS 1 message it has already acknowledged. A SUBACK, by contrast, does wait for the NATS server to confirm the subscription — otherwise a message published immediately after subscribing could be dropped, which is a loss the client has no way to notice.
• Topics containing a space, a tab, *, > or DEL are refused with 0x90 Topic Name invalid. MQTT permits them; a NATS subject cannot carry them. nats-server lets * and > through, which silently turns an MQTT subscription to a/*/b into a NATS wildcard that receives topics it must not see. Failing loudly is the deliberate choice here; see ROADMAP.md for the escaping option.

Correctness

The MQTT layer is written against the OASIS MQTT v5.0 Standard of 07 March 2019, and the tests cite it:

• The codec tests assert the literal bytes from the spec's own worked examples (Figures 3-6, 3-9, 3-19, 3-21) and every boundary in Table 1-1, rather than only round-tripping our own encoder against our own decoder.
• The topic mapping is checked against all 40 conversion fixtures from nats-server's TestMQTTTopicAndSubjectConversion and TestMQTTFilterConversion, so a divergence from nats-server fails the build.
• The matching tests use the examples in §4.7.1.2, §4.7.1.3, §4.7.2 and §4.8.2.
• Integration tests run a real NATS server in-process with JetStream, a real broker, and the Eclipse Paho v5 client over a real TCP socket.
• The packet decoder and the topic mapping are fuzzed in CI.

Packages

Package	What it is
github.com/taumatix/natsmqtt5	The broker
.../packet	MQTT v5 wire codec — standard library only, no NATS
.../topic	Topic validation, matching and the NATS subject mapping

packet and topic are usable on their own; neither imports NATS.

Licence

Apache 2.0. See LICENSE.

Documentation

Overview

Package natsmqtt5 is an MQTT version 5.0 broker that uses an existing NATS server as its messaging core.

The broker is a NATS client, not a fork of nats-server. Point it at a NATS URL you already run and it speaks MQTT v5 on a TCP port, translating topics to NATS subjects, fan-out to NATS subscriptions, shared subscriptions to NATS queue groups and retained messages to a JetStream stream. Two brokers against the same NATS server see each other's traffic, so an MQTT client on one reaches a subscriber on the other.

The subject mapping is byte-for-byte the one nats-server uses for its own MQTT 3.1.1 support, so topics land where a NATS-native subscriber expects.

Minimal use:

b, err := natsmqtt5.New(natsmqtt5.Options{NATSURL: "nats://localhost:4222"})
if err != nil {
	return err
}
defer b.Close()
return b.Serve(ctx)

Section numbers in the documentation refer to the OASIS MQTT v5.0 Standard of 07 March 2019.

Example

Run a broker against an existing NATS server.

package main

import (
	"context"
	"log"

	"github.com/taumatix/natsmqtt5"
)

func main() {
	broker, err := natsmqtt5.New(natsmqtt5.Options{
		NATSURL: "nats://localhost:4222",
		Listen:  ":1883",
	})
	if err != nil {
		log.Fatal(err)
	}
	defer broker.Close()

	if err := broker.Serve(context.Background()); err != nil {
		log.Fatal(err)
	}
}

Example (Authentication)

Reject unknown credentials with the CONNACK Reason Code of your choosing, and confine each user to its own topic subtree.

package main

import (
	"context"
	"errors"
	"log"
	"strings"

	"github.com/nats-io/nuid"

	"github.com/taumatix/natsmqtt5"
	"github.com/taumatix/natsmqtt5/packet"
)

func main() {
	passwords := map[string]string{"alice": "s3cret"}

	opts := natsmqtt5.Options{
		NATSURL: "nats://localhost:4222",

		Authenticator: natsmqtt5.AuthenticatorFunc(
			func(_ context.Context, req *natsmqtt5.AuthRequest) (*natsmqtt5.AuthResult, error) {
				if want, ok := passwords[req.Username]; !ok || want != string(req.Password) {
					return nil, &natsmqtt5.ConnectError{Code: packet.BadUserNameOrPassword}
				}
				// A Session is keyed by its Client Identifier alone, so an
				// authenticated user must not be free to choose any: without
				// this, alice connecting as "bob-sensor" inherits bob's
				// session and everything it is subscribed to.
				//
				// A zero-length identifier is the client asking the server to
				// assign one [MQTT-3.1.3-6], so it is answered rather than
				// refused.
				if req.ClientID == "" {
					return &natsmqtt5.AuthResult{
						Identity:       req.Username,
						AssignClientID: req.Username + "/" + nuid.Next(),
					}, nil
				}
				// Equality on the first segment, not a prefix test: a user
				// named "a" passes HasPrefix("a/b/sensor", "a/") and would
				// inherit the session of the user named "a/b".
				if owner, _, found := strings.Cut(req.ClientID, "/"); !found || owner != req.Username {
					return nil, &natsmqtt5.ConnectError{Code: packet.ClientIdentifierNotValid}
				}
				return &natsmqtt5.AuthResult{Identity: req.Username}, nil
			}),

		Authorizer: natsmqtt5.AuthorizerFunc(
			func(_ context.Context, req *natsmqtt5.AuthzRequest) error {
				// A denied publish comes back as 0x87 in the PUBACK; a denied
				// subscription as 0x87 in the SUBACK. The connection stays up.
				if !strings.HasPrefix(req.Topic, "users/"+req.Identity+"/") {
					return errors.New("outside the user's subtree")
				}
				return nil
			}),
	}

	broker, err := natsmqtt5.New(opts)
	if err != nil {
		log.Fatal(err)
	}
	defer broker.Close()

	log.Fatal(broker.Serve(context.Background()))
}

Example (Embedded)

Share one NATS connection between the broker and the rest of an application, and keep MQTT traffic in its own subject namespace.

package main

import (
	"context"
	"log"

	"github.com/nats-io/nats.go"

	"github.com/taumatix/natsmqtt5"
)

func main() {
	// An existing connection the application already uses. The broker does not
	// close a connection it did not open.
	nc, err := nats.Connect("nats://localhost:4222")
	if err != nil {
		log.Fatal(err)
	}
	defer nc.Close()

	broker, err := natsmqtt5.New(natsmqtt5.Options{
		Conn:          nc,
		Listen:        "127.0.0.1:1883",
		SubjectPrefix: "iot",
		StreamPrefix:  "IOT",
	})
	if err != nil {
		log.Fatal(err)
	}
	defer broker.Close()

	// An MQTT client publishing to "sensors/7/temp" now reaches any NATS
	// subscriber on "iot.sensors.7.temp", and the other way round.
	log.Fatal(broker.Serve(context.Background()))
}

Index

• Constants
• Variables
• func Ptr[T any](v T) *T
• type Action
  • func (a Action) String() string
• type AuthRequest
• type AuthResult
• type Authenticator
• type AuthenticatorFunc
  • func (f AuthenticatorFunc) Authenticate(ctx context.Context, req *AuthRequest) (*AuthResult, error)
• type Authorizer
• type AuthorizerFunc
  • func (f AuthorizerFunc) Authorize(ctx context.Context, req *AuthzRequest) error
• type AuthzRequest
• type Broker
  • func New(opts Options) (*Broker, error)
  • func NewWithContext(ctx context.Context, opts Options) (*Broker, error)
  • func (b *Broker) Close() error
  • func (b *Broker) ListenAddr() net.Addr
  • func (b *Broker) NATS() *nats.Conn
  • func (b *Broker) Reauthorize(ctx context.Context, clientID string) error
  • func (b *Broker) Serve(ctx context.Context) error
• type ConnectError
  • func (e *ConnectError) Error() string
• type Options

Examples

• Package
• Package (Authentication)
• Package (Embedded)

Constants

const (
	DefaultListen        = ":1883"
	DefaultSubjectPrefix = "mqtt"
	DefaultStreamPrefix  = "MQTT5"
	// DefaultReceiveMaximum bounds how many QoS 1 and QoS 2 publications a
	// client may have in flight towards the broker (MQTT-5.0 §3.2.2.3.3).
	DefaultReceiveMaximum = 1024
	// DefaultTopicAliasMaximum is the highest Topic Alias the broker accepts
	// from a client (MQTT-5.0 §3.1.2.11.5).
	DefaultTopicAliasMaximum = 64
	// DefaultMaximumPacketSize caps an inbound packet at 64 MiB
	// (MQTT-5.0 §3.2.2.3.6).
	DefaultMaximumPacketSize = 64 << 20
	// DefaultConnectTimeout bounds how long a new connection may stay silent
	// before sending its CONNECT (MQTT-5.0 §3.1.4, step 1).
	DefaultConnectTimeout = 10 * time.Second
	// DefaultMaxSessionExpiry caps the Session Expiry Interval a client may
	// request. The broker returns the capped value in CONNACK, as
	// MQTT-5.0 §3.2.2.3.2 provides for.
	DefaultMaxSessionExpiry = time.Hour
)

Defaults applied to the zero value of the corresponding Options field.

const MaxPersistentClientIDLen = 128

MaxPersistentClientIDLen caps the Client Identifier a broker with Options.PersistentSessions accepts. The identifier is encoded into a JetStream key-value key, and a key is part of a NATS subject, so it cannot be the 65535 bytes MQTT permits. A longer one is refused with 0x85 (Client Identifier not valid), which MQTT-5.0 §3.1.3.1 provides for: a server states the identifiers it accepts, and 128 bytes is more than five times the 23 that every MQTT server must accept.

const ReauthorizeForwardWait = time.Second

ReauthorizeForwardWait is how long Broker.Reauthorize waits for another broker to say it holds a session this one does not. Every broker answers at once, so the wait is spent only when no broker holds it.

Variables

var ErrInvalidOptions = errors.New("natsmqtt5: invalid options")

ErrInvalidOptions reports an Options value the broker cannot run with.

Functions

func Ptr

func Ptr[T any](v T) *T

Ptr returns a pointer to v. The optional Options fields are pointers so that "not set" is distinguishable from a meaningful zero, and this saves callers declaring a variable to take the address of.

natsmqtt5.Options{MaximumQoS: natsmqtt5.Ptr(uint8(1))}

Types

type Action

type Action int

Action is an operation an Authorizer decides on.

const (
	// ActionPublish is a client publishing to a Topic Name.
	ActionPublish Action = iota
	// ActionSubscribe is a client subscribing to a Topic Filter.
	ActionSubscribe
)

func (Action) String

func (a Action) String() string

type AuthRequest

type AuthRequest struct {
	// ClientID is the Client Identifier from CONNECT, before the broker
	// assigns one for an empty value.
	ClientID string
	// Username is empty when the User Name Flag was not set.
	Username string
	// Password is nil when the Password Flag was not set. MQTT v5 permits a
	// password with no username (MQTT-5.0 §3.1.2.9).
	Password []byte
	// RemoteAddr is the client's network address.
	RemoteAddr net.Addr
	// TLS is the connection's TLS state, or nil for a plaintext connection.
	// Use it for client-certificate authentication.
	TLS *tls.ConnectionState
	// Properties are the CONNECT properties, including any User Properties.
	Properties *packet.Properties
}

AuthRequest describes a connection attempt for an Authenticator.

type AuthResult

type AuthResult struct {
	// Identity, when non-empty, names the authenticated principal. It is
	// passed to the Authorizer and included in log records; it does not change
	// the Client Identifier.
	Identity string
	// AssignClientID, when non-empty, overrides the Client Identifier for the
	// session and is returned to the client as the Assigned Client Identifier
	// [MQTT-3.1.3-7].
	AssignClientID string
}

AuthResult is an Authenticator's verdict.

type Authenticator

type Authenticator interface {
	Authenticate(ctx context.Context, req *AuthRequest) (*AuthResult, error)
}

Authenticator decides whether a CONNECT is accepted.

Returning a nil error accepts the connection. Returning an error rejects it; wrap or return a *ConnectError to choose the CONNACK Reason Code, otherwise the broker answers 0x87 (Not authorized) and does not disclose the reason.

The Client Identifier is the session key

A Session is identified by its Client Identifier alone (MQTT-5.0 §4.1), so every connection an implementation accepts under a given identifier inherits that session: its subscriptions, its unacknowledged messages and its QoS 2 receive state. An implementation that authenticates a principal but lets it choose any ClientID therefore lets it read another principal's session.

So an implementation must either refuse a CONNECT whose AuthRequest.ClientID is not one the authenticated principal owns, or ignore it and return an AuthResult.AssignClientID derived from the principal. Nothing downstream can make up for this: the Authorizer decides topics, not sessions.

The broker re-runs the Authorizer over a resumed session's Topic Filters, so a permission narrowed between two connections takes effect on the second one rather than when the session expires. That bounds the damage; it does not replace binding the identifier, because the filters a principal is allowed are usually not what distinguishes it from the session's rightful owner.

type AuthenticatorFunc

type AuthenticatorFunc func(ctx context.Context, req *AuthRequest) (*AuthResult, error)

AuthenticatorFunc adapts a function to the Authenticator interface.

func (AuthenticatorFunc) Authenticate

func (f AuthenticatorFunc) Authenticate(ctx context.Context, req *AuthRequest) (*AuthResult, error)

Authenticate calls f.

type Authorizer

type Authorizer interface {
	Authorize(ctx context.Context, req *AuthzRequest) error
}

Authorizer decides whether a session may publish to or subscribe to a topic. A nil error permits the operation.

A denied publish is answered with 0x87 (Not authorized) in the PUBACK or PUBREC, or dropped silently at QoS 0, as MQTT-5.0 §3.3.4 requires. A denied SUBSCRIBE is answered with 0x87 in the SUBACK for that filter. A denied Will Message, asked about with AuthzRequest.Will set, refuses the CONNECT with 0x87 in the CONNACK.

It is also called while a session is being resumed

A session is keyed by its Client Identifier alone, so the broker puts every filter of a resumed session back past the Authorizer before answering the CONNECT — once per live filter, with AuthzRequest.Resume set. That is what makes a permission narrowed between two connections take effect on the second one rather than when the session expires.

Two consequences an implementation has to plan for:

A slow Authorizer delays the CONNACK in proportion to how many filters the session holds. The ctx carries no deadline of the broker's making.

A denial there is silent and it is final. There is no per-filter Reason Code in a CONNACK, so the filter is torn down, logged at warn level, and the client is told SessionPresent: true with no indication that anything is missing. With Options.PersistentSessions the reduced set is written straight back to the durable record. An error returned because the policy store could not be reached is therefore indistinguishable from a decision to deny, and costs the client its subscription permanently — an implementation that cannot decide should say so by failing the connection from the Authenticator, not by returning an error from here.

type AuthorizerFunc

type AuthorizerFunc func(ctx context.Context, req *AuthzRequest) error

AuthorizerFunc adapts a function to the Authorizer interface.

func (AuthorizerFunc) Authorize

func (f AuthorizerFunc) Authorize(ctx context.Context, req *AuthzRequest) error

Authorize calls f.

type AuthzRequest

type AuthzRequest struct {
	Action Action
	// Resume distinguishes a filter being re-checked because a session is
	// being resumed from one the client has just asked for in a SUBSCRIBE. It
	// is only ever set with ActionSubscribe.
	//
	// The two are otherwise identical, and an implementation that audit-logs,
	// meters or rate-limits subscriptions needs to tell them apart: the client
	// performed no action, and the same filters come back on every reconnect.
	// It is also the cheap way out of the latency this check adds to a CONNACK
	// — an implementation may answer a resume from a cache it would not trust
	// for a fresh SUBSCRIBE.
	Resume bool
	// Will marks the Will Message of a CONNECT, asked about as an
	// ActionPublish before the connection is accepted. It is only ever set with
	// ActionPublish.
	//
	// The broker publishes a Will on the client's behalf after the client has
	// gone, so this is the only chance to refuse it: a denial refuses the
	// CONNECT with 0x87 (Not authorized), and the Will is not re-checked when it
	// fires. An implementation that audit-logs publishes can use the flag to
	// tell a Will — which may never be sent — from a PUBLISH the client made.
	Will bool
	// ClientID is the session's Client Identifier, after any assignment.
	ClientID string
	// Identity is the AuthResult.Identity from authentication, if any.
	Identity string
	// Username is the User Name from CONNECT, if any.
	Username string
	// Topic is the Topic Name for ActionPublish or the Topic Filter, shared
	// prefix included, for ActionSubscribe.
	Topic string
	// QoS is the requested QoS.
	QoS packet.QoS
	// Retain is set for a publish with the RETAIN flag.
	Retain bool
}

AuthzRequest describes an operation for an Authorizer.

type Broker

type Broker struct {
	// contains filtered or unexported fields
}

Broker is an MQTT v5 broker backed by a NATS server. Create one with New, run it with Serve, and release its resources with Close.

A Broker is safe for concurrent use.

func New

func New(opts Options) (*Broker, error)

New connects to NATS, provisions the JetStream assets the broker needs and binds the MQTT listener. It does not accept connections until Serve is called, so a caller can read ListenAddr first.

The returned Broker must be closed with Close.

func NewWithContext

func NewWithContext(ctx context.Context, opts Options) (*Broker, error)

NewWithContext is New with a context bounding the NATS connection and JetStream setup.

func (*Broker) Close

func (b *Broker) Close() error

Close stops the broker: it closes the listener, disconnects every client with 0x8B (Server shutting down) and releases the NATS resources it owns. A NATS connection passed in through Options.Conn is left open.

func (*Broker) ListenAddr

func (b *Broker) ListenAddr() net.Addr

ListenAddr reports the address the broker accepts MQTT connections on. It is valid as soon as New returns, which lets a test bind :0 and discover the port.

func (*Broker) NATS

func (b *Broker) NATS() *nats.Conn

NATS returns the NATS connection the broker uses, so an embedding application can share it rather than opening a second one.

func (*Broker) Reauthorize

func (b *Broker) Reauthorize(ctx context.Context, clientID string) error

Reauthorize re-runs the Authorizer over everything the session for clientID holds: each subscription, as at a resume, and its Will Message, as at CONNECT. Call it when a principal's permissions change; the broker does not poll a policy store it knows nothing about, and does not ask the Authorizer on every delivery.

A denied subscription is removed. It stops delivering at once, including messages already queued for the connection, and the messages it had in flight are withdrawn, so a later resume does not resend them; their acknowledgements are still accepted. A denied Will is discarded, including one already waiting out its Will Delay Interval. The connection itself is left alone: it is still allowed, and a DISCONNECT would publish the very Will being revoked. The client is not told which filters it lost; MQTT has no packet for that.

The Authorizer sees a subscription with AuthzRequest.Resume set and a Will with Will set, as it does for those checks elsewhere. As there, any error it returns is a denial, so an Authorizer that cannot reach its policy should not be asked to re-check a fleet during an outage.

Any broker will do. One that does not hold the session asks the others sharing its NATS connection and SubjectPrefix, and the one holding it runs the sweep and answers. When none holds it after ReauthorizeForwardWait (or when ctx ends first), it returns nil: a session held nowhere, only in the session store, is re-checked when it is next resumed. It returns ctx's error, having changed nothing further, if ctx ends during this broker's own sweep, and the holder's error if its sweep failed.

func (*Broker) Serve

func (b *Broker) Serve(ctx context.Context) error

Serve accepts MQTT connections until ctx is cancelled or Close is called. It returns nil on a clean shutdown.

type ConnectError

type ConnectError struct {
	Code   packet.ReasonCode
	Reason string
}

ConnectError carries the CONNACK Reason Code an Authenticator wants the broker to send. Any code of 0x80 or greater from MQTT-5.0 Table 3-1 is valid; common choices are BadUserNameOrPassword (0x86), NotAuthorized (0x87), Banned (0x8A) and ClientIdentifierNotValid (0x85).

func (*ConnectError) Error

func (e *ConnectError) Error() string

type Options

type Options struct {
	// NATSURL is the NATS server to use. Defaults to nats.DefaultURL. Ignored
	// when Conn is set.
	NATSURL string
	// NATSOptions are passed to nats.Connect for credentials, TLS and so on.
	// Ignored when Conn is set.
	NATSOptions []nats.Option
	// Conn is an already-established NATS connection to use instead of
	// dialling. The broker does not close a connection it did not open.
	Conn *nats.Conn

	// Listen is the TCP address to accept MQTT connections on. Defaults to
	// DefaultListen. Ignored when Listener is set.
	Listen string
	// Listener is an already-open listener to accept on instead of binding
	// Listen. The broker closes it on shutdown.
	Listener net.Listener
	// TLSConfig, when non-nil, wraps the listener in TLS.
	TLSConfig *tls.Config

	// SubjectPrefix namespaces every NATS subject the broker publishes MQTT
	// traffic on, keeping it clear of NATS system subjects. Defaults to
	// DefaultSubjectPrefix. Set it to the same value on every broker that
	// should share traffic.
	SubjectPrefix string
	// StreamPrefix names the JetStream assets the broker creates. Defaults to
	// DefaultStreamPrefix. Brokers sharing a SubjectPrefix must share this.
	StreamPrefix string
	// DisableRetained turns off retained-message support, so no JetStream
	// assets are created and CONNACK advertises Retain Available = 0
	// (MQTT-5.0 §3.2.2.3.5). Use it against a NATS server without JetStream.
	DisableRetained bool
	// RetainedStorage selects file or memory storage for the retained-message
	// stream. Defaults to file storage.
	RetainedStorage jetstream.StorageType
	// RetainedReplicas is the JetStream replica count for the retained-message
	// stream. Defaults to 1.
	RetainedReplicas int

	// PersistentSessions keeps session state in a JetStream key-value bucket
	// instead of in the broker process, so a client reconnecting with Clean
	// Start 0 resumes its subscriptions after a broker restart and on a broker
	// that has never seen it. Brokers sharing a StreamPrefix share the bucket
	// and arbitrate ownership between themselves, so exactly one of them serves
	// a given Client Identifier [MQTT-3.1.4-3].
	//
	// It needs JetStream, and it caps the Client Identifier at
	// MaxPersistentClientIDLen bytes.
	//
	// A resumed session carries its subscription set and nothing else.
	// In-flight QoS 1 and QoS 2 messages are not stored, because resending
	// them needs an offline queue the broker does not have; the Will Message is
	// not stored because it belongs to the network connection rather than to
	// the session (MQTT-5.0 §3.1.2.5). ROADMAP.md has both.
	PersistentSessions bool
	// SessionStorage selects file or memory storage for the session bucket.
	// Defaults to file storage. Memory storage makes sessions survive a broker
	// restart but not a NATS restart.
	SessionStorage jetstream.StorageType
	// SessionReplicas is the JetStream replica count for the session bucket.
	// Defaults to 1.
	SessionReplicas int

	// Authenticator, when non-nil, decides whether a CONNECT is accepted. When
	// nil every connection is accepted, which is only appropriate on a trusted
	// network.
	Authenticator Authenticator
	// Authorizer, when non-nil, decides whether a session may publish to or
	// subscribe to a topic. When nil everything is permitted.
	Authorizer Authorizer

	// ReceiveMaximum is advertised in CONNACK and bounds the client's in-flight
	// QoS 1 and QoS 2 publications. Defaults to DefaultReceiveMaximum.
	ReceiveMaximum uint16
	// MaximumQoS is the highest QoS the broker accepts in a PUBLISH and grants
	// on a SUBSCRIBE, advertised in CONNACK [MQTT-3.2.2-9]. Values above 2 are
	// clamped. Leave it nil for the default of 2; it is a pointer so that
	// Ptr(uint8(0)) can express a broker that only does QoS 0, which the zero
	// value would otherwise make unreachable.
	MaximumQoS *uint8
	// TopicAliasMaximum is advertised in CONNACK. Zero disables inbound topic
	// aliases; leave it unset for DefaultTopicAliasMaximum.
	TopicAliasMaximum *uint16
	// MaximumPacketSize caps an inbound packet. Defaults to
	// DefaultMaximumPacketSize.
	MaximumPacketSize uint32
	// ServerKeepAlive, when non-zero, overrides the client's requested Keep
	// Alive and is returned in CONNACK [MQTT-3.1.2-21].
	ServerKeepAlive uint16
	// ConnectTimeout bounds the wait for a CONNECT on a new connection.
	// Defaults to DefaultConnectTimeout.
	ConnectTimeout time.Duration
	// MaxSessionExpiry caps a requested Session Expiry Interval. Defaults to
	// DefaultMaxSessionExpiry.
	MaxSessionExpiry time.Duration

	// Logger receives broker events. Defaults to slog.Default().
	Logger *slog.Logger
}

Options configures a Broker. The zero value is usable: it connects to nats.DefaultURL and listens on :1883.

Fields are only ever added to this struct, never removed or repurposed, so a composite literal with field names keeps compiling across versions.