comqttauth

README

comqttauth

Authentication and authorization storage for comqtt-derived MQTT brokers, with backends for file (YAML), Redis, MySQL, and PostgreSQL. Wire-compatible with stock comqtt: data written by this library is readable by comqtt's upstream plugin/auth/* hooks, so rollback is lossless.

comqttauth is the storage layer extracted from debsahu/comqtt-dashboard. It ships as a Go library and as a drop-in container image of the comqtt broker with the regex-ACL hook pre-installed.

Status

• v0.1.0 - storage layer only (Backend interface + file/redis/mysql/postgres implementations).
• v0.2.0 - runtime mqtt.Hook for direct use in comqtt-based brokers, plus regex-based authorization rules.
• v0.2.1 - runnable integration examples for all four backends.
• v0.3.0 (latest) - comqttauth-broker container image (ghcr.io/debsahu/comqttauth-broker) shipping /comqtt and /comqtt-cluster binaries that mirror upstream cmd/single and cmd/cluster with comqttauth.Hook pre-installed. Drop-in image override for the upstream Helm chart.

Install

# library
go get github.com/debsahu/comqttauth@v0.3.0

# container image (single + cluster binaries in one image)
docker pull ghcr.io/debsahu/comqttauth-broker:v0.3.0

Quick start

package main

import (
    "context"
    "log"

    "github.com/debsahu/comqttauth"
)

func main() {
    cfg := comqttauth.Config{
        Kind:     "file",
        Mode:     comqttauth.ModeUsername,
        HashType: comqttauth.HashBcrypt,
        File: &comqttauth.FileConfig{
            Path: "/etc/comqtt/ledger.yml",
        },
    }
    be, err := comqttauth.New(cfg)
    if err != nil {
        log.Fatal(err)
    }
    defer be.Close()

    ctx := context.Background()
    if err := be.PutUser(ctx, comqttauth.User{Subject: "alice", Allow: true}, "p4ssw0rd"); err != nil {
        log.Fatal(err)
    }

    users, _ := be.Users(ctx)
    for _, u := range users {
        log.Printf("user=%s allow=%v", u.Subject, u.Allow)
    }
}

For runnable, end-to-end integrations (comqtt broker + comqttauth, one per backend) see examples/.

Deploy with the comqtt Helm chart

The container image is a drop-in for the upstream comqtt chart. Override image.repository / image.tag and keep everything else stock:

# values.yaml fragment for `helm install ... wind-c/comqtt`
image:
  repository: ghcr.io/debsahu/comqttauth-broker
  tag: v0.3.0

# rest is plain comqtt chart config:
config:
  auth:
    way: 1            # 1 = username/password
    datasource: 1     # 1 = redis (also: 2 mysql, 3 postgresql)
    conf-path: /etc/comqtt/auth-redis.yml

The image ships two binaries with the names the chart expects:

Mode	Binary	Selected by
mode: single	/comqtt	Deployment manifest command: ["/comqtt"]
mode: cluster	/comqtt-cluster	StatefulSet entrypoint script exec /comqtt-cluster

Both binaries are byte-for-byte equivalent to upstream cmd/single and cmd/cluster except that after the upstream auth hook is registered, comqttauth.Hook is also registered against the same backend — so connection auth runs through upstream (OnConnectAuthenticate) and regex ACL runs through this library (OnACLCheck), AND-combined by the broker. Regex rules supported for redis / mysql / postgresql auth datasources.

For Gateway API routing (TCPRoute for MQTT, HTTPRoute for REST + /metrics), enable gateway.enabled in chart values. See the chart's values.yaml for the full surface.

Backends

Backend	Wire format	Notes
file	comqtt's auth.Ledger YAML	Reads/writes the same YAML schema the comqtt file-backed auth.Hook consumes. Live-reload behavior depends on whether the consumer rebuilds the ledger between requests.
redis	HASH comqtt:auth field=subject value=JSON, HASH comqtt:acl:<subject> field=topic value=access	Drop-in replacement for comqtt's plugin/auth/redis storage. Same keys.
mysql	Table auth(username, password, allow), table acl(username, topic, access)	Schema mirrors plugin/auth/mysql/testdata/init.sql. Table and column names configurable via Config.SQL.
postgres	Same as MySQL	Driver: pgx/v5/stdlib. Column names same.

