extension

package
v0.0.0-...-c848cff Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 5, 2026 License: MIT Imports: 43 Imported by: 0

Documentation

Overview

Package extension mounts Tether in a Forge app: the MQTT broker, its REST API, and the keysmith and warden plugins that make live connections react to key and role changes.

Build keysmith and warden through the extension so the plugins are registered with their engines: 

t := extension.New(extension.WithAdminMiddleware(adminAuth))
app.RegisterExtension(t.Keysmith())
app.RegisterExtension(t.Warden())
app.RegisterExtension(t)

Tether depends on both, so Forge starts them first. Start then builds the store (WithStore, else the container's grove database), the services, the broker and the routes.

Index

Constants

View Source 
const (
	// MetricConnections is a gauge of live MQTT connections on this node.
	MetricConnections = "tether_connections"

	// MetricMessagesIn counts client PUBLISH packets that passed the ACL
	// and the tenant's rate limit.
	MetricMessagesIn = "tether_messages_in_total"

	// MetricMessagesOut counts PUBLISH packets written to clients.
	MetricMessagesOut = "tether_messages_out_total"

	// MetricAuthFailures counts refused CONNECTs, labelled reason (one of
	// the broker.AuthReason constants).
	MetricAuthFailures = "tether_auth_failures_total"

	// MetricACLDenials counts failed ACL checks, labelled op ("publish" or
	// "subscribe").
	MetricACLDenials = "tether_acl_denials_total"

	// MetricMeteringErrors counts the meter's failures: a usage value ledger
	// did not take (a temporary error, retried in the next window, or a
	// rejection, dropped) and a tenant it could not look up.
	MetricMeteringErrors = "tether_metering_errors_total"

	// MetricEntitlementChecks counts the feature checks that reached
	// ledger, labelled result: "allowed", "denied" (ledger answered no) or
	// "error" (no answer). A check skipped for want of a slot is not one.
	MetricEntitlementChecks = "tether_entitlement_checks_total"

	// MetricAuditDropped counts audit events the recorder dropped: its
	// queue was full, or Stop ran out of time. It catches up with the
	// recorder every 5 s and once more at Stop.
	MetricAuditDropped = "tether_audit_dropped_total"

	// MetricAuditFailures counts audit events chronicle refused or did not
	// answer for: they reached the sink and were lost there. With
	// MetricAuditDropped it is what the trail is missing. It catches up with
	// the recorder every 5 s and once more at Stop.
	MetricAuditFailures = "tether_audit_failures_total"

	// MetricClusterFenced is a gauge, 1 while the node is fenced from its
	// cell (it has dropped its clients and refuses new ones), else 0.
	MetricClusterFenced = "tether_cluster_fenced"

	// MetricLogAppendSeconds is a histogram of the time a QoS 1 or 2 append
	// to the cell's log takes from send to commit, client publishes, host
	// publishes and wills alike. The time includes any wait before the send
	// while the node's publisher has 4000 acks outstanding.
	MetricLogAppendSeconds = "tether_log_append_seconds"

	// MetricCASLost counts session record writes that lost their
	// compare-and-set to another writer: contested CONNECTs, takeovers and
	// dead-node reclaims racing each other.
	MetricCASLost = "tether_cas_lost_total"

	// MetricReplayMessages counts messages replayed from the log to
	// sessions that came back, or moved here, behind the tenant's reader.
	MetricReplayMessages = "tether_replay_messages_total"
)

Metric names. None carries a tenant label: per-tenant numbers are in the admin API's stats and each tenant's $SYS topics.

View Source 
const DefaultBasePath = "/tether"

DefaultBasePath is where the REST API mounts when the config names no base path.

View Source 
const ExtensionDescription = "Multi-tenant MQTT broker with keysmith device keys and warden topic permissions"

ExtensionDescription is the human-readable description.

View Source 
const ExtensionName = "tether"

ExtensionName is the name Tether registers with Forge.

View Source 
const ExtensionVersion = tether.Version

ExtensionVersion is the extension's version, Tether's.

Variables

View Source 
var (
	// ErrKeysmithNotWired is returned by Start when the app's keysmith
	// extension was not built with (*Extension).Keysmith and
	// WithoutKeysmithEvents was not set.
	ErrKeysmithNotWired = errors.New("tether: keysmith was built without the tether plugin; " +
		"register (*Extension).Keysmith() instead of keysmith's extension, or pass WithoutKeysmithEvents")

	// ErrWardenNotWired is returned by Start when the warden plugin could
	// not be registered (warden built without (*Extension).Warden and with
	// no plugin registry) and WithoutWardenEvents was not set.
	ErrWardenNotWired = errors.New("tether: the warden engine has no plugin registry for the tether plugin; " +
		"register (*Extension).Warden() instead of warden's extension, or pass WithoutWardenEvents")

	// ErrAdminAuthRequired is returned by Start when routes are on and
	// neither WithAdminMiddleware nor WithInsecureAdminRoutes was given.
	ErrAdminAuthRequired = errors.New("tether: the admin API needs authentication; " +
		"pass WithAdminMiddleware, or WithInsecureAdminRoutes to mount it open, or disable routes")
)
View Source 
var (
	// ErrLedgerRequired is returned by Start when metering or entitlements
	// are enabled and no *ledger.Ledger is in the container.
	ErrLedgerRequired = errors.New("tether: metering and entitlements need ledger; " +
		"register ledger's extension (github.com/xraph/ledger/extension), " +
		"or set metering.enabled and entitlements.enabled to false")

	// ErrChronicleRequired is returned by Start when audit is enabled and no
	// chronicle.Emitter is in the container.
	ErrChronicleRequired = errors.New("tether: audit needs chronicle; " +
		"register chronicle's extension (github.com/xraph/chronicle/extension), " +
		"or set audit.enabled to false")
)

Functions

This section is empty.

Types

type Config 

type Config struct {
	// BasePath prefixes every REST route: the admin API under
	// <base>/v1/tenants and the device endpoint <base>/v1/publish.
	BasePath string `json:"base_path" yaml:"base_path"`

	// DisableRoutes mounts no REST routes at all. The broker still runs.
	DisableRoutes bool `json:"disable_routes" yaml:"disable_routes"`

	// GroveDatabase names the grove database in the DI container the store
	// is built on. Empty uses the container's default one. WithStore
	// overrides both.
	GroveDatabase string `json:"grove_database" yaml:"grove_database"`

	// DisableMigrate stops Start from migrating the store, for a host that
	// runs Tether's migration group itself. Migrate and Rollback then refuse
	// to run. WithDisableMigrate sets the same thing.
	DisableMigrate bool `json:"disable_migrate" yaml:"disable_migrate"`

	// Broker configures the MQTT listeners and protocol limits.
	Broker broker.Config `json:"broker" yaml:"broker"`

	// JWT turns on authsome JWT authentication next to device keys: a CONNECT
	// whose password is a JWT is checked against the identity provider's
	// JWKS, any other password goes to keysmith. The zero value leaves it off.
	// See auth.JWTConfig for the fields.
	JWT auth.JWTConfig `json:"jwt" yaml:"jwt"`

	// DashboardOperators makes every signed-in dashboard user an operator:
	// they may act on any tenant and create, change and delete tenants.
	// Without it a dashboard user is confined to the tenant their session
	// names. WithDashboardOperators sets the same thing.
	DashboardOperators bool `json:"dashboard_operators" yaml:"dashboard_operators"`

	// Cluster makes this node one broker of a cell over NATS and JetStream
	// (package cluster): sessions, QoS 1 and 2 messages, retained messages
	// and shared subscriptions live in the cell, so a client can connect to
	// any node. Off while cluster.enabled is false, the default, and then
	// the broker runs alone as before. See cluster.Config for the keys;
	// log_max_age (default 7 days) is how long an offline session's
	// messages wait in the log. WithCluster sets the same thing.
	Cluster cluster.Config `json:"cluster" yaml:"cluster"`

	// Metering sends each tenant's usage (messages and bytes in and out,
	// connection minutes) to ledger every interval, under the tenant's host
	// pair. Off while metering.enabled is false, the default. On, Start
	// needs ledger's extension in the app (ErrLedgerRequired). See
	// metering.Config for the keys. WithMetering sets the same thing.
	Metering metering.Config `json:"metering" yaml:"metering"`

	// Entitlements refuses a CONNECT with 0x97 for a tenant whose ledger
	// plan does not include every listed feature, and disconnects its live
	// clients with 0x97 at the next check. Off while entitlements.enabled is
	// false, the default. On, Start needs ledger's extension in the app
	// (ErrLedgerRequired). See metering.EntitlementConfig for the keys.
	// WithEntitlements sets the same thing.
	Entitlements metering.EntitlementConfig `json:"entitlements" yaml:"entitlements"`

	// Audit records connects, disconnects, refused CONNECTs, ACL denials,
	// takeovers and admin changes in chronicle. Off while audit.enabled is
	// false, the default. On, Start needs chronicle's extension in the app
	// (ErrChronicleRequired). See audit.Config for the keys. WithAudit sets
	// the same thing.
	Audit audit.Config `json:"audit" yaml:"audit"`

	// RequireConfig makes Register fail when the app's config has no
	// tether section.
	RequireConfig bool `json:"-" yaml:"-"`
}

Config configures the Tether extension. It loads from "extensions.tether" (or "tether") in the app's config, layered over DefaultConfig and under the options passed to New.

func DefaultConfig 

func DefaultConfig() Config

DefaultConfig returns the base path "/tether", broker.DefaultConfig (TCP on :1883, 1 MiB packets, 7 day session cap, 1 KiB topics), cluster.DefaultConfig (off), and metering, entitlements and audit at their packages' defaults (all off).

type Extension 

type Extension struct {
	*forge.BaseExtension
	// contains filtered or unexported fields
}

Extension is Tether as a Forge extension.

func New 

func New(opts ...Option) *Extension

New returns the extension. Options override DefaultConfig; the app's config file (extensions.tether or tether) is layered in by Register.

func (*Extension) Broker 

func (e *Extension) Broker() *broker.Broker

Broker returns the broker, nil before Start.

func (*Extension) DashboardStatus 

func (e *Extension) DashboardStatus() dashboard.DashboardStatus

DashboardStatus implements dashboard.DashboardStatusAware: configured once Start has succeeded, until Stop.

func (*Extension) Health 

func (e *Extension) Health(ctx context.Context) error

Health fails when the broker is not running or the store does not answer, and in a cell when the node cannot act for it (broker.ClusterHealth): an error wrapping cluster.ErrFenced while it is fenced, or cluster.ErrUnavailable while NATS or JetStream is out of reach.

func (*Extension) Keysmith 

func (e *Extension) Keysmith(opts ...keysmithext.ExtOption) forge.Extension

Keysmith returns the keysmith extension with Tether's keysmith plugin registered, built from opts. Register it with the app instead of calling keysmithext.New yourself: keysmith takes plugins only when its engine is built, and without the plugin a revoked device key would keep its live MQTT connections. Start fails with ErrKeysmithNotWired when the keysmith engine in the container is not the one an extension built here owns (the app registered keysmith some other way, or never registered this one), unless WithoutKeysmithEvents is set.

func (*Extension) Migrate 

func (e *Extension) Migrate(ctx context.Context) (*forge.MigrationResult, error)

Migrate runs Tether's pending migrations under grove's lock. Forge's migrate up calls it. On a store that is not on grove it runs the store's own Migrate and reports nothing applied. Once the group went to the central migrator it does nothing, since that pass already ran it.

func (*Extension) MigrationStatus 

func (e *Extension) MigrationStatus(ctx context.Context) ([]*forge.MigrationGroupInfo, error)

MigrationStatus reports Tether's group, applied and pending. A store that is not on grove has no group and reports none.

func (*Extension) Register 

func (e *Extension) Register(app forge.App) error

Register loads the config and puts the extension in the container. When the app runs central migrations and grove's MigrationRegistry is in the container, it also hands the store's migration group to the registry (see contribute), and Start then leaves migrating to that pass. Everything else waits for Start, once keysmith and warden have started.

func (*Extension) RegisterContractContributor 

func (e *Extension) RegisterContractContributor(disp *dispatcher.Dispatcher, reg dashcontract.Registry, wreg dashcontract.WardenRegistry) error

RegisterContractContributor implements dashboard.ContractContributorAware: the dashboard calls it during its own Start and gets Tether's admin intents (see package contract). The handlers read the services through Services at request time, so they answer UNAVAILABLE until Tether's Start has built them, whichever extension started first.

Every intent refuses a request whose principal has no authenticated user (UNAUTHENTICATED), so a dashboard with its auth off serves nothing here unless you pass WithInsecureDashboard. The contributor stays registered either way, so the plugin still shows its pages and the refusal.

A signed-in user is confined to their own tenant unless the extension runs with WithDashboardOperators (or dashboard_operators: true), which makes every signed-in user an operator. WithInsecureDashboard implies it.

func (*Extension) Rollback 

func (e *Extension) Rollback(ctx context.Context) (*forge.MigrationResult, error)

Rollback rolls back the last applied migration, the way grove's own extension does: grove looks at the newest row in grove_migrations, so on a database other extensions share it fails when that row is not Tether's. Forge's migrate down calls it.

func (*Extension) Services 

func (e *Extension) Services() *api.Services

Services returns what the REST handlers run on, nil before Start.

func (*Extension) Start 

func (e *Extension) Start(ctx context.Context) error

Start wires and starts Tether. In order: the admin auth check, the metering, entitlements and audit configs and the ledger or chronicle they need (ErrLedgerRequired, ErrChronicleRequired), the keysmith engine from the container (which must be the one the extension from Keysmith owns, so the plugin is on it), the warden engine and its plugin (registered now if Warden did not), the store (migrated), the services, the cell when the config has one (connect, provision, the cluster services, the tenant cache's eviction on the control bus), the audit recorder, the broker with the entitlement checks and the plugins bound to its reactor, the routes, the control bus, the broker's listeners, Members, so the cell's callbacks find a running broker, and last the meter and the entitlement loop. A failure undoes whatever had started.

Start is idempotent: on an extension that is already started it returns nil and does nothing. That matters because *Extension is provided to the container, and resolving it there (vessel.Inject[*extension.Extension]) makes vessel start it once more.

func (*Extension) Stop 

func (e *Extension) Stop(ctx context.Context) error

Stop unbinds the plugins, ends the entitlement checks and stops the broker, which disconnects every client. The meter's last window and the audit recorder's drain run after the broker and the cell. In a cell it then stops Members, the control bus and the NATS connection (and the embedded server), in that order: Members goes after the broker, since its peers take over this node's sessions and publish the delayed wills it held once it stops heartbeating. The store is left open: Tether did not open it. Stopping an extension that is not started returns nil.

func (*Extension) Warden 

func (e *Extension) Warden(opts ...wardenext.Option) forge.Extension

Warden returns the warden extension with Tether's warden plugin registered, built from opts. Registering the plugin at construction always works; registering it at Start (what happens when the app built warden some other way) needs the engine to have a plugin registry, which warden v1.8.0 creates only for an engine with a plugin, a decision cache or the audit log. Without one Start fails with ErrWardenNotWired, unless WithoutWardenEvents is set.

type Option 

type Option func(*Extension)

Option configures the extension.

func WithAdminMiddleware 

func WithAdminMiddleware(mw ...forge.Middleware) Option

WithAdminMiddleware wraps the admin routes in mw, in order. This is how the admin API gets authenticated: Tether adds no admin auth of its own. The device endpoint POST <base>/v1/publish never gets these; it authenticates the device with HTTP Basic itself.

func WithAudit 

func WithAudit(cfg audit.Config) Option

WithAudit records the audit trail in chronicle, configured by cfg. Enabled false leaves it off; an empty Events list and zero numbers take audit.DefaultConfig's. The app must register chronicle's extension, or Start fails with ErrChronicleRequired. The "audit" block of the config file sets the same thing.

func WithBrokerConfig 

func WithBrokerConfig(cfg broker.Config) Option

WithBrokerConfig sets the broker's listeners and limits.

func WithCluster 

func WithCluster(cfg cluster.Config) Option

WithCluster runs the broker as one node of a cell, configured by cfg. It turns cfg.Enabled on: passing it means you want the cell. Start connects to the cell's NATS (or starts the embedded server), provisions its streams and buckets, and joins; see cluster.Config for the keys and the README for the embedded and external shapes. The "cluster" block of the config file sets the same thing. Mode, Heartbeat, Lease, LogMaxAge, PullHeartbeat and LoadInterval left zero take cluster.DefaultConfig's, so a literal with only the fields you need passes Validate.

func WithConfig 

func WithConfig(cfg Config) Option

WithConfig replaces the whole config. Fields the app's config file sets win over the ones here that equal DefaultConfig.

func WithDashboardOperators 

func WithDashboardOperators() Option

WithDashboardOperators makes every signed-in dashboard user an operator, who may act on any tenant and create, change and delete tenants. Without it a dashboard user is confined to the one tenant their session names (a tenant_id claim, or the org of the forge Scope the host's auth middleware sets), mapped to a Tether tenant; a user with no such tenant is refused. The config key dashboard_operators: true does the same.

func WithDisableMigrate 

func WithDisableMigrate() Option

WithDisableMigrate stops Start from migrating the store. Use it when the host runs the store's grove group (postgres.Migrations or sqlite.Migrations) some other way. The disable_migrate config key sets the same thing.

func WithEntitlements 

func WithEntitlements(cfg metering.EntitlementConfig) Option

WithEntitlements checks each tenant's ledger plan at CONNECT and every interval, configured by cfg: a tenant missing any of cfg.Features is refused, and disconnected, with 0x97. Enabled false leaves it off. An empty feature list and zero durations take metering.DefaultEntitlementConfig's. The app must register ledger's extension, or Start fails with ErrLedgerRequired. The "entitlements" block of the config file sets the same thing.

func WithGroveDatabase 

func WithGroveDatabase(name string) Option

WithGroveDatabase names the grove database in the DI container to build the store on. Without it (and without WithStore) the container's default grove database is used.

func WithInsecureAdminRoutes 

func WithInsecureAdminRoutes() Option

WithInsecureAdminRoutes mounts the admin routes with no admin middleware. Anyone who reaches the HTTP port can then create tenants and devices and read their credentials. It is for tests and for apps that put the whole router behind their own gate. Without it, Start fails with ErrAdminAuthRequired when routes are on and WithAdminMiddleware was not given.

func WithInsecureDashboard 

func WithInsecureDashboard() Option

WithInsecureDashboard serves the dashboard intents to requests that carry no authenticated user. Without it they answer UNAUTHENTICATED, which is what you get when the dashboard's own auth is off, the same refusal the REST admin API gives without WithAdminMiddleware. Use it only when the dashboard is reachable by trusted callers alone.

It implies WithDashboardOperators: an anonymous caller has no session to take a tenant from, so it is served as an operator.

func WithJWT 

func WithJWT(cfg auth.JWTConfig) Option

WithJWT lets users and service accounts connect with an authsome JWT as the MQTT password, next to device keys. The username must be the token's sub. The JWT carries the tenant (app_id and org_id) and warden roles assigned to the user decide the topics. Without it only keysmith device keys connect. The "jwt" block of the config file sets the same thing; the option wins where it differs from the empty default. A config with any field set is validated at Start, which fails if it is incomplete.

POST <base>/v1/publish stays device-only.

func WithLogger 

func WithLogger(l forge.Logger) Option

WithLogger sets the logger the extension, the broker and mochi log through. Without it they use the app's logger.

func WithMetering 

func WithMetering(cfg metering.Config) Option

WithMetering sends each tenant's usage to ledger, configured by cfg. Enabled false leaves metering off. A zero Interval takes the default (60 s), and Features with every key empty take metering.DefaultConfig's five, so WithMetering(metering.Config{Enabled: true}) meters everything under the tether.* keys. Name any key and the others stay as you gave them: an empty one turns that meter off. The app must register ledger's extension, or Start fails with ErrLedgerRequired. The "metering" block of the config file sets the same thing, and there a block with every key empty fails Start (metering.Config.Validate).

func WithStore 

func WithStore(s store.Store) Option

WithStore uses s for tenants and devices instead of a grove database from the container. The extension migrates it on Start (unless WithDisableMigrate) and never closes it. A WithStore store never goes to the central migrator, since its database need not be the one the migrator runs on.

func WithoutKeysmithEvents 

func WithoutKeysmithEvents() Option

WithoutKeysmithEvents lets Start run although the keysmith extension was not built with (*Extension).Keysmith. Revoking, suspending or rotating a device key then leaves live MQTT connections alone, and the cached validation keeps a revoked key working for up to a minute.

func WithoutWardenEvents 

func WithoutWardenEvents() Option

WithoutWardenEvents lets Start run although the warden plugin could not be registered (warden built without (*Extension).Warden and with no plugin registry). Role and assignment changes then reach live MQTT connections only when they reconnect.

Source Files

View all Source files

Directories

Path Synopsis
Package contract publishes Tether's admin operations to the Forge dashboard as a contract contributor: a declarative manifest plus typed handlers that call the same services the REST API in package api does.
 Package contract publishes Tether's admin operations to the Forge dashboard as a contract contributor: a declarative manifest plus typed handlers that call the same services the REST API in package api does.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.