Embedding the broker

mercurio-server is the broker as a library: run it inside your own program, with your own authentication, access control and storage.

Adding it

[dependencies]
mercurio-server = "0.5"
mercurio-storage = { version = "0.3", features = ["sqlite"] }
tokio = { version = "1", features = ["full"] }

Running a broker

use mercurio_server::server::{ListenerConfig, Server, Transport};
use tokio::net::TcpListener;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let listener = TcpListener::bind("0.0.0.0:1883").await?;
    Server::builder()
        .listener(ListenerConfig::new(listener, Transport::Mqtt))
        .run(tokio::signal::ctrl_c())
        .await?;
    Ok(())
}

The broker runs until the future given to run completes, then shuts down gracefully. Server::builder() starts a ServerBuilder with no listener, its state in memory and the default settings:

build() checks the configuration and restores what storage holds, and returns a Server that serves nobody yet; run(shutdown) on the builder builds and serves at once. A broker needs no network listener: with none, it serves the in-process clients of Server::local_connector.

A configuration that can't be used fails build() before anyone is served, with ServerError::Config. Its ConfigError names the listener, by its position among those given and its address, and its kind says which setting is wrong, such as ConfigErrorKind::MountPointWildcard for a mount point with a wildcard. Certificates and keys that can't be read are a TlsError, from Tls::new.

When run returns, every task the broker started has ended: connections still open after a few seconds, session expiry timers, wills waiting for their delay and the $SYS publisher are stopped, and what they were waiting for is taken up from storage by the next broker. The broker no longer holds its storage, so the same SQLite file can be opened again in the same process. An Admin handle kept from Server::admin still holds it until dropped.

Configuration

ServerConfig holds the broker-wide settings, which listeners use unless they have their own:

use mercurio_server::{
    server::{Limits, ListenerConfig, Server, ServerConfig, Transport},
    tls::{Tls, TlsConfig},
};
use tokio::net::TcpListener;

let tls = Tls::new(TlsConfig::new("server.crt", "server.key"))?;
let plain = ListenerConfig::new(TcpListener::bind("0.0.0.0:1883").await?, Transport::Mqtt);
let secure = ListenerConfig::new(TcpListener::bind("0.0.0.0:8883").await?, Transport::Mqtt)
    .with_tls(tls.clone());

let config = ServerConfig::default()
    .with_limits(Limits::default().with_max_connections(10_000));
let server = Server::builder()
    .config(config)
    .listener(plain)
    .listener(secure)
    .build()
    .await?;

TLS

Tls::new reads the certificate, key, CA and CRL files a TlsConfig names, and fails if they can't be used. Tls::reload reads them again, for example after a certificate is renewed: connections from then on get the new files, and those already up keep theirs. A reload that fails returns the error and keeps the previous files. Clones of a Tls share it, so one reload reaches every listener given a clone.

For settings TlsConfig has no field for, such as ALPN, session resumption, a certificate resolver or a client verifier of your own, build a rustls::ServerConfig and serve it with Tls::from_rustls(config, username_from_certificate). Reloading is then up to you: Tls::reload returns an error, and Tls::set_server_config swaps in a new configuration for later connections. With a CertificateName, clients that present a certificate are named by it; whether they must present one is up to your client verifier.

Configuration types are built from their default, or from a constructor such as ListenerConfig::new, then changed with their with_* methods, one per field, or field by field: new fields may appear in minor releases, so they can't be written out as struct literals.

Plugins

One Plugin trait covers authentication, MQTT 5 enhanced authentication, topic access control, message hooks, lifecycle events, periodic work and $CONTROL commands. A plugin implements the hooks it needs; every other one does nothing. The broker runs its plugins as a chain, in the order configured:

• Plugin::authenticate and Plugin::authorize: the first Decision::Allow or Decision::Deny decides, and Decision::Defer asks the next plugin. Credentials are checked whenever a client sends them, even with require_auth off, so a plugin's Deny always refuses them. When every plugin defers, a client with credentials is refused if require_auth is on and let in unverified if not, an anonymous one follows allow_anonymous, and access is allowed.
• Plugin::authorize_control: decides who may publish to $CONTROL/..., which administers the broker, in place of authorize. It is a separate question, so a plugin that allows every topic doesn't make every client an administrator. It chains like authorize, but when every plugin defers, the client is refused. Requests to $CONTROL/broker/v1 are asked about that topic, then about each command as $CONTROL/broker/v1/<command>. Subscribing to and receiving $CONTROL topics stay with authorize, as answers go only to the client that asked.
• Plugin::on_message: every plugin, each seeing the changes of the ones before; the first MessageAction::Drop drops the message. Changes are trusted: a new topic isn't checked against access control, and a changed retain or qos isn't checked against the retain and QoS limits. The exception is a message moved to a different topic that starts with $, such as $SYS/... or $CONTROL/...: it is dropped, still acknowledged, with a warning in the log.
• Plugin::on_event and Plugin::on_tick: every plugin.

