kernelnodeops

type Acceptance ¶

type Acceptance struct {
	Channel             kernelnodeopsv1.Channel
	NodeRevision        uint64
	KernelOperationID   string
	ForwardRuntimeJobID uint64
}

Acceptance is what the channel reported when it took the operation.

type AdminError ¶

type AdminError struct {
	Code      string `json:"code"`
	Message   string `json:"message"`
	Retryable bool   `json:"retryable"`
}

AdminError is how an operation failed.

type AdminFanOut ¶

type AdminFanOut struct {
	Total     uint32 `json:"total"`
	Succeeded uint32 `json:"succeeded"`
	Failed    uint32 `json:"failed"`
	Pending   uint32 `json:"pending"`
}

AdminFanOut counts a fan-out's operations.

type AdminOperation ¶

type AdminOperation struct {
	Cursor              uint64          `json:"cursor"`
	OperationID         string          `json:"operation_id"`
	RequestID           string          `json:"request_id"`
	Digest              string          `json:"digest"`
	PackageID           string          `json:"package_id"`
	PackageGeneration   uint64          `json:"package_generation"`
	SubmittedBy         string          `json:"submitted_by"`
	Family              string          `json:"family"`
	Kind                string          `json:"kind"`
	State               string          `json:"state"`
	Terminal            bool            `json:"terminal"`
	Channel             string          `json:"channel"`
	Attempt             uint32          `json:"attempt"`
	NodeRevision        uint64          `json:"node_revision"`
	Resource            string          `json:"resource"`
	Targets             []AdminTarget   `json:"targets"`
	ParentOperationID   string          `json:"parent_operation_id"`
	FanOut              *AdminFanOut    `json:"fan_out,omitempty"`
	Reason              string          `json:"reason"`
	Operation           json.RawMessage `json:"operation"`
	Result              json.RawMessage `json:"result,omitempty"`
	Error               *AdminError     `json:"error,omitempty"`
	Evidence            json.RawMessage `json:"evidence,omitempty"`
	KernelOperationID   string          `json:"kernel_operation_id"`
	ForwardRuntimeJobID uint64          `json:"forward_runtime_job_id"`
	CancelRequestedAt   *time.Time      `json:"cancel_requested_at"`
	CancelReason        string          `json:"cancel_reason"`
	CreatedAt           time.Time       `json:"created_at"`
	AcceptedAt          *time.Time      `json:"accepted_at"`
	FinishedAt          *time.Time      `json:"finished_at"`
	DeadlineAt          time.Time       `json:"deadline_at"`
	UpdatedAt           time.Time       `json:"updated_at"`
}

AdminOperation is one ledger row as administrators audit it: who asked for what, on which nodes, and how it went. It holds no secret: the operation is canonical, results and errors were scrubbed when stored.

type AdminPage ¶

type AdminPage struct {
	Operations []AdminOperation `json:"operations"`
	NextBefore uint64           `json:"next_before"`
}

AdminPage is one page of the listing, newest first. NextBefore pages on; zero on the last page.

func AdminList(ctx context.Context, db *gorm.DB, query AdminQuery) (AdminPage, error)

type AdminQuery ¶

type AdminQuery struct {
	PackageID   string
	Family      string
	Kind        string
	State       string
	RequestID   string
	OperationID string
	Parent      string
	TargetKind  string
	TargetID    uint64
	// Before pages: operations listed before the one with this cursor.
	Before uint64
	Limit  int
}

AdminQuery filters the administrator's listing of node operations (GET /api/v4/kernel/node-operations).

func ParseAdminQuery(get func(string) string) (AdminQuery, error)

type AdminTarget ¶

type AdminTarget struct {
	Kind string `json:"kind"`
	ID   uint64 `json:"id"`
}

AdminTarget is one node an operation acts on.

type AgentDiagnostic ¶

type AgentDiagnostic struct {
	Task agentstreams.DiagnosticTask
	// Row is the task's row as the legacy routes answer it: reloaded after
	// the dispatch, or as created when the task could not be sent.
	Row      *model.AgentDiagnosticTask
	Dispatch agentstreams.DiagnosticDispatch
}

AgentDiagnostic is what RunAgentDiagnostic did.

func RunAgentDiagnostic(ctx context.Context, db *gorm.DB, request AgentDiagnosticRequest, transport agentstreams.DiagnosticTransport) (*AgentDiagnostic, error)

type AgentDiagnosticRequest ¶

type AgentDiagnosticRequest struct {
	NodeID uint
	Action string
	Params map[string]any
	// Timeout is the task's timeout as the agent receives it.
	Timeout int
}

AgentDiagnosticRequest is one diagnostic task for a proxy node's agent, as POST /admin/agent/tasks and /admin/agent/execute take it.

type AgentDirectory ¶

type AgentDirectory interface {
	// Advertises reports whether node's agent holds a session whose Hello
	// lists the capability name at version.
	Advertises(node *kernelnodeopsv1.NodeRef, name, version string) bool
}

AgentDirectory answers what nodes' agents advertise on their live Agent Control sessions.

type AgentOperation ¶

type AgentOperation struct {
	Desired *agentv1pb.DesiredOperation
	Ack     *agentv1pb.OperationAck
	// contains filtered or unexported fields
}

AgentOperation is a dispatched Agent Control operation.

func DispatchAgentOperation(ctx context.Context, streams agentstreams.Streams, node agentcontrol.AgentNode, request AgentOperationRequest) (*AgentOperation, error)

func (o *AgentOperation) Release()

type AgentOperationRequest ¶

type AgentOperationRequest struct {
	// OperationID is the stream's operation id; one is generated when
	// empty.
	OperationID string
	// Kind is agent.ping, node.reload or users.reload.
	Kind        string
	PayloadJSON []byte
	// Timeout bounds the wait for the acknowledgement:
	// DefaultAgentAckTimeout when zero, at most MaxAgentAckTimeout.
	Timeout time.Duration
}

AgentOperationRequest is one Agent Control operation to send on a node's stream, as POST /admin/nodes/:id/agent-control/operations takes it.

type AgentSources ¶

type AgentSources struct {
	Streams    agentstreams.Streams
	WebSockets agentstreams.WebSockets
}

AgentSources are the live agent sessions the executors dispatch on and the agent session RPCs read: the Agent Control streams of every node kind, and the legacy WebSocket. Either may be nil in a kernel that does not hold it.

func DefaultAgentSources() AgentSources

type Authorizer ¶

type Authorizer interface {
	AuthorizeCapability(ctx context.Context, host packagebridge.HostIdentity, capability string) error
}

Authorizer admits a calling host to a capability.

type BoundRequest ¶

type BoundRequest struct {
	RequestID string
	RouteID   string
	// ActorID and Admin are the request's kernel-authenticated principal.
	ActorID  uint64
	Admin    bool
	Deadline time.Time
	// contains filtered or unexported fields
}

BoundRequest is a verified request binding: the live v2 request the calling package is serving, as the kernel's gateway dispatched it. It is never stored.

type ConnectionTest ¶

type ConnectionTest struct {
	// Probes resolves the host for the public-address check; the zero
	// value uses the system resolver.
	Probes service.DiagnosisProbes
	// Now purges the held tokens; time.Now when nil.
	Now func() time.Time
	// contains filtered or unexported fields
}

ConnectionTest executes diagnose.forward_backend (node-ops-service.md section 6.1, NO-7): the gost API connection test of POST /admin/forward/test-connection, run from the kernel with the token the administrator typed. The gost-mesh package host submits the operation with its request binding and the sealed handle of the token; the Preparer resolves the handle in the submitting call and keeps the value in kernel memory until the executor dials. The token is scrubbed from the result and never recorded.

