grpc

package
v0.0.0-...-8d3d8c4 Latest Latest
Warning

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

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

Documentation

Overview

Package grpc provides TLS certificate management for the gRPC server.

EnsureCerts generates a self-signed CA and server certificate under <data-dir>/grpc/ on first startup. Subsequent calls are idempotent — if the files already exist they are loaded without regeneration.

The generated PKI is intentionally simple:

  • One root CA (ECDSA P-256, 10-year validity)
  • One server certificate signed by that CA (SAN: DNS:arkeep-grpc)
  • Per-agent client certificates issued on demand via IssueCertificate

Agents verify the server using the CA cert and a fixed ServerName ("arkeep-grpc"), which decouples TLS verification from the server's actual hostname or IP address. This allows the same certificate to work regardless of how the server is addressed.

Package grpc implements the gRPC server that agents connect to.

The server listens on a dedicated port (default: 9090) separate from the REST API port (8080). It implements the AgentService defined in shared/proto/agent.proto and acts as the bridge between connected agents and the rest of the server: it delegates connection lifecycle to agentmanager and persistence to AgentRepository.

TLS: when TLSCertFile and TLSKeyFile are set in Config, the gRPC listener is wrapped with TLS. In production always provide a certificate — either issued by a trusted CA (Let's Encrypt via Caddy/Nginx) or self-signed. Agents authenticate via a shared token in gRPC metadata (see authInterceptor).

Index

Constants

View Source 
const (
	// GRPCServerName is the fixed TLS ServerName used for the auto-generated
	// PKI. Agents must set tls.Config.ServerName to this value when connecting
	// with a CA cert issued by EnsureCerts.
	GRPCServerName = "arkeep-grpc"
)

Variables

This section is empty.

Functions

This section is empty.

Types

type AutoCerts 

type AutoCerts struct {
	// CACertFile is the on-disk path of the CA certificate (PEM).
	// Agents receive this file during enrollment.
	CACertFile string
	// CACertPEM is the raw PEM bytes of the CA certificate.
	// The enrollment handler returns this directly in the JSON response.
	CACertPEM []byte
	// ServerCertFile and ServerKeyFile are passed to the gRPC server's
	// tls.Config as the server identity certificate.
	ServerCertFile string
	ServerKeyFile  string
	// CAPool is a certificate pool containing only the generated CA.
	// Used by the gRPC server for RequireAndVerifyClientCert.
	CAPool *x509.CertPool
	// contains filtered or unexported fields
}

AutoCerts holds the in-memory representation of the auto-generated PKI. It is created once at startup by EnsureCerts and shared across the gRPC server and the enrollment HTTP handler.

func EnsureCerts 

func EnsureCerts(dataDir string, logger *zap.Logger) (*AutoCerts, error)

EnsureCerts loads the auto-generated PKI from <dataDir>/grpc/ if the files already exist, or generates a new CA + server certificate pair if they do not. Returns an error only if key generation or file I/O fails.

The function is safe to call at startup before the gRPC listener opens.

func (*AutoCerts) IssueCertificate 

func (ac *AutoCerts) IssueCertificate(cn string) (certPEM, keyPEM []byte, err error)

IssueCertificate signs a new client certificate with the given common name. Returns PEM-encoded certificate and private key bytes ready to be saved by the agent and loaded into a tls.Certificate.

func (*AutoCerts) TLSConfig 

func (ac *AutoCerts) TLSConfig() (*tls.Config, error)

TLSConfig returns a *tls.Config suitable for the gRPC server with mTLS enabled. It loads the server certificate from disk (so gRPC can reload it if needed) and configures the CA pool for client certificate verification.

type Config 

type Config struct {
	// SharedSecret is the shared secret agents must send in the "agent-secret"
	// metadata key to authenticate. If empty, a warning is logged and
	// authentication is disabled (development mode only — always set in production).
	SharedSecret string
	// TLSCertFile is the path to the PEM-encoded TLS certificate file.
	// Both TLSCertFile and TLSKeyFile must be set to enable TLS.
	TLSCertFile string
	// TLSKeyFile is the path to the PEM-encoded TLS private key file.
	TLSKeyFile string
	// AutoCerts holds the auto-generated PKI. When set, the gRPC server enables
	// mTLS (RequireAndVerifyClientCert) and the shared-secret token check is
	// bypassed — the client certificate is the authentication proof.
	AutoCerts *AutoCerts
	// PendingDispatch is called once when an agent opens its StreamJobs stream
	// to flush any jobs that were created while the agent was offline. Optional
	// — if nil, pending jobs are not re-dispatched on reconnect (test default).
	PendingDispatch PendingDispatcher
	// NotifService is used to send notifications when jobs complete or agents
	// go offline. Optional — if nil, notifications are silently skipped.
	NotifService notification.Service
	// Metrics is the Prometheus metrics collector. Optional — if nil, no
	// job metrics are recorded.
	Metrics *metrics.Metrics
}

Config holds the configuration for the gRPC server.

type PendingDispatcher 

type PendingDispatcher interface {
	DispatchPending(ctx context.Context, agentID uuid.UUID)
}

PendingDispatcher is implemented by the scheduler to flush pending jobs when an agent reconnects. Defined as an interface to avoid importing the scheduler package from the grpc package.

type Server 

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

Server is the gRPC server that handles agent connections. It wraps the generated UnimplementedAgentServiceServer to ensure forward compatibility when new RPCs are added to the proto.

func New 

func New(
	cfg Config,
	agentManager *agentmanager.Manager,
	agentRepo repositories.AgentRepository,
	jobRepo repositories.JobRepository,
	snapshotRepo repositories.SnapshotRepository,
	policyRepo repositories.PolicyRepository,
	destRepo repositories.DestinationRepository,
	hub *websocket.Hub,
	logger *zap.Logger,
) *Server

New creates a new Server instance with the given dependencies.

func (*Server) Heartbeat 

func (s *Server) Heartbeat(ctx context.Context, req *proto.HeartbeatRequest) (*proto.HeartbeatResponse, error)

Heartbeat handles periodic liveness signals from agents. It updates the agent's status to "online" and last_seen_at to now, then publishes the received system metrics to the WebSocket hub so the GUI can display live resource utilization on the agent detail page.

Using UpdateStatus with "online" is intentional: if an agent is sending heartbeats it is by definition online, so we can skip a read of the current status and update both fields in a single query.

func (*Server) ListenAndServe 

func (s *Server) ListenAndServe(ctx context.Context, listenAddr string) error

ListenAndServe starts the gRPC server and blocks until the context is cancelled or a fatal error occurs. It registers the AgentService and attaches the auth interceptor to all incoming RPCs.

The caller is responsible for passing a context that is cancelled on shutdown (e.g. via signal handling in cmd/server/main.go).

func (*Server) Register 

func (s *Server) Register(ctx context.Context, req *proto.RegisterRequest) (*proto.RegisterResponse, error)

Register handles the initial agent registration RPC. Upsert logic:

  • If agent sends a persisted agent_id → look up by ID, update metadata.
  • Otherwise (first-ever run, or DB wiped) → create a new record.

hostname is stored as display/operational metadata only; it is never used as an identity key.

func (*Server) ReportDestinationStatus 

func (s *Server) ReportDestinationStatus(ctx context.Context, req *proto.DestinationStatusReport) (*proto.DestinationStatusResponse, error)

ReportDestinationStatus handles per-destination result reports from agents. Called once per destination after the backup to that destination completes or fails. Persists the restic snapshot ID, byte count, and final status so the GUI can show per-destination outcomes on the job detail page.

On success, a Snapshot record is also created so the snapshot appears in the snapshots list without requiring a separate restic catalog scan.

func (*Server) ReportJobStatus 

func (s *Server) ReportJobStatus(ctx context.Context, req *proto.JobStatusReport) (*proto.JobStatusResponse, error)

ReportJobStatus handles job lifecycle updates from agents. It persists the status change to the database so the GUI can display real-time job progress.

func (*Server) ReportSnapshotBrowse 

func (s *Server) ReportSnapshotBrowse(ctx context.Context, req *proto.SnapshotBrowseReport) (*proto.SnapshotBrowseResponse, error)

ReportSnapshotBrowse receives the snapshot file listing from an agent in response to a JOB_TYPE_LIST_SNAPSHOT_FILES request. It delivers the result to the waiting RequestSnapshotBrowse call via the agent manager.

func (*Server) ReportSnapshotImport 

func (s *Server) ReportSnapshotImport(ctx context.Context, req *proto.SnapshotImportReport) (*proto.SnapshotImportResponse, error)

ReportSnapshotImport receives the snapshot list from an agent in response to a JOB_TYPE_IMPORT_SNAPSHOTS request. It delivers the result to the waiting RequestSnapshotImport call via the agent manager.

func (*Server) ReportVolumeList 

func (s *Server) ReportVolumeList(ctx context.Context, req *proto.VolumeListReport) (*proto.VolumeListResponse, error)

ReportVolumeList receives the Docker volume list from an agent in response to a JOB_TYPE_LIST_VOLUMES request sent via StreamJobs. It delivers the result to the waiting RequestVolumeList call via the agent manager.

func (*Server) Serve 

func (s *Server) Serve(ctx context.Context, lis net.Listener) error

Serve starts the gRPC server on an existing listener and blocks until the context is cancelled or a fatal error occurs.

This is the lower-level counterpart to ListenAndServe — it accepts a pre-created net.Listener so callers (e.g. integration tests) can control the bind address and retrieve the actual port before Serve is called.

func (*Server) StreamJobs 

func (s *Server) StreamJobs(req *proto.StreamJobsRequest, stream proto.AgentService_StreamJobsServer) error

StreamJobs opens the persistent job delivery stream for an agent. The agent calls this once after Register and keeps the stream open for its entire session. The method blocks until the stream closes (agent disconnects or context is cancelled), then cleans up.

func (*Server) StreamLogs 

func (s *Server) StreamLogs(stream proto.AgentService_StreamLogsServer) error

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.