Each hook call has a time limit, ServerConfig::plugin_timeout. A hook that runs out of time or panics denies, for authentication and access control, and drops, for messages. A listener can have its own chain in ListenerConfig::plugins; ListenerConfig::auth only sets require_auth and allow_anonymous for it.

use std::sync::Arc;

use mercurio_server::{
    plugin::{ready, Access, BoxFuture, ClientInfo, Credentials, Decision, Plugin},
    server::{AuthConfig, ServerConfig},
};

/// Lets each user use the topics under their own name.
struct OwnTopics;

impl Plugin for OwnTopics {
    fn name(&self) -> &str {
        "own-topics"
    }

    fn authenticate<'a>(
        &'a self,
        _: &'a ClientInfo,
        credentials: &'a Credentials<'a>,
    ) -> BoxFuture<'a, Decision> {
        // A real plugin would check a database here, and could await it
        let known = matches!(
            (credentials.username, credentials.password),
            ("alice", b"alice-secret") | ("bob", b"bob-secret")
        );
        ready(if known { Decision::Allow } else { Decision::Deny })
    }

    fn authorize<'a>(
        &'a self,
        client: &'a ClientInfo,
        topic: &'a str,
        _: Access,
    ) -> BoxFuture<'a, Decision> {
        let own = client
            .username
            .as_deref()
            .is_some_and(|user| topic.starts_with(&format!("{user}/")));
        ready(if own { Decision::Allow } else { Decision::Deny })
    }
}

let config = ServerConfig::default()
    .with_auth(AuthConfig::default().with_require_auth(true))
    .with_plugins(vec![Arc::new(OwnTopics)]);

• Access is Write to publish, Subscribe to subscribe, with the topic filter, and Read to receive a message.
• ClientInfo::username is only set for users the broker verified. client_id is empty in authenticate when the client asked the server to assign one.
• authorize runs for every message, so keep it fast; a hook that is ready at once, as above, costs no timer. Cache answers that need I/O.
• Plugin::on_control handles publishes to $CONTROL/<name>/.... Its answer goes only to the client that sent the request: on the request's Response Topic, with its Correlation Data, or on $CONTROL/<name>/<path>/response without one.
• A client publishing to $CONTROL/<name>/x/response reaches Plugin::on_control with path x/response; plugins that only answer requests should ignore such paths.
• auth::SimpleCredentialValidator checks a list of users held in plaintext, for tests and demos; it leaves users it doesn't hold to the next plugin. acl::AclFile::parse reads mosquitto's ACL format, if you want it from your own code.

Enhanced authentication

A plugin takes part in MQTT 5 enhanced authentication by returning an AuthExchange from Plugin::begin_auth for the methods it knows. begin_auth gets the client: during CONNECT its user name is the one the client claims, if any; on re-authentication it is the verified user. The broker calls step once per message from the client; each step returns AuthResult::Continue(challenge), Success { username, data } or Failed { reason, reason_string }, where reason_string optionally tells the client why in the CONNACK, or in the DISCONNECT that ends a failed re-authentication, unless the client sent Request Problem Information 0 or the text exceeds its Maximum Packet Size; it must not reveal, for example, whether a user exists. AuthResult::failed(reason) fails without one. data, such as SCRAM's final server message, goes to the client as Authentication Data in the CONNACK, or in the AUTH that ends a re-authentication. auth::PlainAuth is a complete example, with passwords held in plaintext for tests and demos: add it to ServerConfig::plugins to accept the PLAIN method.

auth::ScramAuth implements SCRAM-SHA-256 over credentials you store. Keep each user's StoredCredentials, made with StoredCredentials::generate(password, iterations) when the password is set, and look them up through ScramCredentialStore; the plain password is never kept. Return None for a user who doesn't exist or may not log in as that client: the broker then answers with a salt derived from the user name and fails the exchange at the proof, so the client can't tell. Give decoy_iterations the count your credentials use, mercurio_scram::DEFAULT_ITERATIONS unless you chose another.

