Help us shape the future of Go! Take the 2026 Go Developer Survey

broker

package
v0.36.0 Latest Latest
Warning

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

Go to latest
Published: Sep 19, 2026 License: Apache-2.0, Apache-2.0 Imports: 21 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type AgentInfo added in v0.1.2

type AgentInfo struct {
	Hash        string                 `json:"hash"`
	Connection  wire.Connection        `json:"connection"`
	SocketPath  string                 `json:"socketPath"`
	ExpiresAt   time.Time              `json:"expiresAt"`
	Certificate sshcert.RawCertificate `json:"certificate"`
}

AgentInfo contains information about a running agent

type Auth

type Auth struct {
	// contains filtered or unexported fields
}

Auth caches a JWT obtained from fetch and refreshes it proactively before it expires.

Concurrency: Auth is safe for concurrent use. Concurrent Token() calls are coalesced (singleflight): the first caller invokes fetch, later callers join the in-flight attempt, receive a replay of its user-visible output so far, and share its result. fetch's context is canceled when every caller waiting on it has canceled its own context.

Locking invariants:

  • mu protects: token, expiresAt, inflight
  • the flight goroutine writes token/expiresAt back under mu on success; no other flight can start while one is in flight
  • fetch is immutable after NewAuth()

func NewAuth

func NewAuth(fetch TokenFunc) *Auth

NewAuth creates a new Auth backed by fetch.

func (*Auth) ForceRefresh added in v0.18.0

func (a *Auth) ForceRefresh(ctx context.Context, out io.Writer) (string, error)

ForceRefresh discards the cached token and fetches a new one. This is the 401 safety net: the CA rejected a token we believed was still valid (server-side revocation, clock skew), so we bypass the cache once and tell fetch (via force=true) that it must not hand back that same credential.

func (*Auth) Token

func (a *Auth) Token(ctx context.Context, out io.Writer) (string, error)

Token returns a JWT valid for at least expiryBuffer, fetching or refreshing it if needed. Concurrent callers share one in-flight fetch; joiners get a replay of its user-visible output so far.

type Broker

type Broker struct {
	// contains filtered or unexported fields
}

Broker manages authentication and per-connection SSH agents. Certificates are minted fresh from the CA for every match past the agents-map fast path - there is no cross-connection certificate cache.

Concurrency: Broker is safe for concurrent access from multiple RPC clients. The primary lock (b.lock) protects the agents map and coordinates with Auth.

Locking invariants:

  • b.lock protects: agents map (both reads and writes)
  • Auth has its own internal lock (auth.mu) - safe to call without b.lock
  • Match() only holds b.lock around agents-map access, never across auth or CA calls, so concurrent matches can share an in-flight auth attempt. Concurrent matches for the same connection may each request a certificate from the CA; ensureAgent() reconciles them (last one wins).
  • ensureAgent() acquires b.lock itself; do not call it with b.lock held

Immutable after New(): brokerSocketPath, agentSocketDir, caClient, inventoryClient, log Protected by b.lock: agents map Protected by closeOnce: brokerListener, done channel Self-synchronized: auth (has internal lock)

func New

func New(log slog.Logger, socketPath string, fetch TokenFunc, caClient *caclient.Client, publicCAURL string, inventoryClient *inventoryclient.Client, verifyIdentity IdentityVerifier, agentSocketDir string) (*Broker, error)

New creates a new Broker instance. This does not start listening - call Serve() to begin accepting connections.

func (*Broker) AgentSocketPath

func (b *Broker) AgentSocketPath(hash wire.ConnectionHash) string

AgentSocketPath returns the socket path for a given connection hash. This is used by SSH to connect to the per-connection agent.

func (*Broker) Close

func (b *Broker) Close()

func (*Broker) Done

func (b *Broker) Done() <-chan struct{}

func (*Broker) IdentityWithUserOutput added in v0.27.0

func (b *Broker) IdentityWithUserOutput(ctx context.Context, out io.Writer) IdentityResponse

IdentityWithUserOutput authenticates through the same cache and in-flight login as SSH requests. It does not request a certificate or require an inventory record, so administrators can use it to prepare that record.

func (*Broker) Inspect added in v0.1.2

func (b *Broker) Inspect(request InspectRequest, output *InspectResponse) error

