nodeapi

package
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Jul 23, 2026 License: Apache-2.0 Imports: 25 Imported by: 0

Documentation

Index

Constants

View Source 
const DefaultDebouncePeriod = 5 * time.Second

DefaultDebouncePeriod is the default debounce period.

View Source 
const DefaultHTTPListen = "127.0.0.1:9100"

DefaultHTTPListen is the default HTTP listen address.

View Source 
const DefaultShutdownTimeout = 5 * time.Second

DefaultShutdownTimeout is the default graceful shutdown timeout.

View Source 
const DefaultSocketPath = "/var/run/plexd/api.sock"

DefaultSocketPath is the default Unix domain socket path.

Variables

View Source 
var (
	// ErrVersionConflict is returned when an optimistic locking check fails.
	ErrVersionConflict = errors.New("nodeapi: version conflict")
	// ErrNotFound is returned when the requested entry does not exist.
	ErrNotFound = errors.New("nodeapi: not found")
)

Functions

func BearerAuthMiddleware 

func BearerAuthMiddleware(token string) func(http.Handler) http.Handler

BearerAuthMiddleware returns middleware that validates Bearer token authentication. Requests without a valid token receive 401 Unauthorized. Unix socket requests bypass this middleware (it is only applied to the TCP listener).

func DecryptSecret 

func DecryptSecret(nsk []byte, envelope []byte) (string, error)

DecryptSecret opens the raw AES-256-GCM envelope <12-byte nonce> || <ciphertext + 16-byte GCM tag> under the NSK. The NSK (node secret key) must be exactly 32 bytes. Returns a generic error on every failure to avoid leaking cryptographic details.

func SecretAuthMiddleware 

func SecretAuthMiddleware(checker GroupChecker, getter PeerCredGetter, logger *slog.Logger) func(http.Handler) http.Handler

SecretAuthMiddleware returns HTTP middleware that restricts access to secret endpoints. Access is granted to root (UID 0) or processes whose user is a member of the plexd-secrets group.

The middleware extracts peer credentials from the request's underlying connection using a PeerCredGetter. In production, this is backed by SO_PEERCRED; in tests, a mock can be injected.

func SetSocketPermissions 

func SetSocketPermissions(socketPath string, logger *slog.Logger) error

SetSocketPermissions sets ownership and permissions on the Unix socket file. If the plexd group exists, the socket is chowned to root:plexd with mode 0660. If the group does not exist, the socket gets mode 0666 and a warning is logged.

Types

type ActionProvider 

type ActionProvider interface {
	Capabilities() ([]api.ActionInfo, []api.HookInfo)
}

ActionProvider supplies action and hook information to the local API.

type Config 

type Config struct {
	// SocketPath is the path to the Unix domain socket.
	// Default: /var/run/plexd/api.sock
	SocketPath string `yaml:"socket_path"`

	// HTTPEnabled enables the optional HTTP listener.
	// Default: false
	HTTPEnabled bool `yaml:"http_enabled"`

	// HTTPListen is the HTTP listen address.
	// Default: 127.0.0.1:9100
	HTTPListen string `yaml:"http_listen"`

	// HTTPTokenFile is the path to the HTTP bearer token file.
	HTTPTokenFile string `yaml:"http_token_file"`

	// DebouncePeriod is the debounce period for coalescing events.
	// Default: 5s
	DebouncePeriod time.Duration `yaml:"debounce_period"`

	// ShutdownTimeout is the maximum time to wait for a graceful shutdown.
	// Default: 5s
	ShutdownTimeout time.Duration `yaml:"shutdown_timeout"`

	// DataDir is the path to the data directory (required).
	DataDir string `yaml:"data_dir"`

	// SecretAuthEnabled enables SO_PEERCRED-based authentication for
	// /v1/state/secrets/* routes on the Unix socket. When enabled, only
	// root (UID 0) or plexd-secrets group members may access secrets.
	// Default: false (enabled by cmd/plexd/cmd/up.go in production).
	SecretAuthEnabled bool `yaml:"secret_auth_enabled"`
}

Config holds the configuration for the local node API server. Config is passed as a constructor argument — no file I/O in this package.

func (*Config) ApplyDefaults 

func (c *Config) ApplyDefaults()

ApplyDefaults sets default values for zero-valued fields.

func (*Config) Validate 

func (c *Config) Validate() error

Validate checks that required fields are set and values are acceptable.

type ForwarderStatus 