use std::{collections::HashMap, sync::Arc};
use mercurio_server::auth::{ScramAuth, StoredCredentials};

let mut users = HashMap::new();
users.insert(
    "alice".to_string(),
    StoredCredentials::generate("secret", mercurio_scram::DEFAULT_ITERATIONS)?,
);
config.plugins.push(Arc::new(ScramAuth::new(Arc::new(users))));

A HashMap<String, StoredCredentials> is a store, for tests and demos; implement ScramCredentialStore on your own type for anything else. The mercurio-scram crate holds the messages and key derivation the broker and the client share.

Storage

use std::sync::Arc;

use mercurio_server::server::{ListenerConfig, Server, Transport};
use mercurio_storage::sqlite::SqliteStore;

let storage = Arc::new(SqliteStore::new("/var/lib/myapp/mqtt.db")?);
Server::builder()
    .storage(storage)
    .listener(ListenerConfig::new(listener, Transport::Mqtt))
    .run(shutdown)
    .await?;

Any type implementing the storage traits of mercurio-storage (sessions, retained messages, wills, in-flight messages and queues) can replace SQLite; postgres::PostgresStore::connect_exclusive(url, &options, wait), behind the postgres feature, keeps it in PostgreSQL; postgres::ConnectOptions sets the CA certificate and the size of its connection pool. It returns a DatabaseLock along with the store: only its holder may write to the database, and its lost() completes when another broker may have taken over, the moment to stop serving.

In-process clients

A LocalConnector opens connections to the broker in memory, without TCP. Combined with mercurio-client's connector option, it lets your program publish and subscribe on its own broker. These connections are trusted: authentication and access control don't apply to them.

use mercurio_client::{ConnectOptions, MqttClient, QoS};
use mercurio_server::server::Server;

let server = Server::builder().build().await?;
let connector = server.local_connector();
tokio::spawn(server.run(std::future::pending::<()>()));

let options = ConnectOptions::new("localhost", 0)
    .client_id("internal")
    .connector(move || std::future::ready(connector.connect()));
let client = MqttClient::connect(options).await?;
client.publish("app/started", "1", QoS::AtLeastOnce).await?;

Code that must hold a connector before the broker is built, as the daemon's plugins and bridges do, creates the pair with LocalConnector::new() and gives the LocalListener to ServerBuilder::local_listener. The daemon's bridges and dynamic security work this way.

Administering it

Server separates setting a broker up from serving it. In between it hands out an Admin, which runs the commands of the admin API as async methods, before and while the broker serves. ServerBuilder::run is a shortcut for ServerBuilder::build followed by Server::run.

use mercurio_server::{
    admin::ClientQuery,
    server::{ListenerConfig, Server, Transport},
};

let server = Server::builder()
    .listener(ListenerConfig::new(listener, Transport::Mqtt))
    .build()
    .await?;
let admin = server.admin();
tokio::spawn(server.run(tokio::signal::ctrl_c()));

let mut query = ClientQuery::default();
query.connected = Some(true);
for client in admin.clients(&query).await.items {
    println!("{} {:?}", client.client_id, client.address);
}

• Methods return plain structs. Failures are an AdminError, whose code is the protocol's.
• Each mount point has its own client ids, so the methods acting on one client, client, disconnect and delete_session, take a SessionKey: the client id and the mount point, None for a listener without one. Admin::session_key finds it from a client id as the protocol's mountpoint argument does, and ClientSummary carries both.
• The broker serves the same commands over MQTT on $CONTROL/broker/v1, to clients access control allows. ServerConfig::admin_extension takes an AdminExtension for commands the broker doesn't know: mercuriod adds its bridge, configuration, log level and reload commands this way. Without one, those answer unavailable.
• A ListenerConfig on a Unix socket can be trusted: its clients skip authentication and access control, and run every command. Give it a peer_filter to choose clients by their user and group ids, as mercuriod's admin socket does, or keep the socket's permissions tight. ServerBuilder::build refuses a trusted listener on TCP, with its own authentication or plugins, with proxy_protocol or serving WebSocket.

Logs

The broker logs through tracing. Install any subscriber, such as tracing_subscriber::fmt().init(). Connection and disconnection messages use the target mercurio::connections, the constant mercurio_server::CONNECTIONS, so they can be filtered alone.