The dial follows the kernel's legacy handler: the same gost client, the same calls, the same error texts, so the route's answer is byte for byte what it was. A refusing or unreachable API is the result, not a failed operation. The SSRF rule of the diagnoses (#84, #90) applies: a request that is not an administrator's dials public addresses only.

func (c *ConnectionTest) Execute(ctx context.Context, run *Run) Outcome

func (c *ConnectionTest) Held() int

func (c *ConnectionTest) Prepare(_ context.Context, submission *Submission) error

func (c *ConnectionTest) Register(registry *Registry) error

type Credentials ¶

type Credentials struct {
	// PKI returns the agent PKI that issues enrollment credentials, on the
	// operation's transaction; nil uses the kernel's configuration
	// (agentpki.FromConfig).
	PKI func(ctx context.Context, tx *gorm.DB) (*agentpki.Service, error)
}

Credentials executes the credentials family (node-ops-service.md section 3.3, NO-5): IssueCredential, RevokeCredential, IssueRegistrationKey, RevokeRegistrationKey and IssueCleanAgent. Each runs the function of internal/service the legacy route runs, in one transaction with every nodesecrets.Sync and every agent certificate revocation, so native and legacy write identical rows.

Secrets in, secrets out. A typed secret arrives as a sealed handle: the Preparer resolves it for the bound request (Submission.Unseal) and keeps it for the executor in memory. A generated secret is minted as a handle (Run.Reveal) inside the transaction that stores it: when the answer cannot show it, nothing is stored. A credential is issued only in an administrator's request (the binding), never outside one, and results never carry a value in clear.

func (c *Credentials) Register(registry *Registry) error

type DesiredConfig ¶

type DesiredConfig struct {
	Node     agentcontrol.AgentNode
	Format   string
	Document map[string]any
	// JSON is the canonical document: sorted keys, no whitespace. Hash is
	// its SHA-256 (hex). The same configuration always has the same hash.
	JSON []byte
	Hash string
	// ExcludedProtocols counts the protocol rows left out because they
	// failed validation; zero while validation is report-only (section 3.8).
	ExcludedProtocols uint32
}

DesiredConfig is a node's desired configuration as the kernel built it from the current rows. The document holds the node's runtime secrets; it is stored in the protected table and never answered.

func BuildDesiredConfig(db *gorm.DB, node agentcontrol.AgentNode) (*DesiredConfig, error)

type Diagnosis ¶

type Diagnosis struct {
	// Probes is how the diagnoses reach the network. The zero value dials
	// from Control, with the system resolver, at most
	// service.DefaultDiagnosisConcurrency probes at once per operation.
	Probes service.DiagnosisProbes
	// Agents answers which nodes' agents advertise diag.v1; nil answers
	// none. No forward node holds an Agent Control session before A2-1, so
	// the kernel's default has none.
	Agents AgentDirectory
	// Now stamps the results; time.Now when nil.
	Now func() time.Time
}

Diagnosis executes the diagnose family's kinds (node-ops-service.md section 3.3, NO-8): CheckEndpoints, CollectNodeStats, DiagnoseForward and DiagnoseTunnel. Each runs the kernel's implementation of the legacy route (internal/service), which the legacy handlers run too, so its result is what the route computes. Results carry no credential: the nodes' tokens are named to the engine's scrubber, and diagnoses send none.

The vantage (D10) is Control by default; it would be the node when the agent of every node a diagnosis concerns advertises diag.v1. The kernel does not run diagnoses on agents yet (A2-5), so it dials from Control and says so in the result's VantageReport. A package host is never a vantage.

func (d *Diagnosis) Register(registry *Registry) error

type DiagnosticValidationError ¶

type DiagnosticValidationError struct {
	Err error
}

DiagnosticValidationError refuses a task the whitelist does not allow.

func (e *DiagnosticValidationError) Error() string

func (e *DiagnosticValidationError) Unwrap() error

type Engine ¶

type Engine struct {
	DB *gorm.DB
	// Executors serve the operation kinds; nil serves none.
	Executors *Registry
	// PollInterval is how often the dispatcher looks for work without a
	// wake-up; DefaultPollInterval when zero.
	PollInterval time.Duration
	// Timeout is an operation's deadline after its submission;
	// DefaultTimeout when zero.
	Timeout time.Duration
	// MaxRunning bounds the executors running at once.
	MaxRunning int
	// PackageQuota and NodeQuota bound the operations that have not ended,
	// per package and per target node.
	PackageQuota int
	NodeQuota    int
	// Now defaults to time.Now; it stamps the ledger.
	Now func() time.Time
	// Secrets holds the sealed handles of the requests in flight; nil
	// selects sealedsecrets.DefaultStore, which the v2 gateway seals into.
	Secrets *sealedsecrets.Store
	// contains filtered or unexported fields
}

Engine is the kernel's KernelNodeOps engine: the ledger, the dispatcher that hands pending operations to their executors, and the change notifications that waits and watches follow. One engine runs its dispatcher per database (Control is a single replica, section 8 R6).

func EngineFor(db *gorm.DB) *Engine

func (e *Engine) Prune(ctx context.Context, now time.Time) (int64, error)

func (e *Engine) Run(ctx context.Context)

func (e *Engine) Wake()

type Executor ¶

type Executor interface {
	Execute(ctx context.Context, run *Run) Outcome
}

Executor carries out the operations of one kind (node-ops-service.md section 3.3). The engine calls Execute once per dispatch, in its own goroutine, with a context that ends at the operation's deadline or when the operation is cancelled. Execute reports the channel's acceptance through Run.Accept and returns the outcome; the engine records it.

An executor must apply all of an operation or nothing it cannot repeat: a level-triggered kind may run again after a kernel restart, every other kind ends FAILED (retryable) when it was interrupted.

type ExecutorFunc ¶

type ExecutorFunc func(ctx context.Context, run *Run) Outcome

ExecutorFunc adapts a function to Executor.

func (f ExecutorFunc) Execute(ctx context.Context, run *Run) Outcome

type Forward ¶

type Forward struct {
	// JobPoll is how often a job backend's job is read; DefaultJobPoll when
	// zero.
	JobPoll time.Duration
	// Now stamps the results; time.Now when nil.
	Now func() time.Time
}

Forward executes the forward family's kinds (node-ops-service.md section 3.3, NO-7): ApplyForward, ApplyTunnel, SyncForwardBackend and ApplyLegacyRule. Each runs the kernel's implementation of the legacy route (internal/service), which the legacy handlers run too, over the executors the kernel already has: NodeX for the gost backend and the legacy rules, the local Ansible job executor and the clean agent job queue. The operation carries ids, never a payload: the kernel reads the rows when it executes, so a repeat converges on the latest state.

Credentials never leave the kernel. A job's stored payload carries no token; the gost executor resolves the ingress node's token when it sends the request to NodeX, and only for the node's pinned endpoint (section 3.8): an address that is not the pin ends ENDPOINT_UNCONFIRMED.

func (f *Forward) Register(registry *Registry) error

type NodeSync ¶

type NodeSync struct {
	Node agentcontrol.AgentNode
	// Channel is AGENT_CONTROL when the node's agent holds a stream, else
	// LEGACY_PULL: the node's next periodic pull reads the configuration.
	Channel kernelnodeopsv1.Channel
	Config  *DesiredConfig
	// Stored is the desired configuration row after the sync; Changed is
	// true when its hash moved (and its revision grew).
	Stored  model.KernelNodeDesiredConfig
	Changed bool
	// Pushed is true when a node.reload or a ConfigSnapshot went out on
	// the stream: the configuration changed, the sync was forced, or (for
	// a snapshot) the agent had not applied the stored configuration.
	Pushed bool
	// Desired and Ack describe a node.reload push; Snapshot and SessionID
	// a ConfigSnapshot push (the agent negotiated config.v1). DispatchErr
	// is why either push failed.
	Desired     *agentv1pb.DesiredOperation
	Ack         *agentv1pb.OperationAck
	Snapshot    *agentv1pb.ConfigSnapshot
	SessionID   string
	DispatchErr error
	// contains filtered or unexported fields
}

NodeSync is what a node sync did (node-ops-service.md section 5.5 and the node.sync kind of section 3.3).

func SyncNode(ctx context.Context, db *gorm.DB, streams agentstreams.Streams, node agentcontrol.AgentNode, opts SyncNodeOptions) (*NodeSync, error)

func (s *NodeSync) Release()

type Outcome ¶

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

Outcome is an executor's report: build it with Succeeded, Failed, Cancelled or FanOut.

func Cancelled(message string) Outcome

func Failed(code kernelnodeopsv1.ErrorCode, message string, retryable bool) Outcome

func FanOut(children ...*kernelnodeopsv1.OperationSpec) Outcome

func Succeeded(result *kernelnodeopsv1.OperationResult) Outcome

func (o Outcome) WithChannel(channel kernelnodeopsv1.Channel) Outcome

func (o Outcome) WithResult(result *kernelnodeopsv1.OperationResult) Outcome

type Preparer ¶

type Preparer interface {
	Prepare(ctx context.Context, submission *Submission) error
}

Preparer is implemented by executors that must act in the submitting call, before the operation is recorded: use the verified request binding, resolve sealed handles (Submission.Unseal), capture what a deletion removes. Prepare may replace submission.Operation; the canonical form of what it leaves is stored and executed, and must hold no sealed handle. An error refuses the submission and records nothing; a gRPC status error is answered as it is.

type QueryError ¶

type QueryError struct{ Message string }

QueryError is a filter the listing does not accept; its text says why.

func (e *QueryError) Error() string

type Registry ¶

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

Registry maps operation kinds to their executors. A kind without one is not served: SubmitOperation answers UNIMPLEMENTED and records nothing.

func NewRegistry() *Registry

func (r *Registry) Kinds() []string

func (r *Registry) Lookup(kindName string) (Executor, bool)

func (r *Registry) Register(kindName string, executor Executor) error

type Retirements ¶

type Retirements struct{}

Retirements executes RetireNode and RetireProtocol (node-ops-service.md section 3.3, NO-5; decision D11). The kernel performs the cascade of the legacy deletion in one transaction: a node's protocols with their users' WireGuard peers, their subscription group links and their secrets, the node's credentials and raw configuration secrets, and its agent certificates and enrollments; a protocol's peers, links and secrets. The package deletes its own row afterwards. The functions are the legacy deletions' (service.RetireProxyNodeTx, RetireForwardNodeTx, RetireProtocolTx), and every step is idempotent: a retirement of what is already gone succeeds with nothing counted.

func (r *Retirements) Register(registry *Registry) error

type Run ¶

type Run struct {
	OperationID       string
	RequestID         string
	PackageID         string
	Kind              string
	ParentOperationID string
	// Operation is the stored canonical operation.
	Operation *kernelnodeopsv1.OperationSpec
	Targets   []*kernelnodeopsv1.NodeRef
	Attempt   uint32
	Deadline  time.Time
	// Request is the submission's verified request binding while its
	// request is live: nil when it carried none, the request ended, or the
	// kernel restarted since. Reveal mints handles for it.
	Request *BoundRequest
	// contains filtered or unexported fields
}

Run is one dispatch of an operation, as its executor sees it.

func (r *Run) Accept(ctx context.Context, acceptance Acceptance) error

func (r *Run) DB() *gorm.DB

func (r *Run) Prepared() any

func (r *Run) Reveal(name, value string) (*kernelnodeopsv1.SecretHandle, error)

func (r *Run) UseSecret(values ...string)

type SecretDocuments ¶

type SecretDocuments struct{}

SecretDocuments executes PutSecretDocument (node-ops-service.md section 3.3, NO-5): the secret-keeping part of the protocol and raw configuration writes. In the document a package sends, a secret is a sealed handle (a value the administrator typed), the placeholder (keep the stored value) or empty (no value); a secret in clear is refused by the kind's check.

The Preparer resolves every handle for the bound request, at the field the gateway sealed it from (the column's pointer followed by the secret's), and keeps the values for the executor. The executor stores the document as the legacy routes store it (service.PutNodeProtocolSecretDocumentTx, PutNodeRawConfigDocumentTx): the stored values at the placeholders, the legacy column and the split table in one transaction, and answers the redacted document the package keeps in its own row.

func (s *SecretDocuments) Register(registry *Registry) error

type Server ¶

type Server struct {
	Engine     *Engine
	Authorizer Authorizer
	// SplitPhases answers the node credential split's phase per table
	// (SplitPhasesFrom). Nil answers LEGACY for every table.
	SplitPhases func(ctx context.Context) ([]*kernelnodeopsv1.TableSplitState, error)
	// Agents are the live agent sessions the session RPCs answer; nil reads
	// the defaults registered with UseAgentStreams and UseWebSocketAgents.
	Agents *AgentSources
}

Server holds what every host's KernelNodeOps calls share.

func (s *Server) For(host packagebridge.HostIdentity) kernelnodeopsv1.KernelNodeOpsServer

type SourcesFunc ¶

type SourcesFunc func() AgentSources

SourcesFunc returns the sources at the time of a call, so that a source registered after the executors were (the WebSocket handler is built with the router) is found when an operation runs.

type Submission ¶

type Submission struct {
	Host      packagebridge.HostIdentity
	Kind      string
	Operation *kernelnodeopsv1.OperationSpec
	// Binding is the request binding's bridge capability; it is never
	// stored.
	Binding []byte
	// Request is the verified request binding: the live v2 request the
	// package is serving. Nil when the submission carries no binding; a
	// binding that names no live request of the package is refused before
	// Prepare.
	Request *BoundRequest
	Targets []*kernelnodeopsv1.NodeRef
	// contains filtered or unexported fields
}

Submission is an operation in its submitting call.

func (s *Submission) Retain(value any)

func (s *Submission) Unseal(uses ...sealedsecrets.Use) ([]sealedsecrets.Secret, error)

type SyncNodeOptions ¶

type SyncNodeOptions struct {
	// Force pushes even when the configuration did not change, as the
	// administrator's sync button does.
	Force bool
	// AckTimeout bounds the wait for the agent's acknowledgement;
	// DefaultAgentAckTimeout when zero.
	AckTimeout time.Duration
	// Now stamps the desired configuration; time.Now when nil.
	Now func() time.Time
}

SyncNodeOptions adjust a node sync.