nodeapi

package
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Jul 18, 2026 License: Apache-2.0 Imports: 23 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, ciphertext string, nonce string) (string, error)

DecryptSecret decrypts an AES-256-GCM encrypted secret. The ciphertext and nonce are base64-encoded (standard encoding). The NSK (node secret key) must be exactly 32 bytes. Returns a generic error on failure to avoid leaking cryptographic details.

func HandleNodeSecretsUpdated 

func HandleNodeSecretsUpdated(cache *StateCache, logger *slog.Logger, env api.SignedEnvelope) error

HandleNodeSecretsUpdated parses the event payload and updates the secret index in the cache.

func HandleNodeStateUpdated 

func HandleNodeStateUpdated(cache *StateCache, logger *slog.Logger, env api.SignedEnvelope) error

HandleNodeStateUpdated parses the event payload and updates metadata and data entries in the cache.

func RegisterEventHandlers 

func RegisterEventHandlers(dispatcher *api.EventDispatcher, cache *StateCache, logger *slog.Logger)

RegisterEventHandlers registers SSE event handlers for node_state_updated and node_secrets_updated with the given dispatcher.

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 NodeSecretsUpdatePayload 

type NodeSecretsUpdatePayload struct {
	SecretRefs []api.SecretRef `json:"secret_refs"`
}

NodeSecretsUpdatePayload is the payload for node_secrets_updated events.

type NodeStateUpdatePayload 

type NodeStateUpdatePayload struct {
	Metadata map[string]string `json:"metadata"`
	Data     []api.DataEntry   `json:"data"`
}

NodeStateUpdatePayload is the payload for node_state_updated events.

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 {
	ActivePolicies() []api.Policy
}

PolicyProvider supplies active network policies.

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 {
	SyncReports(ctx context.Context, nodeID string, req api.ReportSyncRequest) error
}

ReportSyncClient is the interface for syncing reports to the control plane.

type ReportSyncer 

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

ReportSyncer buffers report changes and syncs them to the control plane with debouncing to coalesce rapid updates.

func NewReportSyncer 

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

NewReportSyncer creates a new ReportSyncer.

func (*ReportSyncer) NotifyChange 

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

NotifyChange buffers report changes and signals the run loop. Entries are appended; later entries for the same key overwrite earlier ones. Deleted keys are appended.

func (*ReportSyncer) Run 

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

Run loops, waiting for change notifications, debouncing, and flushing. It returns ctx.Err() when the context is cancelled.

type SecretFetcher 

type SecretFetcher interface {
	FetchSecret(ctx context.Context, nodeID, key string) (*api.SecretResponse, 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 RegisterEventHandlers and 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) ReconcileHandler 

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

ReconcileHandler returns a reconcile.ReconcileHandler that updates the cache when drift is detected in metadata, data, or secret refs.

func (*Server) RegisterEventHandlers 

func (s *Server) RegisterEventHandlers(dispatcher *api.EventDispatcher)

RegisterEventHandlers registers SSE event handlers with the given dispatcher.

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.

func (*StateCache) GetMetadataKey 

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

GetMetadataKey returns the value for a metadata key and whether it exists.

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) 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) 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.