Configuration

The Config struct selects the backend and its connection parameters. See config.go for the full schema. Common shape:

comqttauth.Config{
    Kind:     "redis" | "mysql" | "postgres" | "file",
    Mode:     ModeUsername | ModeClientID | ModeAnonymous,
    HashType: HashBcrypt | HashSHA256 | HashNone,
    // Exactly one of File / Redis / SQL non-nil, matching Kind.
}

Relationship to comqtt and comqtt-dashboard

• comqtt (wind-c/comqtt) is the upstream MQTT broker. This library depends on its mqtt/hooks/auth and plugin/auth packages for shared wire-format types (auth.Ledger, auth.AuthRule, pa.HashType). Data written by comqttauth is identical in wire format to data written by comqtt's own auth plugins.
• comqtt-dashboard (debsahu/comqtt-dashboard) is the web UI that uses this library. Operators get an Authentication and Authorization page driven by comqttauth.Backend. The dashboard is the primary consumer; this library is reusable from other comqtt-derived deployments too.

Integration tests

Unit tests run on every PR via standard go test. SQL backends ship integration tests behind the integration_sql build tag, exercised against real databases:

# MySQL
docker run -d --rm --name comqttauth-mysql -p 3306:3306 \
    -e MYSQL_ROOT_PASSWORD=root -e MYSQL_DATABASE=comqtt mysql:8
# (wait ~15s for mysql)
COMQTTAUTH_TEST_MYSQL_DSN='root:root@tcp(127.0.0.1:3306)/comqtt?parseTime=true' \
    go test -race -tags integration_sql ./...

# Postgres
docker run -d --rm --name comqttauth-pg -p 5432:5432 \
    -e POSTGRES_USER=postgres -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=comqtt postgres:16
COMQTTAUTH_TEST_POSTGRES_DSN='postgres://postgres:postgres@127.0.0.1:5432/comqtt?sslmode=disable' \
    go test -race -tags integration_sql ./...

CI runs both against GitHub Actions services: containers on every PR.

License

MIT - see LICENSE.

Documentation

Overview

Package comqttauth manages comqtt broker authentication and authorization state from the dashboard. Each Backend implementation reads and writes the same on-disk/on-wire shape comqtt's corresponding plugin/auth/* runtime hooks already consume, so a change made through the dashboard is visible to the running broker on its next lookup without any synchronization layer or process restart.

Backends supported in v0.3.0:

• file: auth.Ledger YAML file (built-in comqtt hook)
• redis: plugin/auth/redis HSET/HGETALL key shape
• mysql: plugin/auth/mysql configurable auth/acl tables
• postgres: plugin/auth/postgresql configurable auth/acl tables

The dashboard's Auth and ACL pages consume Backend through a single interface. The active backend is selected by the cmd-binary based on cfg.Auth.Datasource and constructed via factory.New().

This package does not manage the dashboard's own operator credentials (admin/viewer roles) - those live in dashboard/auth and are unrelated to MQTT-broker auth.

Index

• Variables
• type ACLRule
• type ACLTable
• type Access
  • func (a Access) String() string
• type AuthMode
  • func (m AuthMode) String() string
• type AuthTable
• type Backend
  • func New(cfg Config) (Backend, error)
• type Config
• type FileConfig
• type HashType
  • func (h HashType) String() string
• type Hook
  • func (h *Hook) ID() string
  • func (h *Hook) Init(config any) error
  • func (h *Hook) OnACLCheck(cl *mqtt.Client, topic string, write bool) bool
  • func (h *Hook) Provides(b byte) bool
• type HookOptions
• type Permission
  • func ParsePermission(s string) (Permission, bool)
  • func (p Permission) String() string
• type RedisConfig
• type RegexRule
• type RuleAction
  • func ParseRuleAction(s string) (RuleAction, bool)
  • func (a RuleAction) String() string
• type SQLConfig
• type SubjectKind
  • func ParseSubjectKind(s string) (SubjectKind, bool)
  • func (s SubjectKind) String() string
• type User

Variables

var ErrConflict = errors.New("comqttauth: conflict")

ErrConflict is returned when a write would create a duplicate of a unique record (e.g. a second user with the same username).

var ErrInvalidRule = errors.New("comqttauth: invalid regex rule")

