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). It also refuses an identity whose workload was retired, with one refusal that names no reason, and the gateway ends a retired workload's live stream with the same refusal (ADR-0023).
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 ¶
- func DefaultSyncConfig() *innerwallv1.SyncConfig
- func IdentityFromContext(ctx context.Context) (identity.WorkloadID, bool)
- func NewGRPCServer(tlsCfg *tls.Config, srv *Server, opts ...grpc.ServerOption) *grpc.Server
- func ServerTLSConfig(serverCert tls.Certificate, bundlePEM []byte) (*tls.Config, error)
- func WithIdentity(ctx context.Context, id identity.WorkloadID) context.Context
- type Deps
- type PolicyEvents
- type PolicyStore
- type Server
- func (s *Server) EndRetired(id identity.WorkloadID) bool
- func (s *Server) EndStream(id identity.WorkloadID, err error) bool
- func (s *Server) Enroll(ctx context.Context, req *innerwallv1.EnrollRequest) (*innerwallv1.EnrollResponse, error)
- func (s *Server) RenewCredential(ctx context.Context, req *innerwallv1.RenewCredentialRequest) (*innerwallv1.RenewCredentialResponse, error)
- func (s *Server) ReportFlows(stream innerwallv1.AgentService_ReportFlowsServer) error
- func (s *Server) Run(ctx context.Context) error
- func (s *Server) Sync(stream innerwallv1.AgentService_SyncServer) error
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 ¶
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 ¶
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 ¶
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), directed func(compiler.Directive)) 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 (*Server) EndRetired ¶ added in v0.5.0
func (s *Server) EndRetired(id identity.WorkloadID) bool
EndRetired implements fleet.Streams: it ends the live stream of a retired workload with the opaque refusal, if this replica holds it.
func (*Server) EndStream ¶ added in v0.5.0
func (s *Server) EndStream(id identity.WorkloadID, err error) bool
EndStream ends the live sync stream of id, if this replica holds one, with err as the status its agent receives, and reports whether it held one. It is the gateway's one lever for ending a stream from outside it; retirement pulls it, with the opaque refusal. A stream whose credential expires is ended by the same mechanism from inside its own session, with a refusal that names the expiry (boundByCredential).
func (*Server) Enroll ¶
func (s *Server) Enroll(ctx context.Context, req *innerwallv1.EnrollRequest) (*innerwallv1.EnrollResponse, error)
Enroll implements EnrollmentService. The token is the credential; no identity is read from the connection.
func (*Server) RenewCredential ¶
func (s *Server) RenewCredential(ctx context.Context, req *innerwallv1.RenewCredentialRequest) (*innerwallv1.RenewCredentialResponse, error)
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 ¶
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, its workload's retirement included. A reconnect 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 ¶
func (s *Server) Sync(stream innerwallv1.AgentService_SyncServer) error
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.
Source Files