type ForwarderStatus struct {
	Enabled      bool   `json:"enabled"`
	BufferSize   int    `json:"buffer_size"`
	SourceCount  int    `json:"source_count"`
	ErrorCount   int    `json:"error_count"`
	LastReportAt string `json:"last_report_at,omitempty"`
}

ForwarderStatus describes the operational status of a log or audit forwarder.

type ForwarderStatusProvider 

type ForwarderStatusProvider interface {
	ForwarderStatus() ForwarderStatus
}

ForwarderStatusProvider supplies forwarder status information.

type GroupChecker 

type GroupChecker interface {
	// IsInGroup reports whether the user identified by uid belongs to the
	// named group, or if the user's primary group (gid) matches the group.
	IsInGroup(uid, gid uint32, groupName string) bool
}

GroupChecker checks group membership for a given user.

type Handler 

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

Handler provides HTTP handlers for the local node API.

func NewHandler 

func NewHandler(cache *StateCache, secretFetcher SecretFetcher, nodeID string, nsk []byte, logger *slog.Logger) *Handler

NewHandler creates a new Handler.

func (*Handler) Mux 

func (h *Handler) Mux() *http.ServeMux

Mux returns a configured ServeMux with all local node API routes.

func (*Handler) SetActionProvider 

func (h *Handler) SetActionProvider(provider ActionProvider)

SetActionProvider sets the action provider for action/hook endpoints.

func (*Handler) SetAuditStatus 

func (h *Handler) SetAuditStatus(p ForwarderStatusProvider)

SetAuditStatus sets the audit forwarder status provider.

func (*Handler) SetHookReloader 

func (h *Handler) SetHookReloader(reloader HookReloader)

SetHookReloader sets the hook reloader for the reload endpoint.

func (*Handler) SetLogStatus 

func (h *Handler) SetLogStatus(p ForwarderStatusProvider)

SetLogStatus sets the log forwarder status provider.

func (*Handler) SetPeerProvider 

func (h *Handler) SetPeerProvider(p PeerProvider)

SetPeerProvider sets the peer status provider.

func (*Handler) SetPolicyProvider 

func (h *Handler) SetPolicyProvider(p PolicyProvider)

SetPolicyProvider sets the policy provider.

type HookReloader 

type HookReloader interface {
	Hooks() []api.HookInfo
}

HookReloader triggers a re-scan of hooks from the filesystem.

type LocalActionRunner 

type LocalActionRunner interface {
	RunLocal(ctx context.Context, action string, params map[string]string) (stdout, stderr string, exitCode int, err error)
}

LocalActionRunner runs a built-in action synchronously and returns output.

type NodeAPIClient 

type NodeAPIClient interface {
	SecretFetcher
	ReportSyncClient
}

NodeAPIClient combines the control plane methods needed by the node API server.

type OSGroupChecker 

type OSGroupChecker struct{}

OSGroupChecker checks group membership using the OS user/group database.

func (OSGroupChecker) IsInGroup 

func (OSGroupChecker) IsInGroup(uid, gid uint32, groupName string) bool

type PeerCredGetter 

type PeerCredGetter interface {
	GetPeerCredentials(r *http.Request) (*PeerCredentials, error)
}

PeerCredGetter extracts peer credentials from an HTTP request's underlying connection.

type PeerCredentials 

type PeerCredentials struct {
	PID uint32
	UID uint32
	GID uint32
}

PeerCredentials holds the peer credentials extracted from a Unix socket connection.

func GetPeerCredentials 

func GetPeerCredentials(conn net.Conn) (*PeerCredentials, error)

GetPeerCredentials extracts peer credentials from a Unix socket connection using the SO_PEERCRED socket option. Returns an error if the connection is not a Unix socket or the credentials cannot be retrieved.

type PeerProvider 

type PeerProvider interface {
	PeerStatuses() []PeerStatus
}

PeerProvider supplies mesh peer information.

type PeerStatus 

type PeerStatus struct {
	ID        string `json:"id"`
	PublicKey string `json:"public_key"`
	MeshIP    string `json:"mesh_ip"`
	Endpoint  string `json:"endpoint"`
}

PeerStatus describes a single mesh peer's status.

type PolicyProvider 

type PolicyProvider interface {
	ActivePolicy() *api.PolicySnapshot
}

PolicyProvider supplies the active merged network policy.

type ReportEntry 

type ReportEntry struct {
	Key         string          `json:"key"`
	ContentType string          `json:"content_type"`
	Payload     json.RawMessage `json:"payload"`
	Version     int             `json:"version"`
	UpdatedAt   time.Time       `json:"updated_at"`
}