ErrInvalidRule is returned when a regex rule fails validation (invalid regex, invalid CIDR, empty topic patterns, etc).

var ErrNotFound = errors.New("comqttauth: not found")

ErrNotFound is returned when a requested record does not exist.

var ErrUnsupported = errors.New("comqttauth: operation not supported by this backend")

ErrUnsupported is returned by Backend methods when the operation is not supported by the active backend (e.g. user CRUD against an http-delegated auth backend, or wildcard-subject ACL queries against a key-value store that only indexes by exact subject).

Types

type ACLRule

type ACLRule struct {
	ID      string `json:"id"`
	Subject string `json:"subject"`
	Topic   string `json:"topic"`
	Access  Access `json:"access"`
}

ACLRule is the wire shape for one row in the ACL table. ID is backend- assigned and opaque to callers (a row id for SQL backends, the topic filter for redis, or the slice index for the file backend).

type ACLTable

type ACLTable struct {
	Table        string
	UserColumn   string
	TopicColumn  string
	AccessColumn string
}

ACLTable mirrors plugin/auth/{mysql,postgresql}.AclTable.

type Access

type Access uint8

Access matches comqtt's auth.Access constants. The dashboard exposes the four values directly in ACL editing forms; backends store them as the same byte values.

const (
	AccessDeny      Access = 0
	AccessRead      Access = 1 // subscribe only
	AccessWrite     Access = 2 // publish only
	AccessReadWrite Access = 3
)

func (Access) String

func (a Access) String() string

String returns the human-readable name used in ACL UI dropdowns.

type AuthMode

type AuthMode uint8

AuthMode selects what the lookup key is for auth and ACL records. It mirrors comqtt's auth.Access constants where it is overloaded as a mode indicator (0=anon, 1=username, 2=clientid).

const (
	ModeAnonymous AuthMode = iota
	ModeUsername
	ModeClientID
)

func (AuthMode) String

func (m AuthMode) String() string

String returns a stable lowercase mode name for templating and logs.

type AuthTable

type AuthTable struct {
	Table          string
	UserColumn     string
	PasswordColumn string
	AllowColumn    string
}

AuthTable mirrors plugin/auth/{mysql,postgresql}.AuthTable so the dashboard reads and writes the same physical table the broker reads.

type Backend

type Backend interface {
	// Kind returns a short stable name for the backend ("file", "redis",
	// "mysql", "postgres"). Used in admin UI badges and structured logging.
	Kind() string

	// Mode returns how user records are keyed (username or clientid). The UI
	// labels Subject columns accordingly.
	Mode() AuthMode

	// HashType returns the password hash algorithm this backend writes. The UI
	// labels password fields with this so operators know what they are
	// configuring.
	HashType() HashType

	// Users returns all user records. Returns an empty slice (not nil) when
	// no users exist.
	Users(ctx context.Context) ([]User, error)

	// GetUser returns the user with the given subject, or ErrNotFound.
	GetUser(ctx context.Context, subject string) (*User, error)

	// PutUser upserts a user record. plaintextPassword is hashed per
	// HashType() before write. Pass empty plaintextPassword to leave the
	// stored password unchanged on update; ErrNotFound on update of a missing
	// subject; the implementation distinguishes create vs update by existence
	// of the record.
	PutUser(ctx context.Context, u User, plaintextPassword string) error

	// DeleteUser removes the user with the given subject. Returns ErrNotFound
	// when no record matched.
	DeleteUser(ctx context.Context, subject string) error

	// Rules returns ACL rules for the given subject, or all rules when
	// subject is empty. Returns an empty slice when no rules match.
	Rules(ctx context.Context, subject string) ([]ACLRule, error)

	// PutRule inserts or updates an ACL rule. If r.ID is empty, the
	// implementation creates a new record and returns its assigned id.
	// Otherwise the existing record with that id is replaced; ErrNotFound if
	// the id does not exist.
	PutRule(ctx context.Context, r ACLRule) (string, error)

	// DeleteRule removes the ACL rule with the given id. ErrNotFound if no
	// record matched.
	DeleteRule(ctx context.Context, id string) error

	// RegexRules returns all stored regex rules ordered by Order ascending.
	// Returns an empty slice (not nil) when no rules are stored.
	RegexRules(ctx context.Context) ([]RegexRule, error)

	// PutRegexRule stores a rule. If r.ID is empty, the backend assigns
	// one and returns it. If r.ID is set, the backend upserts that id.
	PutRegexRule(ctx context.Context, r RegexRule) (id string, err error)

	// DeleteRegexRule removes the rule by id. Returns ErrNotFound when no
	// rule matches.
	DeleteRegexRule(ctx context.Context, id string) error

	// GetRegexSeeded reports whether the first-time auto-seed has happened
	// against this backend. Used by the dashboard to decide whether to seed
	// a default rule on startup.
	GetRegexSeeded(ctx context.Context) (bool, error)

	// SetRegexSeeded marks the backend as seeded. Idempotent.
	SetRegexSeeded(ctx context.Context) error

	// Close releases any underlying connections. Idempotent.
	Close() error
}

Backend is the unified interface the dashboard's Auth and ACL pages call. Each plugin/auth/* in comqtt has a corresponding implementation under this package's sub-trees.

All Context-taking methods are expected to honor cancellation and obey any deadline set by callers (typically a 3s default applied by the handler layer to keep dashboard responsiveness predictable).

func New

func New(cfg Config) (Backend, error)

New constructs the Backend for cfg.Kind. Returns an error if the per-Kind sub-config is missing or if Kind is unknown.

Chunks 2-5 of v0.3.0 fill in the per-backend constructors. Until then, each constructor returns a stub that responds to interface calls with ErrUnsupported so the surrounding dashboard handler scaffold can be wired independently.

type Config

type Config struct {
	Kind string

	// Mode is the lookup key used by user records.
	// ACLMode is the lookup key used by ACL records. Comqtt allows these to
	// differ (e.g. auth-by-username, ACL-by-clientid) so we keep them
	// independent.
	Mode    AuthMode
	ACLMode AuthMode

	// HashType is the algorithm used for password storage. HashKey is the
	// shared secret consumed by HMAC-* variants and ignored otherwise.
	HashType HashType
	HashKey  string

	// Per-backend specifics. The factory consults the field matching Kind
	// and returns an error if it is nil.
	File  *FileConfig
	Redis *RedisConfig
	SQL   *SQLConfig
}

Config is the disjoint-union of per-backend configurations. Exactly one of File/Redis/SQL is consulted, picked by Kind. Mode, ACLMode, HashType, and HashKey apply across all backends.

Kind is the canonical lowercase backend name: "file" | "redis" | "mysql" | "postgres". Pass-through from the broker's --auth-ds flag.

type FileConfig

type FileConfig struct {
	// Path is the YAML file. Must be writable by the dashboard process; the
	// file backend writes via atomic rename so a partial flush never makes
	// the broker see half a config.
	Path string
}

FileConfig points at a YAML ledger file matching the shape consumed by comqtt's built-in mqtt/hooks/auth.Hook running in LedgerMode.

type HashType

type HashType uint8

HashType identifies which hashing algorithm a backend uses for stored passwords. Values match comqtt's plugin/auth.HashType so a dashboard configured with the same hash as the broker writes hashes the broker can verify.

const (
	HashNone HashType = iota
	HashBcrypt
	HashMD5
	HashSHA1
	HashSHA256
	HashSHA512
	HashHmacSHA1
	HashHmacSHA256
	HashHmacSHA512
)

func (HashType) String

func (h HashType) String() string

String returns the human-readable name used in admin UI dropdowns.

type Hook

type Hook struct {
	mqtt.HookBase
	// contains filtered or unexported fields
}

Hook implements mqtt.Hook. It provides OnACLCheck only; OnConnectAuthenticate is left to comqtt's upstream plugin/auth/* hook in coexist mode.

On no-rule-match, OnACLCheck returns true so the AND-chain of hooks doesn't false-deny legitimate traffic that upstream has already allowed.

func (*Hook) ID

func (h *Hook) ID() string

ID returns the hook's identifier (used by mqtt.Server for logging).

func (*Hook) Init

func (h *Hook) Init(config any) error

Init configures the hook with the runtime Backend. Called by mqtt.Server at AddHook time. The passed config must be a *HookOptions.

func (*Hook) OnACLCheck

func (h *Hook) OnACLCheck(cl *mqtt.Client, topic string, write bool) bool

OnACLCheck evaluates regex rules and returns the verdict. Returns true (allow) when no rule matches, so the AND-combine with upstream's auth hook doesn't false-deny legitimate traffic.

func (*Hook) Provides

func (h *Hook) Provides(b byte) bool

Provides reports which mqtt.Hook methods this hook implements. Regex authorization is ACL-only; connection auth stays with upstream.

type HookOptions

type HookOptions struct {
	// Backend supplies the rules at runtime. Required.
	Backend Backend
	// Logger receives debug / info events. Optional; defaults to slog.Default().
	Logger *slog.Logger
}

HookOptions configures the runtime Hook. Pass via Server.AddHook.

type Permission

type Permission int8

Permission decides whether a rule's match results in allow or deny. Integer values are stable across versions; do not renumber.

const (
	PermissionAllow Permission = 1
	PermissionDeny  Permission = 0
)

func ParsePermission

func ParsePermission(s string) (Permission, bool)

ParsePermission parses the string form. Returns the zero value (deny) and false when the input is not recognised.

func (Permission) String

func (p Permission) String() string

String returns the YAML / JSON / human-readable form.

type RedisConfig

type RedisConfig struct {
	Addr     string
	Username string
	Password string
	DB       int

	// AuthKeyPrefix defaults to "comqtt:auth" when empty (matching
	// plugin/auth/redis.defaultAuthkeyPrefix).
	AuthKeyPrefix string
	// ACLKeyPrefix defaults to "comqtt:acl" when empty.
	ACLKeyPrefix string
}

RedisConfig matches the lookup shape of comqtt's plugin/auth/redis: a HASH at AuthKeyPrefix (default "comqtt:auth") and per-subject HASH at ACLKeyPrefix:<subject>.

type RegexRule

type RegexRule struct {
	ID             string // backend-assigned on Put
	Order          int    // lower runs first
	Permission     Permission
	SubjectKind    SubjectKind
	SubjectPattern string // empty (any) | "re:<regex>" | "cidr:<cidr>" | literal
	Action         RuleAction
	TopicPatterns  []string // length >= 1; each is an MQTT topic filter
}

RegexRule is one regex authorization rule. Rules are evaluated in ascending Order; first match wins (allow or deny). When no rule matches, the Hook default-allows so the AND-chain with comqtt's upstream auth plugin doesn't false-deny legitimate traffic.

QoS and Retain fields from the v0.4.0 design spec are deferred to v0.4.1 because comqtt's OnACLCheck signature does not carry packet QoS or retain flag.

type RuleAction

type RuleAction int8

RuleAction matches against the operation the client is attempting. Integer values are stable across versions.

const (
	ActionPub RuleAction = 0
	ActionSub RuleAction = 1
	ActionAll RuleAction = 2
)

func ParseRuleAction

func ParseRuleAction(s string) (RuleAction, bool)

ParseRuleAction parses the string form.

func (RuleAction) String

func (a RuleAction) String() string

String returns the YAML / JSON / human-readable form.

type SQLConfig

type SQLConfig struct {
	Driver string
	DSN    string

	Auth AuthTable
	ACL  ACLTable
}

SQLConfig is shared between MySQL and Postgres backends; Driver selects. Driver is one of "mysql" | "postgres". DSN is the database/sql connection string the chosen driver expects.

type SubjectKind

type SubjectKind int8

SubjectKind selects which client attribute the rule's subject pattern is matched against. Integer values are stable across versions.

const (
	SubjectUsername    SubjectKind = 0
	SubjectClientID    SubjectKind = 1
	SubjectIPAddr      SubjectKind = 2
	SubjectCertCN      SubjectKind = 3
	SubjectCertSubject SubjectKind = 4
)

func ParseSubjectKind

func ParseSubjectKind(s string) (SubjectKind, bool)

ParseSubjectKind parses the string form.

func (SubjectKind) String

func (s SubjectKind) String() string

String returns the YAML / JSON / human-readable form.

type User

type User struct {
	// Subject is the lookup key for this record: a username when the backend
	// is configured AuthMode=Username, or a clientID when AuthMode=ClientID.
	Subject string `json:"subject"`
	// Allow is whether the user may connect. False denies authentication
	// without removing the record (useful for temporary lockouts).
	Allow bool `json:"allow"`
}

User is the wire shape returned by Backend.Users / Backend.GetUser. Passwords are never read back through this struct; PutUser takes the plaintext separately.