gateway

package
v0.4.3 Latest Latest
Warning

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

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

Documentation

Overview

Package gateway is the agent-facing gRPC surface of the control plane. It terminates TLS on one listener and serves two services split on the authentication boundary (ADR-0015): EnrollmentService, reachable with a provisioning token and no client certificate, and AgentService, reachable only with a verified workload credential. An interceptor enforces that split per service and is the only place identity is derived from a connection; handlers read it from the request context and nowhere else (ADR-0016).

The sync stream (ADR-0002, ADR-0015) lives here too: one live stream per workload, registered in an in-process map so that a render's announcement, or a directive addressed to one workload, delivered over the database's notification channel, finds the stream to push to. Stream state is the only state held in process, and a reconnect rebuilds it from a snapshot. Flow ingestion follows.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func DefaultSyncConfig

func DefaultSyncConfig() *innerwallv1.SyncConfig

DefaultSyncConfig is what the control plane hands every agent unless the operator configures otherwise. Centralizing these means fleet-wide tuning without an agent redeployment.

func IdentityFromContext

func IdentityFromContext(ctx context.Context) (identity.WorkloadID, bool)

IdentityFromContext returns the workload identity the interceptor derived from the connection credential. Handlers on AgentService call this and nothing else to learn who is calling.

func NewGRPCServer

func NewGRPCServer(tlsCfg *tls.Config, srv *Server, opts ...grpc.ServerOption) *grpc.Server

NewGRPCServer builds a gRPC server with the listener's TLS configuration (ServerTLSConfig) and the per-service authentication interceptors, which re-verify client chains against the same bundle, and registers both services on it.

func ServerTLSConfig

func ServerTLSConfig(serverCert tls.Certificate, bundlePEM []byte) (*tls.Config, error)

ServerTLSConfig builds the listener's TLS configuration: the control plane's own certificate, and client certificates verified against the authority bundle when presented. Presenting none is allowed at the transport layer because EnrollmentService is authenticated by a token; the interceptor decides per service whether a certificate was required.

The handshake verifies the chain, the client-auth usage, and possession of the key, but does not judge time: a credential that has expired may still renew within the grace, and only the interceptor knows which method is being called. It then re-verifies the same chain and applies the enrollment domain's validity rule per method, so a credential that is expired, or not yet valid, is refused there with a reason the agent can log, rather than as an anonymous handshake failure.

func WithIdentity

func WithIdentity(ctx context.Context, id identity.WorkloadID) context.Context

WithIdentity returns a context carrying id. Exported for tests of handlers; production code only ever calls it from the interceptor.

Types

type Deps

type Deps struct {
	Enroll   *enroll.Service
	Registry registry.Store
	Policies PolicyStore
	Engine   *compiler.Engine
	Events   PolicyEvents
	// Ingest receives reported flow windows; ReportFlows refuses with
	// FailedPrecondition when nil.
	Ingest *ingest.Service
	// SyncConfig is handed to every agent in HelloAck; DefaultSyncConfig
	// when nil.
	SyncConfig *innerwallv1.SyncConfig
	Log        *slog.Logger
	// Now is the clock; time.Now if nil.
	Now func() time.Time
}

Deps is what the gateway is built from. Enroll is required. Registry, Policies, Engine, and Events are required for the sync stream and Ingest for flow telemetry; a deployment that serves enrollment only, such as a test harness, may leave them nil, and the corresponding streams then refuse with FailedPrecondition.

type PolicyEvents

type PolicyEvents interface {
	ListenPolicyChanges(ctx context.Context, log *slog.Logger, onReady func(), fn func(compiler.Announcement), reconnect func(identity.WorkloadID)) error
}

PolicyEvents delivers render announcements and the directives addressed to one workload's stream. The store implements it over the database's notification channel, so a render or a directive in any process reaches the replica holding the workload's stream without anything polling (ADR-0002, ADR-0018 as amended).

type PolicyStore

type PolicyStore interface {
	// GetWorkloadPolicy returns the current rendered policy of a workload
	// with its version, or nil when none has been rendered.
	GetWorkloadPolicy(ctx context.Context, id identity.WorkloadID) (*innerwallv1.WorkloadPolicy, error)
}

PolicyStore reads persisted rendered policy.

type Server

type Server struct {
	innerwallv1.UnimplementedEnrollmentServiceServer
	innerwallv1.UnimplementedAgentServiceServer
	// contains filtered or unexported fields
}

Server implements EnrollmentService and AgentService.

func New

func New(d Deps) *Server

New constructs the service implementation.

func (*Server) Enroll

Enroll implements EnrollmentService. The token is the credential; no identity is read from the connection.

func (*Server) RenewCredential

RenewCredential implements AgentService. The workload being renewed is whatever the interceptor derived from the connection credential, which may have expired within the renewal grace: renewal is the one call such a credential may make (ADR-0016 as amended). A late renewal is logged at WARN, naming the workload and how late it was, and recorded on the workload.

func (*Server) ReportFlows

func (s *Server) ReportFlows(stream innerwallv1.AgentService_ReportFlowsServer) error

ReportFlows implements the flow telemetry stream (ADR-0015): one request per aggregation window, ingested as it arrives, and the cumulative accepted count returned when the agent closes the stream. The reporting workload is the identity the interceptor derived from the connection credential; the payload names nobody.

func (*Server) Run

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

Run listens for render announcements and directives and routes each to the stream of the workload it names, until ctx ends. It must run for pushes and directives to happen; without it, agents receive policy only at connect. Whenever the listener (re)connects, every live stream is reconciled against persisted state. A directive naming a workload whose stream this replica does not hold is dropped here, with a warning naming the workload: another replica holds it, or none does, and no replica can tell which.

func (*Server) Sync

Sync implements the desired-state stream (ADR-0002, ADR-0015): Hello, then HelloAck, then an unconditional snapshot, then deltas pushed as renders announce them, with acknowledgements, inventory, and heartbeats flowing up on the same stream.

Jump to

Keyboard shortcuts

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