ReportEntry represents a locally-managed report entry.

type ReportSyncClient 

type ReportSyncClient interface {
	PutStateReport(ctx context.Context, nodeID, key string, req api.NodeStateReportRequest) (*api.NodeStateReportResponse, error)
	DeleteStateReport(ctx context.Context, nodeID, key string) error
}

ReportSyncClient publishes per-key node state reports to the control plane.

type ReportSyncer 

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

ReportSyncer reconciles per-key report changes to the control plane. Changes are held in a dirty map keyed by report key; a nil value marks a pending delete. NotifyChange debounces to coalesce bursts, flush walks the dirty keys in ascending order publishing one at a time, and keys left dirty by a retryable failure are re-flushed on a timer so state converges without a new local mutation.

func NewReportSyncer 

func NewReportSyncer(client ReportSyncClient, debouncePeriod time.Duration, logger *slog.Logger) *ReportSyncer

NewReportSyncer creates a ReportSyncer. The node ID is supplied later to Run because the syncer is constructed in NewServer, before the node has registered and its ID is known.

func (*ReportSyncer) NotifyChange 

func (s *ReportSyncer) NotifyChange(entries []ReportEntry, deleted []string)

NotifyChange merges report changes into the dirty map and wakes the run loop. entries are pending PUTs and deleted keys are pending deletes; the last change to a given key wins.

func (*ReportSyncer) Run 

func (s *ReportSyncer) Run(ctx context.Context, nodeID string) error

Run reconciles report changes to the control plane for nodeID until ctx is cancelled, at which point it returns ctx.Err(). Dirty state is preserved in memory across the return so a subsequent Run resumes where this one left off.

type SecretFetcher 

type SecretFetcher interface {
	FetchSecret(ctx context.Context, nodeID, name string, version int) (*api.SecretEnvelope, error)
}

SecretFetcher abstracts the control plane client for secret retrieval.

type Server 

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

Server is the local node API server. It serves HTTP over a Unix socket and optionally over TCP with bearer token authentication.

func NewServer 

func NewServer(cfg Config, client NodeAPIClient, nsk []byte, logger *slog.Logger) *Server

NewServer creates a new Server. Config defaults are applied automatically. The cache is initialized eagerly so that ReconcileHandler can be called before Start.

func (*Server) Cache 

func (s *Server) Cache() *StateCache

Cache returns the server's state cache for use by SSE event handlers.

func (*Server) PublishReport added in v0.2.0 

func (s *Server) PublishReport(key, contentType string, payload json.RawMessage) error

PublishReport writes a report entry through the cache and notifies the syncer so it converges to the control plane. It is the seam through which internal producers publish status blocks, and it holds key and payload to the same grammar and 4096-byte value cap as the local HTTP API. content_type and the resulting version stay local-only; the syncer ships only the payload value.

func (*Server) ReconcileHandler 

func (s *Server) ReconcileHandler() reconcile.ReconcileHandler

ReconcileHandler returns a reconcile.ReconcileHandler that refreshes the cache from the desired state block whenever it changes. The block's metadata bucket becomes the cache metadata map, and its opaque data bucket feeds the versioned data store (GET /v1/state/data): each api.StateEntry value is JSON-encoded into an api.DataEntry payload under dataStateContentType. A nil state block authoritatively clears both buckets.

The secret index is not fed here: secret references no longer ride the snapshot and secret values are always fetched live, so the pull leaves it untouched.

func (*Server) ReportPayload added in v0.2.0 

func (s *Server) ReportPayload(key string) (json.RawMessage, bool)

ReportPayload returns the payload currently stored under key, and whether the key exists. Internal producers use it to compare what is actually published against what they would publish, so a report another local caller overwrote or deleted is re-asserted rather than assumed to still hold their last value.

func (*Server) SetActionProvider 

func (s *Server) SetActionProvider(provider ActionProvider)

SetActionProvider sets the action provider for action/hook endpoints. Must be called before Start.

func (*Server) SetAuditStatus 

func (s *Server) SetAuditStatus(p ForwarderStatusProvider)

SetAuditStatus sets the audit forwarder status provider. Must be called before Start.

func (*Server) SetHookReloader 

func (s *Server) SetHookReloader(reloader HookReloader)

SetHookReloader sets the hook reloader for the reload endpoint. Must be called before Start.

func (*Server) SetLogStatus 

func (s *Server) SetLogStatus(p ForwarderStatusProvider)

SetLogStatus sets the log forwarder status provider. Must be called before Start.

func (*Server) SetPeerProvider 

func (s *Server) SetPeerProvider(p PeerProvider)

SetPeerProvider sets the peer status provider. Must be called before Start.

func (*Server) SetPolicyProvider 

func (s *Server) SetPolicyProvider(p PolicyProvider)

SetPolicyProvider sets the policy provider. Must be called before Start.

func (*Server) Start 

func (s *Server) Start(ctx context.Context, nodeID string) error

Start initializes and runs the server. It blocks until ctx is cancelled.

type StateCache 

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

StateCache holds node state in memory with file persistence.

func NewStateCache 

func NewStateCache(dataDir string, logger *slog.Logger) *StateCache

NewStateCache creates a new StateCache with empty maps. dataDir is the base path; the state subdirectory tree will be created under dataDir/state/.

func (*StateCache) DeleteReport 

func (sc *StateCache) DeleteReport(key string) error

DeleteReport removes a report entry and its file. Returns ErrNotFound if the key does not exist.

func (*StateCache) GetData 

func (sc *StateCache) GetData() map[string]api.DataEntry

GetData returns a copy of the data map.

func (*StateCache) GetDataEntry 

func (sc *StateCache) GetDataEntry(key string) (api.DataEntry, bool)

GetDataEntry returns a data entry by key and whether it exists.

func (*StateCache) GetMetadata 

func (sc *StateCache) GetMetadata() map[string]string

GetMetadata returns a copy of the metadata map, with the delivery-mode diagnostic overlaid as the delivery_mode key when a mode is set.

func (*StateCache) GetMetadataKey 

func (sc *StateCache) GetMetadataKey(key string) (string, bool)

GetMetadataKey returns the value for a metadata key and whether it exists. The delivery_mode key resolves to the delivery-mode diagnostic.

func (*StateCache) GetReport 

func (sc *StateCache) GetReport(key string) (ReportEntry, bool)

GetReport returns a report entry by key and whether it exists.

func (*StateCache) GetReports 

func (sc *StateCache) GetReports() map[string]ReportEntry

GetReports returns a copy of the reports map.

func (*StateCache) GetSecretIndex 

func (sc *StateCache) GetSecretIndex() []api.SecretRef

GetSecretIndex returns a copy of the secret index.

func (*StateCache) Load 

func (sc *StateCache) Load() error

Load reads persisted state from disk. Missing files or directories are treated as fresh (empty) state. The directory tree is created if absent.

func (*StateCache) OrphanedReportKeys added in v0.2.0 

func (sc *StateCache) OrphanedReportKeys() []string

OrphanedReportKeys returns the keys of the persisted reports the last Load rejected as outside the current key grammar. A previous release may have synced them to the control plane, where nothing local can reach them any more, so the caller queues a delete for each.

func (*StateCache) PutReport 

func (sc *StateCache) PutReport(key, contentType string, payload json.RawMessage, ifMatch *int) (ReportEntry, error)

PutReport creates or updates a report entry. If the entry exists and ifMatch is non-nil, it must equal the current version or ErrVersionConflict is returned. Version starts at 1 for new entries and increments on update.

func (*StateCache) SetDeliveryMode added in v0.2.0 

func (sc *StateCache) SetDeliveryMode(mode string)

SetDeliveryMode records the SSE delivery channel diagnostic. It is stored separately from the snapshot-owned metadata map and is intentionally not persisted: the value is re-seeded from the live SSE manager on every startup.

func (*StateCache) UpdateData 

func (sc *StateCache) UpdateData(entries []api.DataEntry)

UpdateData replaces data entries in memory and persists each to data/{key}.json. Files for entries no longer present are removed.

func (*StateCache) UpdateMetadata 

func (sc *StateCache) UpdateMetadata(m map[string]string)

UpdateMetadata replaces the metadata in memory and persists to metadata.json.

func (*StateCache) UpdateSecretIndex 

func (sc *StateCache) UpdateSecretIndex(refs []api.SecretRef)

UpdateSecretIndex replaces the secret index in memory and persists to secrets.json.

type StateSummary 

type StateSummary struct {
	Metadata   map[string]string  `json:"metadata"`
	DataKeys   []dataKeySummary   `json:"data_keys"`
	SecretKeys []secretKeySummary `json:"secret_keys"`
	ReportKeys []reportKeySummary `json:"report_keys"`
}

StateSummary is the response for GET /v1/state.

Source Files

View all Source files

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.