Inspect is invoked by `epithet agent inspect` (over the broker's JSON line protocol) to get broker state.

func (*Broker) InventoryWithUserOutput added in v0.31.0

func (b *Broker) InventoryWithUserOutput(ctx context.Context, request inventoryapi.ControlRequest, out io.Writer) *InventoryResponse

InventoryWithUserOutput sends an administrative operation through the agent's CA-discovered inventory endpoint. Only the short-lived ID token goes to that endpoint; no token or refresh state is returned over the local protocol.

func (*Broker) Kill added in v0.25.0

func (b *Broker) Kill(request KillRequest, output *KillResponse) error

Kill evicts one per-connection agent and its in-memory credential. The next Match for the connection requests a fresh certificate. This does not revoke the issued certificate or disconnect an already-established SSH session.

func (*Broker) MatchWithUserOutput added in v0.7.0

func (b *Broker) MatchWithUserOutput(ctx context.Context, conn wire.Connection, userOutput io.Writer) MatchResponse

MatchWithUserOutput performs the match operation, streaming auth user output to userOutput. This is the core match implementation used by the JSON protocol server (protocol.go). Canceling ctx (e.g. the requesting `epithet match` process went away) abandons this match's auth and CA work.

func (*Broker) Ready added in v0.3.5

func (b *Broker) Ready() <-chan struct{}

Ready returns a channel that is closed when the broker is ready to accept connections. Use this to wait for the broker to start: <-b.Ready()

func (*Broker) Running

func (b *Broker) Running() bool

func (*Broker) Serve

func (b *Broker) Serve(ctx context.Context) error

Serve starts the broker listening on the configured socket and blocks until the context is cancelled. Returns an error if the listener cannot be started, otherwise returns ctx.Err() when shutdown completes.

func (*Broker) SetShutdownTimeout added in v0.1.2

func (b *Broker) SetShutdownTimeout(d time.Duration)

SetShutdownTimeout sets the timeout for waiting on in-flight RPCs during shutdown. Use 0 to skip waiting (useful for tests).

type CAEndpointInfo added in v0.17.0

type CAEndpointInfo struct {
	URL      string `json:"url"`
	Priority int    `json:"priority"`
	State    string `json:"state"` // "closed", "open", "half-open"
}

CAEndpointInfo contains the current state of a CA endpoint.

type Event added in v0.18.0

type Event struct {
	Inventory *InventoryResponse `json:"inventory,omitempty"`
	Identity  *IdentityResponse  `json:"identity,omitempty"`
	Output    string             `json:"output,omitempty"`
	Result    *MatchResponse     `json:"result,omitempty"`
	Inspect   *InspectResponse   `json:"inspect,omitempty"`
	Kill      *KillResponse      `json:"kill,omitempty"`
}

Event is one line of JSON sent by the broker in response to a Request. For a Match request: zero or more Output events (auth progress, e.g. the auth-code+PKCE URL to visit) followed by exactly one Result event. For an Inspect request: exactly one Inspect event. Identity requests stream auth Output events followed by one Identity event. Inventory requests similarly stream Output events followed by one Inventory event.

type Identity added in v0.27.0

type Identity struct {
	Issuer        string `json:"issuer"`
	Subject       string `json:"subject"`
	OID           string `json:"oid,omitempty"`
	Email         string `json:"email,omitempty"`
	EmailVerified *bool  `json:"email_verified,omitempty"`
}

Identity reports selected claims from a verified token, never bearer credentials. EmailVerified is the issuer assertion, not an inventory authorization decision.

type IdentityResponse added in v0.27.0

type IdentityResponse struct {
	Identity *Identity `json:"identity,omitempty"`
	Error    string    `json:"error,omitempty"`
}

IdentityResponse is the terminal event for an identity request.

type IdentityVerifier added in v0.27.0

type IdentityVerifier func(context.Context, string) (*Identity, error)

IdentityVerifier verifies the agent's ID token and extracts diagnostic claims. It must be safe for concurrent calls and must not return token contents in errors.

type InspectRequest added in v0.1.2

type InspectRequest struct {
	ID wire.ConnectionHash `json:"id,omitempty"`
}

InspectRequest optionally selects an agent by full ID or unique prefix. An empty ID returns all agents.

type InspectResponse added in v0.1.2

type InspectResponse struct {
	SocketPath     string           `json:"socketPath"`
	AgentSocketDir string           `json:"agentSocketDir"`
	Agents         []AgentInfo      `json:"agents"`
	CAEndpoints    []CAEndpointInfo `json:"caEndpoints"`
}

InspectResponse contains the current broker state

type InventoryResponse added in v0.35.0

type InventoryResponse struct {
	inventoryapi.ControlResponse
	CAURL string `json:"ca-url,omitempty"`
}

InventoryResponse adds local agent context to the service result. CAURL comes from the broker's CA discovery and is used by the CLI for enrollment instructions; it is not part of the remote inventory API.

type KillRequest added in v0.25.0

type KillRequest struct {
	ID wire.ConnectionHash `json:"id"`
}

KillRequest identifies one per-connection agent by the ID returned from Inspect. The ID may be a full OpenSSH connection hash or a unique prefix.

type KillResponse added in v0.25.0

type KillResponse struct {
	ID         wire.ConnectionHash `json:"id"`
	Connection wire.Connection     `json:"connection"`
	Error      string              `json:"error,omitempty"`
}

KillResponse confirms which agent was evicted. Error is populated on the wire when a protocol request cannot be completed.

type MatchResponse

type MatchResponse struct {
	// Should the `Match exec` actually match?
	Allow bool `json:"allow"`

	// Error contains any error which should be reported to the user on stderr
	Error string `json:"error,omitempty"`
}

type Request added in v0.18.0

type Request struct {
	Inventory *inventoryapi.ControlRequest `json:"inventory,omitempty"`
	Identity  *struct{}                    `json:"identity,omitempty"`
	Match     *wire.Connection             `json:"match,omitempty"`
	Inspect   *InspectRequest              `json:"inspect,omitempty"`
	Kill      *KillRequest                 `json:"kill,omitempty"`
}

Request is one line of JSON sent by the client. Exactly one field is set.

type TokenFunc added in v0.18.0

type TokenFunc func(ctx context.Context, out io.Writer, force bool) (string, error)

TokenFunc acquires a fresh JWT. force is true when the caller has already determined the current credential was rejected (ForceRefresh) and a genuinely new token is required, not merely "no cached token yet". Implementations backed by refreshable state (e.g. an oauth2.Token) must not treat force=true as an ordinary call and hand back a still-"valid" credential — that would silently defeat the 401 safety net. Invocations are serialized by Auth, so a stateful closure needs no locking of its own.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL