Documentation
¶
Overview ¶
Package events turns keysmith key changes and warden role changes into reactions on live MQTT connections.
Two plugins listen to the engines: KeysmithPlugin for key rotation, revocation, suspension and expiry, and WardenPlugin for role, permission and assignment changes. Each reports what happened to a Handler in broker terms: a key is no longer valid, a role's grants changed, or a subject's roles changed. NewReactor returns the Handler that applies those to the broker: it evicts the key from the authenticator's cache, and disconnects or reloads the affected connections. In a cell, NewClusterReactor sends each change through the control bus instead, and every node applies it.
Plugin hooks run synchronously inside the engine call that made the change. With NewReactor the reaction is done when that call returns. With NewClusterReactor the call returns once the bus has stored the change, and each node, this one included, applies it asynchronously when the event reaches it. Hooks never fail and never panic into the engine: keysmith stops dispatching at the first plugin error.
Index ¶
- func EvictTenantsOnUpdate(bus Bus, evict func(tenantID string)) error
- type BrokerControl
- type Bus
- type Handler
- type KeysmithPlugin
- func (p *KeysmithPlugin) Bind(h Handler)
- func (p *KeysmithPlugin) MarkWired()
- func (p *KeysmithPlugin) Name() string
- func (p *KeysmithPlugin) OnKeyExpired(ctx context.Context, k *key.Key) error
- func (p *KeysmithPlugin) OnKeyRevoked(ctx context.Context, k *key.Key, _ string) error
- func (p *KeysmithPlugin) OnKeyRotated(ctx context.Context, k *key.Key, rec *rotation.Record) error
- func (p *KeysmithPlugin) OnKeySuspended(ctx context.Context, k *key.Key) error
- func (p *KeysmithPlugin) Wired() bool
- type WardenPlugin
- func (p *WardenPlugin) Bind(h Handler)
- func (p *WardenPlugin) MarkWired()
- func (p *WardenPlugin) Name() string
- func (p *WardenPlugin) OnPermissionAttached(ctx context.Context, roleID id.RoleID, _ id.PermissionID) error
- func (p *WardenPlugin) OnPermissionDetached(ctx context.Context, roleID id.RoleID, _ id.PermissionID) error
- func (p *WardenPlugin) OnRoleAssigned(ctx context.Context, a *assignment.Assignment) error
- func (p *WardenPlugin) OnRoleDeleted(ctx context.Context, roleID id.RoleID) error
- func (p *WardenPlugin) OnRoleUnassigned(ctx context.Context, a *assignment.Assignment) error
- func (p *WardenPlugin) OnRoleUpdated(ctx context.Context, r *role.Role) error
- func (p *WardenPlugin) Wired() bool
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func EvictTenantsOnUpdate ¶
EvictTenantsOnUpdate subscribes evict to the bus's tenant.updated events, so every node of the cell drops its cached copy of a tenant that changed (service.Tenants.Evict). The broker applies the change itself (broker.Cluster.Control); subscribe this before the broker's New, so the eviction runs first, and a client the change disconnects cannot reconnect with the old tenant from the cache. evict runs on the bus's goroutine and must be quick.
Types ¶
type BrokerControl ¶
type BrokerControl interface {
DisconnectPrincipal(principalID string, code packets.Code) int
ExpirePrincipal(principalID string, connectedBefore, at time.Time) int
ReloadRole(ctx context.Context, roleID string)
ReloadPrincipal(ctx context.Context, principalID string)
}
BrokerControl is the part of *broker.Broker the reactor drives.
type Bus ¶
type Bus interface {
Publish(ctx context.Context, kind string, v any) error
Subscribe(kind string, fn func([]byte)) error
}
Bus is the cell's control bus a clustered reactor publishes its events to and applies them from. *cluster.Control satisfies it.
type Handler ¶
type Handler interface {
// KeyInvalidated reports that the keysmith key keyID changed.
// disconnect is true when the key may no longer be used at all
// (revoked, suspended or expired) and false when only its secret
// changed (rotated; the old one may still work during a grace window).
KeyInvalidated(ctx context.Context, keyID string, disconnect bool)
// KeyRotated reports that the keysmith key keyID was rotated with a
// grace window: its previous secret keeps working until graceEnds, and
// no longer. rotatedAt is a time after keysmith stored the new secret,
// so every connection made with the previous secret predates it. A
// rotation with no grace window is KeyInvalidated(keyID, true) instead.
KeyRotated(ctx context.Context, keyID string, rotatedAt, graceEnds time.Time)
// RolesChanged reports that the warden role roleID changed: updated,
// deleted, or a permission attached or detached.
RolesChanged(ctx context.Context, roleID string)
// SubjectChanged reports that a role was assigned to or unassigned
// from a warden subject. subjectKind is warden's subject kind, "api_key"
// for a keysmith key.
SubjectChanged(ctx context.Context, subjectKind, subjectID string)
}
Handler receives key and role changes from the plugins.
Implementations are called from inside keysmith and warden admin calls (and from keysmith's ValidateKey for an expired key), so they should finish quickly and must not call back into the engine that called them. The cluster reactor (NewClusterReactor) waits for the control bus to store each change, so it can hold such a call, a device's ValidateKey included, for up to its publish wait (5 s) while the bus does not answer.
func NewClusterReactor ¶
func NewClusterReactor(evict func(keyID string), b BrokerControl, bus Bus, log *slog.Logger) (Handler, error)
NewClusterReactor is NewReactor for a broker that is one node of a cell: every change goes to every node through bus, and each node, this one included, applies it once when the event comes back (ruling R41). The change reaches the clients of every node, and every node's authenticator cache drops the key.
The Handler methods publish the change as a cluster event (key.invalidated, role.changed, subject.changed) and return once the bus has stored it, within publishWait (5 s). If the publish fails, the change is applied on this node as NewReactor would, logged at Warn on log (slog.Default when nil), and sent again in the background until the bus takes it (ruling R51; cluster.Republish: until it lands, the bus stops, or an hour has passed), so the other nodes get it once the bus is back. This node then applies it a second time, which is harmless.
NewClusterReactor subscribes its handlers on bus, so call it before the bus starts. An applied event evicts the key at once, on the bus's goroutine, and runs the broker call (disconnect, expiry, reload) on a goroutine of its own, so a slow reload never holds up the bus. A key.invalidated event with Disconnect is a revocation; one with RotatedAt or GraceEnds set is a rotation with a grace window; one with neither only evicts. The error is the bus's Subscribe error.
func NewReactor ¶
func NewReactor(evict func(keyID string), b BrokerControl) Handler
NewReactor returns the Handler that applies key and role changes to a broker. evict is the authenticator's cache eviction ((*auth.Keysmith).Evict) and may be nil.
- KeyInvalidated with disconnect: evict the key, then disconnect every connection authenticated with it (0x98 administrative action).
- KeyInvalidated without disconnect: evict the key only.
- KeyRotated (a rotation with a grace window): evict the key, then end every connection made before rotatedAt, which covers all that used the previous secret, when the grace window closes (ExpirePrincipal; 0xA0 then). Connections made with the new secret stay, and one made with the previous secret during the window already expires with it, because keysmith says so when it validates (auth.Keysmith).
- RolesChanged: reload the connections whose grants came from the role.
- SubjectChanged for an "api_key", "user" or "service_acct" subject: reload that principal's connections. Other subject kinds are ignored, since only those connect.
Eviction comes first, so a client that reconnects after the disconnect is validated against keysmith, not the cache. Empty IDs are ignored.
type KeysmithPlugin ¶
type KeysmithPlugin struct {
// contains filtered or unexported fields
}
KeysmithPlugin is a keysmith plugin that reports key changes to a Handler.
Keysmith takes plugins only when the engine is built (keysmith.WithExtension), usually before the broker that reacts to them exists, so the plugin is created first and bound to its Handler later with Bind. Events that arrive before Bind are dropped.
Rotation reports KeyRotated(id, rotatedAt, graceEnds): the previous secret keeps working until the grace window ends. rotatedAt is when the hook ran, which is after keysmith stored the new secret. A rotation with no grace window (keysmith.WithGrace(0), what a compromised key gets) reports KeyInvalidated(id, true) instead, because the previous secret is dead at once and a connection made with it cannot be told from one made with the new secret. A rotation hook with no record reports KeyInvalidated(id, false). Revocation, suspension and expiry report KeyInvalidated(id, true). Every hook returns nil and recovers a panic from the Handler, because keysmith stops dispatching at the first plugin error and would fail the caller's operation.
func NewKeysmithPlugin ¶
func NewKeysmithPlugin() *KeysmithPlugin
NewKeysmithPlugin returns an unbound plugin.
func (*KeysmithPlugin) Bind ¶
func (p *KeysmithPlugin) Bind(h Handler)
Bind sets the Handler that receives events from now on. A nil Handler unbinds.
func (*KeysmithPlugin) MarkWired ¶
func (p *KeysmithPlugin) MarkWired()
MarkWired records that the plugin was registered with a keysmith engine. The host calls it after keysmith.WithExtension(p), so code that receives the plugin later can check Wired instead of trusting that it was.
func (*KeysmithPlugin) Name ¶
func (p *KeysmithPlugin) Name() string
Name implements keysmith's plugin.Plugin.
func (*KeysmithPlugin) OnKeyExpired ¶
OnKeyExpired implements keysmith's plugin.KeyExpired. Keysmith fires it from ValidateKey when it finds the key expired.
func (*KeysmithPlugin) OnKeyRevoked ¶
OnKeyRevoked implements keysmith's plugin.KeyRevoked.
func (*KeysmithPlugin) OnKeyRotated ¶
OnKeyRotated implements keysmith's plugin.KeyRotated.
func (*KeysmithPlugin) OnKeySuspended ¶
OnKeySuspended implements keysmith's plugin.KeySuspended.
func (*KeysmithPlugin) Wired ¶
func (p *KeysmithPlugin) Wired() bool
Wired reports whether MarkWired was called.
type WardenPlugin ¶
type WardenPlugin struct {
// contains filtered or unexported fields
}
WardenPlugin is a warden plugin that reports role changes to a Handler.
Role updates and deletions, and permissions attached to or detached from a role, report RolesChanged with the role's ID. Assignments and unassignments report SubjectChanged with the assignment's subject kind and ID. Warden emits no tenant with these, which is fine: role IDs and keysmith key IDs are unique across tenants.
Register it when building the engine (warden.WithPlugin) or later (Engine.Plugins().Register). Note that warden v1.8.0's Engine.Plugins() is nil unless the engine was built with a plugin or a decision cache, so registering at construction is the dependable way. The Handler (a reactor over a broker that needs the engine) usually does not exist yet then: pass nil and Bind it later, as with KeysmithPlugin. Events before Bind are dropped.
Every hook returns nil and recovers a panic from the Handler. Warden itself only logs plugin errors, but a failing hook would still be noise in its metrics.
func NewWardenPlugin ¶
func NewWardenPlugin(h Handler) *WardenPlugin
NewWardenPlugin returns a plugin that reports to h. A nil h leaves it unbound until Bind.
func (*WardenPlugin) Bind ¶
func (p *WardenPlugin) Bind(h Handler)
Bind sets the Handler that receives events from now on. A nil Handler unbinds.
func (*WardenPlugin) MarkWired ¶
func (p *WardenPlugin) MarkWired()
MarkWired records that the plugin was registered with a warden engine, at construction (warden.WithPlugin) or through Engine.Plugins().Register. As with KeysmithPlugin, code that receives the plugin later checks Wired instead of trusting that it was.
func (*WardenPlugin) Name ¶
func (p *WardenPlugin) Name() string
Name implements warden's plugin.Plugin.
func (*WardenPlugin) OnPermissionAttached ¶
func (p *WardenPlugin) OnPermissionAttached(ctx context.Context, roleID id.RoleID, _ id.PermissionID) error
OnPermissionAttached implements warden's plugin.PermissionAttached.
func (*WardenPlugin) OnPermissionDetached ¶
func (p *WardenPlugin) OnPermissionDetached(ctx context.Context, roleID id.RoleID, _ id.PermissionID) error
OnPermissionDetached implements warden's plugin.PermissionDetached.
func (*WardenPlugin) OnRoleAssigned ¶
func (p *WardenPlugin) OnRoleAssigned(ctx context.Context, a *assignment.Assignment) error
OnRoleAssigned implements warden's plugin.RoleAssigned.
func (*WardenPlugin) OnRoleDeleted ¶
OnRoleDeleted implements warden's plugin.RoleDeleted.
func (*WardenPlugin) OnRoleUnassigned ¶
func (p *WardenPlugin) OnRoleUnassigned(ctx context.Context, a *assignment.Assignment) error
OnRoleUnassigned implements warden's plugin.RoleUnassigned.
func (*WardenPlugin) OnRoleUpdated ¶
OnRoleUpdated implements warden's plugin.RoleUpdated.
func (*WardenPlugin) Wired ¶
func (p *WardenPlugin) Wired() bool
Wired reports whether MarkWired was called.
Source Files