Documentation
¶
Overview ¶
Package cluster implements llmbox's hub-and-spoke model: a single hub (the MCP front-end the chatbot talks to) drives box operations on one or more spokes, each of which owns a local Docker daemon. A spoke dials the hub over a WebSocket and the hub pushes box verbs down that connection; the spoke executes them against its local *box.Manager and replies.
The wire surface is deliberately the box verbs of BoxManager and nothing more: a spoke is never a generic Docker proxy. See docs/hub-and-spoke.md.
Index ¶
- Variables
- func CreateJoinToken(store Store, name, backend string, ttl time.Duration, now time.Time) (string, error)
- func Run(ctx context.Context, dial Dialer, mgr BoxManager, joinToken string, ...) error
- func RunWithCaller(ctx context.Context, dial Dialer, mgr BoxManager, joinToken string, ...) error
- type BoxDialer
- type BoxManager
- type BoxPortInfo
- type BoxPortService
- type Credentials
- type Dialer
- type Hub
- type HubCaller
- type JoinTokenInfo
- type JoinTokenRecord
- type SpokeRecord
- type Store
Constants ¶
This section is empty.
Variables ¶
var ErrEnrollRejected = errors.New("enrollment rejected by hub")
ErrEnrollRejected reports that the hub refused enrollment (bad/expired/used join token, or a credential that no longer matches). It is terminal: retrying with the same token will not succeed, so a caller should stop rather than reconnect.
Functions ¶
func CreateJoinToken ¶
func CreateJoinToken(store Store, name, backend string, ttl time.Duration, now time.Time) (string, error)
CreateJoinToken mints a one-time join token for a spoke name, stores its hash with the given expiry, and returns the plaintext token to show the operator once. backend records the box backend the spoke was created for (presentation only, not validated here; empty means docker) so setup instructions can be re-rendered after creation. ttl<=0 is rejected so a token always expires.
@arg store The cluster store to persist the token in. @arg name The spoke name baked into the token; required. @arg backend The box backend recorded on the token; empty means docker. @arg ttl How long the token stays valid; must be positive. @arg now The current time (for the expiry). @return string The plaintext join token (shown once, never recoverable). @error error if the name is empty, ttl is non-positive, the secret cannot be generated, or the store write fails.
@testcase TestCreateJoinTokenStoresHash stores the token hash (with its backend) and returns a usable secret. @testcase TestCreateJoinTokenRejectsEmptyName rejects an empty spoke name. @testcase TestCreateJoinTokenRejectsTTL rejects a non-positive ttl.
func Run ¶
func Run(ctx context.Context, dial Dialer, mgr BoxManager, joinToken string, creds *Credentials, save func(Credentials) error) error
Run connects a spoke to the hub and serves box verbs against mgr until ctx is cancelled or the connection drops. It enrolls using joinToken when creds is nil; otherwise it reconnects with the saved creds. On first enrollment it invokes save with the minted credentials so the caller can persist them. Run returns when the connection ends; the caller decides whether to retry.
@arg ctx Context whose cancellation stops the spoke. @arg dial The dialer establishing the transport to the hub. @arg mgr The local box manager verbs are executed against. @arg joinToken The one-time join token for first enrollment; ignored when creds is set. @arg creds Saved credentials for reconnect; nil for first enrollment. @arg save Callback invoked with freshly minted credentials on first enrollment; may be nil. @error error if dialing, enrollment, or the serve loop fails.
@testcase TestSpokeRunEnrollsAndServes enrolls with a join token and serves a verb. @testcase TestSpokeRunReconnectsWithCreds reconnects using saved credentials. @testcase TestSpokeRunEnrollRejected returns the hub's rejection error.
func RunWithCaller ¶
func RunWithCaller(ctx context.Context, dial Dialer, mgr BoxManager, joinToken string, creds *Credentials, save func(Credentials) error, caller *HubCaller) error
RunWithCaller is Run plus an optional HubCaller: once enrollment succeeds the caller is attached to this connection so spoke-side components (the per-box port API) can issue spoke→hub requests over it, and it is detached — failing its in-flight calls — when the connection ends. A nil caller disables the spoke→hub direction, making this exactly Run.
@arg ctx Context whose cancellation stops the spoke. @arg dial The dialer establishing the transport to the hub. @arg mgr The local box manager verbs are executed against. @arg joinToken The one-time join token for first enrollment; ignored when creds is set. @arg creds Saved credentials for reconnect; nil for first enrollment. @arg save Callback invoked with freshly minted credentials on first enrollment; may be nil. @arg caller The long-lived caller to attach to this connection; nil disables spoke→hub requests. @error error if dialing, enrollment, or the serve loop fails.
@testcase TestSpokeCallerRoundTrip attaches a caller and round-trips box-port verbs. @testcase TestSpokeCallerReconnects re-attaches the same caller across connections.
Types ¶
type BoxDialer ¶
type BoxDialer interface {
DialBox(ctx context.Context, idOrName string, port int) (net.Conn, error)
}
BoxDialer is the box-reachability capability the spoke-side stream tunnel needs from its local box manager: open a connection to a port inside a box. The in-process *box.Manager implements it. It is kept here (not on BoxManager) so the box-verb RPC allowlist is unchanged and only a spoke that can dial boxes services proxy tunnels. The dial resolves through the box layer's managed-only check (see box.Manager.DialBox), so a tunnel can only ever reach a port inside a box the spoke created — never an arbitrary host address.
type BoxManager ¶
type BoxManager interface {
Create(ctx context.Context, opts sandbox.CreateOptions) (sandbox.CreateResult, error)
List(ctx context.Context) ([]sandbox.Box, error)
Destroy(ctx context.Context, idOrName string) error
Pause(ctx context.Context, idOrName string) error
Resume(ctx context.Context, idOrName string) error
Exec(ctx context.Context, idOrName string, cmd []string) (sandbox.ExecResult, error)
}
BoxManager is the box-lifecycle surface the hub needs from a spoke. The local in-process implementation is *box.Manager; the remote implementation (remoteSpoke) round-trips each call over a transport to a spoke process. It is the complete RPC allowlist of the cluster protocol — no operation outside it can cross the hub/spoke boundary.
type BoxPortInfo ¶
type BoxPortInfo struct {
Port int `json:"port"`
URL string `json:"url"`
Description string `json:"description,omitempty"`
}
BoxPortInfo is the box-facing view of one published port: just the port, its public URL, and the caller's description. It deliberately omits hub-side details (slug, spoke, creator) that a box has no business seeing.
type BoxPortService ¶
type BoxPortService interface {
// OpenBoxPort publishes a box's port and returns its public view.
OpenBoxPort(ctx context.Context, spokeName, boxID string, port int, description string) (BoxPortInfo, error)
// CloseBoxPort unpublishes a box's port.
CloseBoxPort(ctx context.Context, spokeName, boxID string, port int) error
// ListBoxPorts returns the box's published ports, and only that box's.
ListBoxPorts(ctx context.Context, spokeName, boxID string) ([]BoxPortInfo, error)
}
BoxPortService handles spoke-originated box-port requests on the hub. It is implemented by the hub server and injected into NewHub so the cluster layer stays free of hub imports. spokeName is the authenticated name of the connection a request arrived on (bound at enrollment, never caller-supplied), and boxID is the identity the SPOKE stamped from its own record of the originating box — implementations MUST verify that box actually lives on that spoke before acting, so a spoke (or a compromised box) can never manipulate another spoke's boxes.
type Credentials ¶
Credentials is a spoke's persisted enrollment state: the name the hub assigned and the bearer credential it minted. A spoke saves this after first enrollment and presents it to reconnect without the (one-time) join token.
type Dialer ¶
Dialer establishes a transport to the hub. It exists so Run can be tested over an in-memory transport instead of a real WebSocket.
func WebSocketDialer ¶
WebSocketDialer dials the hub's spoke-connect URL over a WebSocket.
SECURITY — transport confidentiality and integrity are the DEPLOYMENT's responsibility, not this code's. The enrollment handshake sends the one-time join token and then the long-lived bearer credential in the first frames, and every box verb (including the user's code and session details) flows over this socket. Use wss:// in production and terminate TLS at a trusted reverse proxy in front of the hub; a ws:// URL sends the credential and all traffic in cleartext and MUST only be used on a fully trusted network (e.g. loopback or a private mesh). This dialer accepts whichever scheme the operator passes — it does not (and cannot) verify the link is encrypted. The bearer credential is a static secret presented verbatim on every reconnect, so anyone who captures it (on the wire or at rest) can impersonate this spoke until it is revoked. For a wss:// hub with a private-CA or self-signed certificate, use WebSocketDialerTLS to trust the right CA (preferred) rather than disabling verification.
@arg url The hub's spoke-connect URL (ws:// or wss://). @return Dialer A dialer that opens a WebSocket transport to that URL.
@testcase TestSpokeRunEnrollsAndServes uses an in-memory dialer in place of this.
func WebSocketDialerTLS ¶
WebSocketDialerTLS is WebSocketDialer with a caller-supplied TLS client config for wss:// links — used to trust a hub's private-CA / self-signed certificate (config.RootCAs) or, as a last resort for testing, to skip verification (config.InsecureSkipVerify). A nil config uses the system trust store, exactly like WebSocketDialer. The same SECURITY notes as WebSocketDialer apply.
@arg url The hub's spoke-connect URL (ws:// or wss://). @arg tlsConf The TLS client config for wss:// dials; nil uses the system default. @return Dialer A dialer that opens a WebSocket transport to that URL.
@testcase TestHubEnrollAndRoute dials a real hub through this (with a nil config).
type Hub ¶
type Hub struct {
// contains filtered or unexported fields
}
Hub is the hub-side of the cluster: it accepts spoke connections on an HTTP route, authenticates their enrollment, and keeps a registry of connected spokes that the server routes box verbs to. A spoke connection is one long-lived WebSocket; the hub pushes verb requests down it via a remoteSpoke.
func NewHub ¶
func NewHub(ctx context.Context, store Store, now func() time.Time, log *slog.Logger, ports BoxPortService) *Hub
NewHub builds a Hub over the given store. ctx bounds the lifetime of accepted spoke connections (cancelling it closes them). now defaults to time.Now and log to slog.Default when nil.
@arg ctx Base context; its cancellation closes all spoke connections. @arg store The cluster store holding join tokens and enrolled spokes. @arg now Clock for token-expiry checks; nil uses time.Now. @arg log Logger for connection lifecycle; nil uses slog.Default. @arg ports The service handling spoke-originated box-port requests; nil rejects them. @return *Hub A ready hub with an empty connected-spoke registry.
@testcase TestHubEnrollAndRoute enrolls a spoke and routes a verb to it. @testcase TestHubWithoutBoxPortServiceRejects rejects box-port requests when ports is nil.
func (*Hub) ConnectHandler ¶
func (h *Hub) ConnectHandler(w http.ResponseWriter, r *http.Request)
ConnectHandler is the HTTP handler for the spoke connection route (/spoke/connect). It upgrades to a WebSocket, performs the enrollment handshake, registers the spoke, and serves verb requests over the connection until it drops or the hub's context is cancelled.
SECURITY — like the spoke dialer (see WebSocketDialer), this endpoint relies on the DEPLOYMENT for transport security. The route is unauthenticated until the enrollment frame arrives (the handshake IS the auth), so it must be served over TLS (terminate wss:// at a trusted reverse proxy in front of the hub) so the join token / bearer credential a spoke presents are not exposed on the wire. The credential is compared timing-safely against a stored hash, but it is a static bearer secret: protect it in transit and at rest, and revoke a spoke if it may have leaked.
@arg w The response writer (upgraded to a WebSocket). @arg r The upgrade request.
@testcase TestHubEnrollAndRoute drives a real spoke through this handler over loopback. @testcase TestHubRejectsBadEnrollment closes the connection when enrollment is rejected.
func (*Hub) Disconnect ¶
Disconnect force-closes the live connection for a named spoke, if any, so the spoke is dropped immediately (e.g. after an admin revokes its enrollment). The read loop tears down and unregisters the connection as a result; disconnecting an unknown or already-gone spoke is a no-op. It does not delete the spoke's enrolled record — the caller does that so the spoke cannot simply reconnect.
@arg name The spoke name whose live connection should be closed.
@testcase TestHubDisconnectClosesConnection closes a connected spoke's link.
func (*Hub) Spoke ¶
func (h *Hub) Spoke(name string) (BoxManager, bool)
Spoke returns the connected spoke with the given name as a BoxManager.
@arg name The spoke name. @return BoxManager The connected spoke, or nil when not connected. @return bool True when a spoke with that name is currently connected.
@testcase TestHubEnrollAndRoute looks up the enrolled spoke by name.
func (*Hub) Spokes ¶
func (h *Hub) Spokes() map[string]BoxManager
Spokes returns a snapshot of the currently connected spokes keyed by name.
@return map[string]BoxManager One entry per connected spoke.
@testcase TestHubEnrollAndRoute lists the connected spokes after enrollment.
type HubCaller ¶
type HubCaller struct {
// contains filtered or unexported fields
}
HubCaller lets spoke-side components (the per-box port API) issue requests to the hub over whichever cluster connection is currently live. A spoke creates exactly one HubCaller for its whole lifetime and hands it to both the box backend and RunWithCaller: each (re)connection attaches its transport after enrollment and detaches it when the connection drops, failing the calls that were in flight. Calls made while detached fail immediately with a clear "not connected to hub" error instead of blocking.
func NewHubCaller ¶
func NewHubCaller() *HubCaller
NewHubCaller returns a detached HubCaller; calls fail until a connection is attached by RunWithCaller.
@return *HubCaller A caller with no live connection yet.
@testcase TestSpokeCallerDisconnected fails a call made before any connection is attached.
func (*HubCaller) CloseBoxPort ¶
CloseBoxPort asks the hub to unpublish a port of the given box.
@arg ctx Context for the call. @arg boxID The spoke-stamped identity of the box the request originated from. @arg port The published port to close. @error error if the spoke is disconnected or the hub rejects the request.
@testcase TestSpokeCallerRoundTrip closes a port through the caller.
func (*HubCaller) ListBoxPorts ¶
ListBoxPorts asks the hub for the given box's published ports.
@arg ctx Context for the call. @arg boxID The spoke-stamped identity of the box the request originated from. @return []BoxPortInfo The box's published ports. @error error if the spoke is disconnected or the hub rejects the request.
@testcase TestSpokeCallerRoundTrip lists ports through the caller.
func (*HubCaller) OpenBoxPort ¶
func (c *HubCaller) OpenBoxPort(ctx context.Context, boxID string, port int, description string) (BoxPortInfo, error)
OpenBoxPort asks the hub to publish a port of the given box and returns the public view of the published port.
@arg ctx Context for the call. @arg boxID The spoke-stamped identity of the box the request originated from. @arg port The TCP port inside the box to publish. @arg description The caller's free-form label for the port. @return BoxPortInfo The published port with its public URL. @error error if the spoke is disconnected or the hub rejects the request.
@testcase TestSpokeCallerRoundTrip opens a port through the caller.
type JoinTokenInfo ¶
JoinTokenInfo describes an outstanding join token for listing/revocation. ID is the token's hash (an opaque handle the operator can revoke by); the secret is never recoverable. Backend is the box backend recorded at creation (empty means docker), kept so setup instructions can be re-rendered after creation.
type JoinTokenRecord ¶
type JoinTokenRecord struct {
Name string `json:"name"`
Backend string `json:"backend,omitempty"`
ExpiresAt time.Time `json:"expires_at"`
}
JoinTokenRecord is the stored form of a one-time join token: the spoke name baked into it, the box backend the spoke was created for (presentation only — enrollment does not check it; empty on records that predate it means docker), and when it expires. The secret itself is not stored (only its hash, which is the key).
type SpokeRecord ¶
type SpokeRecord struct {
Name string `json:"name"`
CredentialHash string `json:"credential_hash"`
EnrolledAt time.Time `json:"enrolled_at"`
}
SpokeRecord is an enrolled spoke: its name, the hash of its bearer credential, and when it enrolled.
type Store ¶
type Store interface {
// PutJoinToken stores a join token record keyed by the hash of its secret.
PutJoinToken(hash string, rec JoinTokenRecord) error
// TakeJoinToken atomically reads and removes the record for a token hash
// (one-time use); found is false when no token matches.
TakeJoinToken(hash string) (rec JoinTokenRecord, found bool, err error)
// ListJoinTokens returns every outstanding join token (its hash as an opaque
// ID, its spoke name, and its expiry — never the secret, which is not stored).
ListJoinTokens() ([]JoinTokenInfo, error)
// DeleteJoinToken removes a join token by its hash ID; deleting a missing one
// is a no-op.
DeleteJoinToken(hash string) error
// PutSpoke stores (creating or replacing) an enrolled spoke keyed by name.
PutSpoke(name string, rec SpokeRecord) error
// GetSpoke returns the spoke record for name; found is false when none matches.
GetSpoke(name string) (rec SpokeRecord, found bool, err error)
// ListSpokes returns every enrolled spoke.
ListSpokes() ([]SpokeRecord, error)
// DeleteSpoke removes an enrolled spoke; deleting a missing name is a no-op.
DeleteSpoke(name string) error
}
Store persists cluster enrollment state: one-time join tokens and the per-spoke bearer credentials minted from them. Secrets are only ever stored hashed (the plaintext join token is shown to the operator once; the plaintext credential is held only by the spoke). All methods must be safe for concurrent use. The bolt-backed implementation lives in the server package.
Source Files