feat(agentapi): gRPC agent service with mTLS and certificate checks

TLS 1.3 listener that requests client certificates and checks them in
interceptors: Enroll needs a valid one-time token (rate limited per
source IP), RenewCertificate accepts certificates expired up to 30 days,
every other call needs the agent's current, unrevoked certificate.
Connect registers the stream in the hub (pushes, revocation, shutdown);
FetchBundle signs the role's artifacts per agent; lease batches are
deduplicated by sequence.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

server/internal/agentapi/rpcs.go

@@ -0,0 +1,330 @@
+// SPDX-License-Identifier: LicenseRef-DDI-Proprietary
+
+package agentapi
+
+import (
+	"context"
+	"errors"
+	"io"
+	"log/slog"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/types/known/durationpb"
+
+	"git.beugsiud.de/beug.k/DDI/internal/bundle"
+	agentv1 "git.beugsiud.de/beug.k/DDI/proto/gen/ddi/agent/v1"
+	"git.beugsiud.de/beug.k/DDI/server/internal/agenthub"
+	"git.beugsiud.de/beug.k/DDI/server/internal/service"
+)
+
+// maxHealthChecks bounds the checks stored per report.
+const maxHealthChecks = 100
+
+func roleName(r agentv1.Role) string {
+	switch r {
+	case agentv1.Role_ROLE_DNS:
+		return service.RoleDNS
+	case agentv1.Role_ROLE_DHCP:
+		return service.RoleDHCP
+	}
+	return ""
+}
+
+func modeProto(m string) agentv1.OperatingMode {
+	if m == agenthub.ModeManual {
+		return agentv1.OperatingMode_OPERATING_MODE_MANUAL
+	}
+	return agentv1.OperatingMode_OPERATING_MODE_AUTO_PULL
+}
+
+func healthStateName(h agentv1.HealthState) string {
+	switch h {
+	case agentv1.HealthState_HEALTH_STATE_OK:
+		return service.HealthOK
+	case agentv1.HealthState_HEALTH_STATE_DEGRADED:
+		return service.HealthDegraded
+	case agentv1.HealthState_HEALTH_STATE_FAILED:
+		return service.HealthFailed
+	}
+	return service.HealthUnknown
+}
+
+func healthFromProto(h *agentv1.HealthStatus) *service.AgentHealth {
+	if h == nil {
+		return nil
+	}
+	out := &service.AgentHealth{State: healthStateName(h.GetState()), Message: h.GetMessage(), Checks: []service.HealthCheck{}}
+	if h.GetCheckedAt() != nil {
+		t := h.GetCheckedAt().AsTime()
+		out.CheckedAt = &t
+	}
+	for i, c := range h.GetChecks() {
+		if i == maxHealthChecks {
+			break
+		}
+		out.Checks = append(out.Checks, service.HealthCheck{Name: c.GetName(), State: healthStateName(c.GetState()), Message: c.GetMessage()})
+	}
+	return out
+}
+
+// Enroll implements agentv1.AgentServiceServer.
+func (s *Server) Enroll(ctx context.Context, req *agentv1.EnrollRequest) (*agentv1.EnrollResponse, error) {
+	c := callerFrom(ctx)
+	role := roleName(req.GetRole())
+	if role == "" {
+		return nil, status.Error(codes.InvalidArgument, "unknown role")
+	}
+	res, err := s.svc.EnrollAgent(ctx, service.EnrollInput{
+		Token: req.GetToken(), CSR: req.GetCsrDer(), Role: role, Hostname: req.GetHostname(),
+		AgentVersion: req.GetAgentVersion(), RemoteIP: c.RemoteIP, RequestID: c.RequestID,
+	})
+	if err != nil {
+		return nil, err
+	}
+	c.AgentID = res.AgentID
+	s.log.Info("agent enrolled", slog.String("request_id", c.RequestID), slog.String("agent_id", res.AgentID),
+		slog.String("role", role), slog.Any("remote", c.RemoteIP))
+	return &agentv1.EnrollResponse{
+		AgentId:                res.AgentID,
+		CertificateDer:         res.Certificate.Raw,
+		CaCertificatesDer:      [][]byte{s.pki.CACertificate().Raw},
+		BundleSigningPublicKey: s.pki.BundlePublicKey(),
+	}, nil
+}
+
+// RenewCertificate implements agentv1.AgentServiceServer.
+func (s *Server) RenewCertificate(ctx context.Context, req *agentv1.RenewCertificateRequest) (*agentv1.RenewCertificateResponse, error) {
+	c := callerFrom(ctx)
+	cert, err := s.svc.RenewAgentCertificate(ctx, c.AgentID, req.GetCsrDer())
+	if err != nil {
+		return nil, err
+	}
+	s.log.Info("agent certificate renewed", slog.String("request_id", c.RequestID), slog.String("agent_id", c.AgentID),
+		slog.Time("not_after", cert.NotAfter))
+	return &agentv1.RenewCertificateResponse{
+		CertificateDer: cert.Raw, CaCertificatesDer: [][]byte{s.pki.CACertificate().Raw},
+	}, nil
+}
+
+// GetDesiredState implements agentv1.AgentServiceServer.
+func (s *Server) GetDesiredState(ctx context.Context, _ *agentv1.GetDesiredStateRequest) (*agentv1.GetDesiredStateResponse, error) {
+	c := callerFrom(ctx)
+	st, err := s.svc.AgentDesiredState(ctx, c.AgentID, c.RemoteIP)
+	if err != nil {
+		return nil, err
+	}
+	return &agentv1.GetDesiredStateResponse{
+		DesiredRevision: st.DesiredRevision, Mode: modeProto(st.Mode), PullInterval: durationpb.New(service.PullInterval),
+	}, nil
+}
+
+// FetchBundle implements agentv1.AgentServiceServer. The bundle is signed
+// for the calling agent and its role; equal input gives equal bytes.
+func (s *Server) FetchBundle(ctx context.Context, req *agentv1.FetchBundleRequest) (*agentv1.FetchBundleResponse, error) {
+	c := callerFrom(ctx)
+	createdAt, arts, err := s.svc.RevisionArtifacts(ctx, req.GetRevision(), c.Role)
+	if errors.Is(err, service.ErrNotFound) {
+		return nil, status.Errorf(codes.NotFound, "revision %d does not exist", req.GetRevision())
+	}
+	if err != nil {
+		return nil, err
+	}
+	manifest, sig, err := bundle.Sign(s.pki.BundleSigningKey(), c.AgentID, c.Role, req.GetRevision(), createdAt, arts)
+	if err != nil {
+		return nil, err
+	}
+	out := &agentv1.FetchBundleResponse{Manifest: manifest, Signature: sig, Artifacts: make([]*agentv1.Artifact, len(arts))}
+	for i, a := range arts {
+		out.Artifacts[i] = &agentv1.Artifact{Path: a.Path, Content: a.Content}
+	}
+	return out, nil
+}
+
+var (
+	triggerNames = map[agentv1.DeploymentTrigger]string{
+		agentv1.DeploymentTrigger_DEPLOYMENT_TRIGGER_PULL: "pull",
+		agentv1.DeploymentTrigger_DEPLOYMENT_TRIGGER_PUSH: "push",
+	}
+	outcomeNames = map[agentv1.DeploymentOutcome]string{
+		agentv1.DeploymentOutcome_DEPLOYMENT_OUTCOME_SUCCEEDED:   "succeeded",
+		agentv1.DeploymentOutcome_DEPLOYMENT_OUTCOME_REJECTED:    "rejected",
+		agentv1.DeploymentOutcome_DEPLOYMENT_OUTCOME_ROLLED_BACK: "rolled_back",
+		agentv1.DeploymentOutcome_DEPLOYMENT_OUTCOME_FAILED:      "failed",
+	}
+	leaseStateNames = map[agentv1.LeaseState]string{
+		agentv1.LeaseState_LEASE_STATE_ACTIVE:   "active",
+		agentv1.LeaseState_LEASE_STATE_EXPIRED:  "expired",
+		agentv1.LeaseState_LEASE_STATE_RELEASED: "released",
+		agentv1.LeaseState_LEASE_STATE_DECLINED: "declined",
+	}
+)
+
+// ReportDeployment implements agentv1.AgentServiceServer.
+func (s *Server) ReportDeployment(ctx context.Context, req *agentv1.ReportDeploymentRequest) (*agentv1.ReportDeploymentResponse, error) {
+	c := callerFrom(ctx)
+	trigger, ok := triggerNames[req.GetTrigger()]
+	if !ok {
+		return nil, status.Error(codes.InvalidArgument, "unknown trigger")
+	}
+	outcome, ok := outcomeNames[req.GetOutcome()]
+	if !ok {
+		return nil, status.Error(codes.InvalidArgument, "unknown outcome")
+	}
+	rep := service.DeploymentReport{
+		CorrelationID: req.GetDeploymentId(), Revision: req.GetRevision(), Trigger: trigger, Outcome: outcome,
+		Message: req.GetMessage(), ActiveRevision: req.GetActiveRevision(),
+	}
+	if req.GetStartedAt() != nil {
+		rep.StartedAt = req.GetStartedAt().AsTime()
+	}
+	if req.GetFinishedAt() != nil {
+		rep.FinishedAt = req.GetFinishedAt().AsTime()
+	}
+	if err := s.svc.RecordDeployment(ctx, c.AgentID, rep); err != nil {
+		return nil, err
+	}
+	s.log.Info("deployment reported", slog.String("request_id", c.RequestID), slog.String("agent_id", c.AgentID),
+		slog.Uint64("revision", req.GetRevision()), slog.String("outcome", outcome),
+		slog.Uint64("active_revision", req.GetActiveRevision()), slog.String("deployment_id", req.GetDeploymentId()))
+	return &agentv1.ReportDeploymentResponse{}, nil
+}
+
+// ReportLeases implements agentv1.AgentServiceServer.
+func (s *Server) ReportLeases(ctx context.Context, req *agentv1.ReportLeasesRequest) (*agentv1.ReportLeasesResponse, error) {
+	c := callerFrom(ctx)
+	in := make([]service.LeaseInput, len(req.GetLeases()))
+	for i, l := range req.GetLeases() {
+		in[i] = service.LeaseInput{IP: l.GetIp(), MAC: l.GetMac(), Hostname: l.GetHostname(),
+			State: leaseStateNames[l.GetState()], SubnetID: l.GetSubnetId(), ClientID: l.GetClientId()}
+		if l.GetExpiresAt() != nil {
+			t := l.GetExpiresAt().AsTime()
+			in[i].ExpiresAt = &t
+		}
+		if l.GetUpdatedAt() != nil {
+			t := l.GetUpdatedAt().AsTime()
+			in[i].UpdatedAt = &t
+		}
+	}
+	res, err := s.svc.StoreLeaseBatch(ctx, c.AgentID, req.GetSequence(), req.GetFullSync(), in)
+	if err != nil {
+		return nil, err
+	}
+	attrs := []any{slog.String("request_id", c.RequestID), slog.String("agent_id", c.AgentID),
+		slog.Uint64("sequence", req.GetSequence()), slog.Bool("full_sync", req.GetFullSync()), slog.Int("leases", len(in))}
+	switch {
+	case res.Duplicate:
+		s.log.Info("dropped duplicate lease batch", append(attrs, slog.Uint64("acknowledged", res.Acknowledged))...)
+	case res.Skipped > 0:
+		s.log.Warn("skipped invalid leases", append(attrs, slog.Int("skipped", res.Skipped))...)
+	default:
+		s.log.Debug("stored lease batch", attrs...)
+	}
+	return &agentv1.ReportLeasesResponse{AcknowledgedSequence: res.Acknowledged}, nil
+}
+
+// Connect implements agentv1.AgentServiceServer: Hello → Welcome, then
+// heartbeats and acknowledgements from the agent and commands from the
+// server until either side ends the stream.
+func (s *Server) Connect(stream agentv1.AgentService_ConnectServer) error {
+	ctx := stream.Context()
+	c := callerFrom(ctx)
+	first, err := stream.Recv()
+	if err != nil {
+		return err
+	}
+	hello := first.GetHello()
+	if hello == nil {
+		return status.Error(codes.InvalidArgument, "the first message must be Hello")
+	}
+	// Register before reading the desired revision, so that a revision
+	// created in between is pushed rather than missed.
+	conn := s.hub.Attach(c.AgentID, c.Role, c.Mode)
+	defer s.hub.Detach(conn)
+	settings, err := s.svc.AgentHello(ctx, c.AgentID, service.HelloInput{
+		Hostname: hello.GetHostname(), AgentVersion: hello.GetAgentVersion(), ServiceVersion: hello.GetServiceVersion(),
+		AppliedRevision: hello.GetAppliedRevision(), LastGoodRevision: hello.GetLastGoodRevision(),
+		Health: healthFromProto(hello.GetHealth()), RemoteIP: c.RemoteIP,
+	})
+	if err != nil {
+		return err
+	}
+	if err := stream.Send(&agentv1.ConnectResponse{Message: &agentv1.ConnectResponse_Welcome{Welcome: &agentv1.Welcome{
+		Mode: modeProto(settings.Mode), PullInterval: durationpb.New(service.PullInterval),
+		HeartbeatInterval: durationpb.New(service.HeartbeatInterval), DesiredRevision: settings.DesiredRevision,
+	}}}); err != nil {
+		return err
+	}
+	s.log.Info("agent connected", slog.String("request_id", c.RequestID), slog.String("agent_id", c.AgentID),
+		slog.String("role", c.Role), slog.Uint64("applied_revision", hello.GetAppliedRevision()),
+		slog.Uint64("desired_revision", settings.DesiredRevision))
+
+	recvErr := make(chan error, 1)
+	go func() {
+		for {
+			msg, err := stream.Recv()
+			if err != nil {
+				recvErr <- err
+				return
+			}
+			s.handleMessage(ctx, c, msg)
+		}
+	}()
+	for {
+		select {
+		case cmd := <-conn.Commands():
+			if err := stream.Send(&agentv1.ConnectResponse{Message: &agentv1.ConnectResponse_Command{Command: cmd}}); err != nil {
+				return err
+			}
+			s.log.Info("command sent", slog.String("request_id", c.RequestID), slog.String("agent_id", c.AgentID),
+				slog.String("command_id", cmd.GetCommandId()), slog.Uint64("revision", cmd.GetApplyRevision().GetRevision()),
+				slog.String("deployment_id", cmd.GetApplyRevision().GetDeploymentId()))
+		case <-conn.Done():
+			reason := conn.Reason()
+			s.log.Info("agent disconnected by server", slog.String("request_id", c.RequestID),
+				slog.String("agent_id", c.AgentID), slog.String("reason", reason))
+			switch reason {
+			case agenthub.ReasonRevoked, agenthub.ReasonDeleted:
+				return status.Error(codes.PermissionDenied, reason)
+			case agenthub.ReasonReplaced:
+				return status.Error(codes.Aborted, reason)
+			default:
+				return status.Error(codes.Unavailable, reason)
+			}
+		case err := <-recvErr:
+			if errors.Is(err, io.EOF) {
+				s.log.Info("agent closed its stream", slog.String("request_id", c.RequestID), slog.String("agent_id", c.AgentID))
+				return nil
+			}
+			return err
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+}
+
+func (s *Server) handleMessage(ctx context.Context, c *caller, msg *agentv1.ConnectRequest) {
+	switch m := msg.GetMessage().(type) {
+	case *agentv1.ConnectRequest_Heartbeat:
+		hb := m.Heartbeat
+		err := s.svc.AgentHeartbeat(ctx, c.AgentID, service.HeartbeatInput{
+			AppliedRevision: hb.GetAppliedRevision(), LastGoodRevision: hb.GetLastGoodRevision(),
+			Health: healthFromProto(hb.GetHealth()),
+		})
+		if errors.Is(err, service.ErrAgentRevoked) {
+			s.hub.Disconnect(c.AgentID, agenthub.ReasonRevoked)
+		} else if err != nil && ctx.Err() == nil {
+			s.log.Warn("heartbeat not stored", slog.String("request_id", c.RequestID), slog.String("agent_id", c.AgentID),
+				slog.Any("error", err))
+		}
+	case *agentv1.ConnectRequest_CommandAck:
+		ack := m.CommandAck
+		s.log.Info("command acknowledged", slog.String("request_id", c.RequestID), slog.String("agent_id", c.AgentID),
+			slog.String("command_id", ack.GetCommandId()), slog.Bool("accepted", ack.GetAccepted()),
+			slog.String("message", ack.GetMessage()))
+	case *agentv1.ConnectRequest_Hello:
+		s.log.Warn("ignoring repeated Hello", slog.String("request_id", c.RequestID), slog.String("agent_id", c.AgentID))
+	default:
+		s.log.Warn("ignoring unknown stream message", slog.String("request_id", c.RequestID), slog.String("agent_id", c.AgentID))
+	}
+}

server/internal/agentapi/server.go

@@ -0,0 +1,292 @@
+// SPDX-License-Identifier: LicenseRef-DDI-Proprietary
+
+// Package agentapi is the gRPC AgentService of the management server
+// (proto/ddi/agent/v1, ADR 0007, 0008, 0021). It listens on its own TLS 1.3
+// port, requests client certificates and checks them itself: Enroll needs
+// none, RenewCertificate accepts a recently expired one, every other call
+// needs the agent's current, unrevoked certificate.
+package agentapi
+
+import (
+	"context"
+	"crypto/rand"
+	"crypto/tls"
+	"encoding/hex"
+	"errors"
+	"log/slog"
+	"net"
+	"net/netip"
+	"regexp"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/keepalive"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/peer"
+	"google.golang.org/grpc/status"
+
+	agentv1 "git.beugsiud.de/beug.k/DDI/proto/gen/ddi/agent/v1"
+	"git.beugsiud.de/beug.k/DDI/server/internal/agenthub"
+	"git.beugsiud.de/beug.k/DDI/server/internal/pki"
+	"git.beugsiud.de/beug.k/DDI/server/internal/service"
+)
+
+// MaxRecvMsgSize bounds requests (large full lease syncs).
+const MaxRecvMsgSize = 32 << 20
+
+// Options configures the server.
+type Options struct {
+	Service *service.Service
+	PKI     *pki.Authority
+	Hub     *agenthub.Hub
+	// ServerNames become the SANs of the server certificate.
+	ServerNames []string
+	Logger      *slog.Logger
+}
+
+// Server implements agentv1.AgentServiceServer.
+type Server struct {
+	agentv1.UnimplementedAgentServiceServer
+
+	svc  *service.Service
+	pki  *pki.Authority
+	hub  *agenthub.Hub
+	log  *slog.Logger
+	grpc *grpc.Server
+}
+
+// New creates the server and issues its certificate.
+func New(opts Options) (*Server, error) {
+	if opts.Service == nil || opts.PKI == nil || opts.Hub == nil {
+		return nil, errors.New("agentapi: service, PKI and hub are required")
+	}
+	if opts.Logger == nil {
+		opts.Logger = slog.New(slog.DiscardHandler)
+	}
+	cert, err := opts.PKI.ServerCertificate(opts.ServerNames)
+	if err != nil {
+		return nil, err
+	}
+	s := &Server{svc: opts.Service, pki: opts.PKI, hub: opts.Hub, log: opts.Logger}
+	tlsCfg := &tls.Config{
+		MinVersion:   tls.VersionTLS13,
+		Certificates: []tls.Certificate{cert},
+		// Checked by the interceptors (Enroll has no certificate,
+		// RenewCertificate may present an expired one).
+		ClientAuth: tls.RequestClientCert,
+	}
+	s.grpc = grpc.NewServer(
+		grpc.Creds(credentials.NewTLS(tlsCfg)),
+		grpc.ChainUnaryInterceptor(s.unaryInterceptor),
+		grpc.ChainStreamInterceptor(s.streamInterceptor),
+		// Agents ping every 30 s; allow some slack.
+		grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{MinTime: 15 * time.Second, PermitWithoutStream: true}),
+		grpc.KeepaliveParams(keepalive.ServerParameters{Time: time.Minute, Timeout: 20 * time.Second}),
+		grpc.MaxRecvMsgSize(MaxRecvMsgSize),
+	)
+	agentv1.RegisterAgentServiceServer(s.grpc, s)
+	return s, nil
+}
+
+// Serve accepts connections on lis until Stop or Shutdown.
+func (s *Server) Serve(lis net.Listener) error {
+	err := s.grpc.Serve(lis)
+	if errors.Is(err, grpc.ErrServerStopped) {
+		return nil
+	}
+	return err
+}
+
+// Shutdown closes the agents' streams, waits for running calls and stops
+// hard when ctx ends.
+func (s *Server) Shutdown(ctx context.Context) {
+	s.hub.CloseAll(agenthub.ReasonShutdown)
+	done := make(chan struct{})
+	go func() {
+		s.grpc.GracefulStop()
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-ctx.Done():
+		s.grpc.Stop()
+		<-done
+	}
+}
+
+// caller is the authenticated agent of a call.
+type caller struct {
+	AgentID   string
+	Role      string
+	Mode      string
+	RequestID string
+	RemoteIP  *netip.Addr
+}
+
+type callerKey struct{}
+
+func callerFrom(ctx context.Context) *caller {
+	c, _ := ctx.Value(callerKey{}).(*caller)
+	if c == nil {
+		return &caller{}
+	}
+	return c
+}
+
+var requestIDPattern = regexp.MustCompile(`^[A-Za-z0-9._-]{1,64}$`)
+
+// requestID takes the agent's x-request-id if it is well-formed, so that
+// agent and server logs correlate; otherwise it makes one up.
+func requestID(ctx context.Context) string {
+	if md, ok := metadata.FromIncomingContext(ctx); ok {
+		if v := md.Get("x-request-id"); len(v) == 1 && requestIDPattern.MatchString(v[0]) {
+			return v[0]
+		}
+	}
+	var b [8]byte
+	_, _ = rand.Read(b[:])
+	return hex.EncodeToString(b[:])
+}
+
+func peerIP(ctx context.Context) *netip.Addr {
+	p, ok := peer.FromContext(ctx)
+	if !ok || p.Addr == nil {
+		return nil
+	}
+	ap, err := netip.ParseAddrPort(p.Addr.String())
+	if err != nil {
+		return nil
+	}
+	ip := ap.Addr().Unmap()
+	return &ip
+}
+
+// authenticate checks the client certificate for method.
+func (s *Server) authenticate(ctx context.Context, method string) (*caller, error) {
+	c := &caller{RequestID: requestID(ctx), RemoteIP: peerIP(ctx)}
+	if method == agentv1.AgentService_Enroll_FullMethodName {
+		return c, nil
+	}
+	p, ok := peer.FromContext(ctx)
+	if !ok {
+		return c, status.Error(codes.Unauthenticated, "client certificate required")
+	}
+	info, ok := p.AuthInfo.(credentials.TLSInfo)
+	if !ok || len(info.State.PeerCertificates) == 0 {
+		return c, status.Error(codes.Unauthenticated, "client certificate required")
+	}
+	cert := info.State.PeerCertificates[0]
+	mode := pki.VerifyCurrent
+	if method == agentv1.AgentService_RenewCertificate_FullMethodName {
+		mode = pki.VerifyRenewal
+	}
+	id, err := s.pki.VerifyAgentCertificate(cert, mode)
+	if err != nil {
+		s.log.Warn("agent certificate rejected", slog.String("request_id", c.RequestID), slog.String("method", method),
+			slog.Any("remote", c.RemoteIP), slog.Any("error", err))
+		return c, status.Error(codes.Unauthenticated, "invalid client certificate")
+	}
+	c.AgentID = id
+	ident, err := s.svc.AgentIdentity(ctx, id)
+	if errors.Is(err, service.ErrNotFound) {
+		return c, status.Error(codes.Unauthenticated, "unknown agent")
+	}
+	if err != nil {
+		return c, s.toStatus(c, err)
+	}
+	if ident.Revoked {
+		return c, status.Error(codes.PermissionDenied, "agent is revoked")
+	}
+	if pki.SerialHex(cert.SerialNumber) != ident.CertSerial {
+		return c, status.Error(codes.Unauthenticated, "certificate has been replaced")
+	}
+	c.Role, c.Mode = ident.Role, ident.Mode
+	return c, nil
+}
+
+func (s *Server) logCall(c *caller, method string, start time.Time, err error) {
+	code := status.Code(err)
+	attrs := []any{slog.String("request_id", c.RequestID), slog.String("method", method),
+		slog.String("code", code.String()), slog.Duration("duration", time.Since(start))}
+	if c.AgentID != "" {
+		attrs = append(attrs, slog.String("agent_id", c.AgentID))
+	}
+	switch code {
+	case codes.OK, codes.Canceled:
+		s.log.Debug("agent call", attrs...)
+	case codes.Internal, codes.Unknown:
+		s.log.Error("agent call failed", append(attrs, slog.Any("error", err))...)
+	default:
+		s.log.Info("agent call rejected", append(attrs, slog.String("reason", status.Convert(err).Message()))...)
+	}
+}
+
+func (s *Server) unaryInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
+	start := time.Now()
+	c, err := s.authenticate(ctx, info.FullMethod)
+	if err == nil {
+		var resp any
+		resp, err = handler(context.WithValue(ctx, callerKey{}, c), req)
+		if err != nil {
+			err = s.toStatus(c, err)
+		}
+		s.logCall(c, info.FullMethod, start, err)
+		return resp, err
+	}
+	s.logCall(c, info.FullMethod, start, err)
+	return nil, err
+}
+
+type wrappedStream struct {
+	grpc.ServerStream
+	ctx context.Context
+}
+
+func (w *wrappedStream) Context() context.Context { return w.ctx }
+
+func (s *Server) streamInterceptor(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+	start := time.Now()
+	c, err := s.authenticate(ss.Context(), info.FullMethod)
+	if err == nil {
+		err = handler(srv, &wrappedStream{ServerStream: ss, ctx: context.WithValue(ss.Context(), callerKey{}, c)})
+		if err != nil {
+			err = s.toStatus(c, err)
+		}
+	}
+	s.logCall(c, info.FullMethod, start, err)
+	return err
+}
+
+var kindCodes = map[service.Kind]codes.Code{
+	service.KindInvalid:      codes.InvalidArgument,
+	service.KindNotFound:     codes.NotFound,
+	service.KindConflict:     codes.FailedPrecondition,
+	service.KindForbidden:    codes.PermissionDenied,
+	service.KindUnauthorized: codes.Unauthenticated,
+	service.KindRateLimited:  codes.ResourceExhausted,
+	service.KindUnavailable:  codes.Unavailable,
+}
+
+// toStatus converts service errors; internal details are logged, not sent.
+func (s *Server) toStatus(c *caller, err error) error {
+	if err == nil {
+		return nil
+	}
+	if _, ok := status.FromError(err); ok {
+		return err
+	}
+	var se *service.Error
+	if errors.As(err, &se) && se.Kind != 0 {
+		return status.Error(kindCodes[se.Kind], se.Message)
+	}
+	switch {
+	case errors.Is(err, context.Canceled):
+		return status.Error(codes.Canceled, "canceled")
+	case errors.Is(err, context.DeadlineExceeded):
+		return status.Error(codes.DeadlineExceeded, "deadline exceeded")
+	}
+	s.log.Error("agent call failed", slog.String("request_id", c.RequestID), slog.String("agent_id", c.AgentID), slog.Any("error", err))
+	return status.Error(codes.Internal, "internal error")
+}

server/internal/agentapi/server_test.go

@@ -0,0 +1,610 @@
+// SPDX-License-Identifier: LicenseRef-DDI-Proprietary
+
+package agentapi
+
+import (
+	"bytes"
+	"context"
+	"crypto/ecdsa"
+	"crypto/ed25519"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
+	"io"
+	"log/slog"
+	"net"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/types/known/timestamppb"
+
+	"git.beugsiud.de/beug.k/DDI/internal/agentpki"
+	"git.beugsiud.de/beug.k/DDI/internal/bundle"
+	agentv1 "git.beugsiud.de/beug.k/DDI/proto/gen/ddi/agent/v1"
+	"git.beugsiud.de/beug.k/DDI/server/internal/agenthub"
+	"git.beugsiud.de/beug.k/DDI/server/internal/pki"
+	"git.beugsiud.de/beug.k/DDI/server/internal/secrets"
+	"git.beugsiud.de/beug.k/DDI/server/internal/service"
+	"git.beugsiud.de/beug.k/DDI/server/internal/store"
+	"git.beugsiud.de/beug.k/DDI/server/internal/store/storetest"
+)
+
+const serverName = "agents.test"
+
+type testEnv struct {
+	t   *testing.T
+	st  *store.Store
+	svc *service.Service
+	pki *pki.Authority
+	hub *agenthub.Hub
+	srv *Server
+	// offset shifts the PKI clock (certificate issuance and checks).
+	offset atomic.Int64
+	addr   string
+}
+
+func newTestEnv(t *testing.T) *testEnv {
+	t.Helper()
+	e := &testEnv{t: t, st: storetest.New(t)}
+	keyData, _ := secrets.GenerateKeyFile("k1")
+	kr, err := secrets.ParseKeyFile(keyData)
+	if err != nil {
+		t.Fatal(err)
+	}
+	clock := func() time.Time { return time.Now().Add(time.Duration(e.offset.Load())) }
+	// The CA must predate the backdated agent certificates of the renewal
+	// tests (chains are checked at the time a certificate was valid).
+	e.offset.Store(int64(-90 * 24 * time.Hour))
+	e.pki, err = pki.LoadOrCreate(context.Background(), e.st, kr, nil, clock)
+	e.offset.Store(0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	e.hub = agenthub.New()
+	log := slog.New(slog.DiscardHandler)
+	e.svc = service.New(service.Options{Store: e.st, Keyring: kr, PKI: e.pki, Hub: e.hub, Logger: log,
+		AgentAPIAddress: serverName + ":8444", AgentServerNames: []string{serverName}})
+	e.srv, err = New(Options{Service: e.svc, PKI: e.pki, Hub: e.hub, ServerNames: []string{serverName}, Logger: log})
+	if err != nil {
+		t.Fatal(err)
+	}
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	e.addr = lis.Addr().String()
+	done := make(chan error, 1)
+	go func() { done <- e.srv.Serve(lis) }()
+	t.Cleanup(func() {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		e.srv.Shutdown(ctx)
+		if err := <-done; err != nil {
+			t.Errorf("serve: %v", err)
+		}
+	})
+	return e
+}
+
+func (e *testEnv) dial(cfg *tls.Config) agentv1.AgentServiceClient {
+	e.t.Helper()
+	cc, err := grpc.NewClient("passthrough:///"+e.addr, grpc.WithTransportCredentials(credentials.NewTLS(cfg)))
+	if err != nil {
+		e.t.Fatal(err)
+	}
+	e.t.Cleanup(func() { _ = cc.Close() })
+	return agentv1.NewAgentServiceClient(cc)
+}
+
+// bootstrapClient trusts the server only through the CA fingerprint, like
+// the agent's enroll command.
+func (e *testEnv) bootstrapClient(fingerprint string) agentv1.AgentServiceClient {
+	return e.dial(&tls.Config{
+		MinVersion: tls.VersionTLS13, ServerName: serverName,
+		InsecureSkipVerify: true, //nolint:gosec // verified by VerifyPeerCertificate
+		VerifyPeerCertificate: func(raw [][]byte, _ [][]*x509.Certificate) error {
+			_, err := agentpki.VerifyChainWithFingerprint(raw, fingerprint, serverName)
+			return err
+		},
+	})
+}
+
+func (e *testEnv) token(role string) string {
+	e.t.Helper()
+	tk, err := e.svc.CreateEnrollmentToken(context.Background(), service.SystemActor, service.TokenInput{Role: role})
+	if err != nil {
+		e.t.Fatal(err)
+	}
+	return tk.Secret
+}
+
+func csr(t *testing.T) (*ecdsa.PrivateKey, []byte) {
+	t.Helper()
+	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	if err != nil {
+		t.Fatal(err)
+	}
+	der, err := x509.CreateCertificateRequest(rand.Reader, &x509.CertificateRequest{}, key)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return key, der
+}
+
+type testAgent struct {
+	id        string
+	role      string
+	ca        *x509.Certificate
+	cert      *x509.Certificate
+	bundlePub ed25519.PublicKey
+	client    agentv1.AgentServiceClient
+}
+
+func ctxT(t *testing.T) context.Context {
+	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+	t.Cleanup(cancel)
+	return ctx
+}
+
+func roleProto(role string) agentv1.Role {
+	if role == "dhcp" {
+		return agentv1.Role_ROLE_DHCP
+	}
+	return agentv1.Role_ROLE_DNS
+}
+
+func (e *testEnv) enroll(role string) *testAgent {
+	e.t.Helper()
+	key, req := csr(e.t)
+	resp, err := e.bootstrapClient(e.pki.Fingerprint()).Enroll(ctxT(e.t), &agentv1.EnrollRequest{
+		Token: e.token(role), CsrDer: req, Role: roleProto(role), Hostname: "host-" + role, AgentVersion: "1.0.0",
+	})
+	if err != nil {
+		e.t.Fatalf("enroll: %v", err)
+	}
+	return e.agentFrom(resp.GetAgentId(), role, key, resp.GetCertificateDer(), resp.GetCaCertificatesDer(), resp.GetBundleSigningPublicKey())
+}
+
+func (e *testEnv) agentFrom(id, role string, key *ecdsa.PrivateKey, certDER []byte, chain [][]byte, pub []byte) *testAgent {
+	e.t.Helper()
+	cert, err := x509.ParseCertificate(certDER)
+	if err != nil {
+		e.t.Fatal(err)
+	}
+	if len(chain) != 1 {
+		e.t.Fatalf("CA chain length %d", len(chain))
+	}
+	ca, err := x509.ParseCertificate(chain[len(chain)-1])
+	if err != nil {
+		e.t.Fatal(err)
+	}
+	client := e.dial(agentpki.ClientTLSConfig(ca, tls.Certificate{Certificate: [][]byte{certDER}, PrivateKey: key}, serverName))
+	return &testAgent{id: id, role: role, ca: ca, cert: cert, bundlePub: pub, client: client}
+}
+
+func wantCode(t *testing.T, err error, want codes.Code) {
+	t.Helper()
+	if status.Code(err) != want {
+		t.Errorf("code %v (%v), want %v", status.Code(err), err, want)
+	}
+}
+
+// eventually polls cond for up to 5 s.
+func eventually(t *testing.T, what string, cond func() bool) {
+	t.Helper()
+	deadline := time.Now().Add(5 * time.Second)
+	for !cond() {
+		if time.Now().After(deadline) {
+			t.Fatalf("timed out waiting for %s", what)
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+}
+
+func (e *testEnv) configureGenerator(ttl uint32) {
+	e.t.Helper()
+	ctx := context.Background()
+	if _, err := e.st.Pool().Exec(ctx, `INSERT INTO networks (cidr, name, domain) VALUES ('192.0.2.0/24', 'Office', 'office.test')
+		ON CONFLICT DO NOTHING`); err != nil {
+		e.t.Fatal(err)
+	}
+	g := service.DefaultGeneratorSettings()
+	g.DNS.PrimaryNS = "ns1.office.test"
+	g.DNS.Hostmaster = "hostmaster@office.test"
+	g.DNS.NameServers = []service.GeneratorNS{{Name: "ns1.office.test", Addresses: []string{"192.0.2.53"}}}
+	g.DNS.DefaultTTL = ttl
+	if _, err := e.svc.SetGeneratorSettings(ctx, service.SystemActor, g); err != nil {
+		e.t.Fatal(err)
+	}
+}
+
+func TestEnrollConnectAndBundle(t *testing.T) {
+	e := newTestEnv(t)
+	ctx := ctxT(t)
+
+	// A server whose CA does not match the fingerprint is not trusted.
+	other, _ := pki.Generate(nil)
+	_, req := csr(t)
+	if _, err := e.bootstrapClient(other.Fingerprint()).Enroll(ctx, &agentv1.EnrollRequest{Token: e.token("dns"), CsrDer: req,
+		Role: agentv1.Role_ROLE_DNS}); err == nil {
+		t.Fatal("enrollment with a wrong CA fingerprint succeeded")
+	}
+
+	a := e.enroll("dns")
+	if id, err := agentpki.AgentID(a.cert); err != nil || id != a.id {
+		t.Errorf("certificate agent ID %q %v", id, err)
+	}
+	if agentpki.Fingerprint(a.ca) != e.pki.Fingerprint() || !a.bundlePub.Equal(e.pki.BundlePublicKey()) {
+		t.Error("enrollment returned foreign trust anchors")
+	}
+	if time.Until(a.cert.NotAfter) > pki.AgentCertLifetime || len(a.cert.ExtKeyUsage) != 1 || a.cert.ExtKeyUsage[0] != x509.ExtKeyUsageClientAuth {
+		t.Errorf("client certificate %v %v", a.cert.NotAfter, a.cert.ExtKeyUsage)
+	}
+
+	st, err := a.client.GetDesiredState(ctx, &agentv1.GetDesiredStateRequest{})
+	if err != nil || st.GetDesiredRevision() != 0 || st.GetMode() != agentv1.OperatingMode_OPERATING_MODE_AUTO_PULL ||
+		st.GetPullInterval().AsDuration() != time.Minute {
+		t.Fatalf("desired state %v %v", st, err)
+	}
+
+	stream, err := a.client.Connect(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := stream.Send(&agentv1.ConnectRequest{Message: &agentv1.ConnectRequest_Hello{Hello: &agentv1.Hello{
+		AgentVersion: "1.0.1", Hostname: "ns1.office.test", ServiceVersion: "BIND 9.20.3",
+		Health: &agentv1.HealthStatus{State: agentv1.HealthState_HEALTH_STATE_OK, CheckedAt: timestamppb.Now(),
+			Checks: []*agentv1.HealthCheck{{Name: "rndc", State: agentv1.HealthState_HEALTH_STATE_OK}}},
+	}}}); err != nil {
+		t.Fatal(err)
+	}
+	msg, err := stream.Recv()
+	if err != nil {
+		t.Fatal(err)
+	}
+	w := msg.GetWelcome()
+	if w == nil || w.GetDesiredRevision() != 0 || w.GetHeartbeatInterval().AsDuration() != 30*time.Second ||
+		w.GetPullInterval().AsDuration() != time.Minute || w.GetMode() != agentv1.OperatingMode_OPERATING_MODE_AUTO_PULL {
+		t.Fatalf("welcome %v", msg)
+	}
+	eventually(t, "connected state", func() bool { return e.hub.Connected(a.id) })
+	ag, err := e.svc.GetAgent(ctx, a.id)
+	if err != nil || !ag.Connected || ag.Hostname != "ns1.office.test" || ag.ServiceVersion != "BIND 9.20.3" ||
+		ag.Health.State != "ok" || len(ag.Health.Checks) != 1 || ag.RemoteAddress != "127.0.0.1" {
+		t.Errorf("agent after hello %+v %v", ag, err)
+	}
+
+	// A new revision is pushed to the connected agent.
+	e.configureGenerator(3600)
+	rev, created, err := e.svc.CreateRevision(ctx, service.SystemActor, "first")
+	if err != nil || !created || rev.Number != 1 {
+		t.Fatalf("create revision %+v %v %v", rev, created, err)
+	}
+	msg, err = stream.Recv()
+	if err != nil {
+		t.Fatal(err)
+	}
+	cmd := msg.GetCommand()
+	if cmd.GetApplyRevision().GetRevision() != 1 || cmd.GetCommandId() == "" || cmd.GetApplyRevision().GetDeploymentId() == "" {
+		t.Fatalf("command %v", msg)
+	}
+	if err := stream.Send(&agentv1.ConnectRequest{Message: &agentv1.ConnectRequest_CommandAck{CommandAck: &agentv1.CommandAck{
+		CommandId: cmd.GetCommandId(), Accepted: true}}}); err != nil {
+		t.Fatal(err)
+	}
+
+	// The bundle verifies with the key from the enrollment.
+	b, err := a.client.FetchBundle(ctx, &agentv1.FetchBundleRequest{Revision: 1})
+	if err != nil {
+		t.Fatal(err)
+	}
+	contents := map[string][]byte{}
+	for _, art := range b.GetArtifacts() {
+		contents[art.GetPath()] = art.GetContent()
+	}
+	m, err := bundle.Verify(a.bundlePub, b.GetManifest(), b.GetSignature(), contents, bundle.Expect{AgentID: a.id, Role: "dns", MinRevision: 1})
+	if err != nil {
+		t.Fatalf("verify bundle: %v", err)
+	}
+	if m.Revision != 1 || len(m.Files) != len(rev.DNS) || !m.CreatedAt.Equal(rev.CreatedAt) {
+		t.Errorf("manifest %+v", m)
+	}
+	if _, ok := contents["zones/db.office.test"]; !ok {
+		t.Errorf("zone file missing: %v", m.Files)
+	}
+	again, err := a.client.FetchBundle(ctx, &agentv1.FetchBundleRequest{Revision: 1})
+	if err != nil || !bytes.Equal(again.GetManifest(), b.GetManifest()) || !bytes.Equal(again.GetSignature(), b.GetSignature()) {
+		t.Errorf("bundle is not deterministic: %v", err)
+	}
+	_, err = a.client.FetchBundle(ctx, &agentv1.FetchBundleRequest{Revision: 99})
+	wantCode(t, err, codes.NotFound)
+
+	// Heartbeat and deployment report.
+	if err := stream.Send(&agentv1.ConnectRequest{Message: &agentv1.ConnectRequest_Heartbeat{Heartbeat: &agentv1.Heartbeat{
+		AppliedRevision: 1, LastGoodRevision: 1,
+		Health: &agentv1.HealthStatus{State: agentv1.HealthState_HEALTH_STATE_DEGRADED, Message: "slow"}}}}); err != nil {
+		t.Fatal(err)
+	}
+	eventually(t, "heartbeat", func() bool {
+		ag, err := e.svc.GetAgent(ctx, a.id)
+		return err == nil && ag.AppliedRevision == 1 && ag.Health.State == "degraded"
+	})
+	start := time.Now().Add(-2 * time.Second)
+	if _, err := a.client.ReportDeployment(ctx, &agentv1.ReportDeploymentRequest{
+		DeploymentId: cmd.GetApplyRevision().GetDeploymentId(), Revision: 1,
+		Trigger: agentv1.DeploymentTrigger_DEPLOYMENT_TRIGGER_PUSH, Outcome: agentv1.DeploymentOutcome_DEPLOYMENT_OUTCOME_SUCCEEDED,
+		ActiveRevision: 1, StartedAt: timestamppb.New(start), FinishedAt: timestamppb.Now(),
+	}); err != nil {
+		t.Fatal(err)
+	}
+	_, err = a.client.ReportDeployment(ctx, &agentv1.ReportDeploymentRequest{Revision: 1})
+	wantCode(t, err, codes.InvalidArgument)
+	ag, _ = e.svc.GetAgent(ctx, a.id)
+	if ag.LastDeployment == nil || ag.LastDeployment.Trigger != "push" || ag.LastGoodRevision != 1 || ag.DesiredRevision != 1 {
+		t.Errorf("agent after deployment %+v", ag)
+	}
+
+	// The agent closes its stream.
+	if err := stream.CloseSend(); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := stream.Recv(); err != io.EOF { //nolint:errorlint // gRPC returns io.EOF as is
+		t.Errorf("recv after close: %v", err)
+	}
+	eventually(t, "disconnect", func() bool { return !e.hub.Connected(a.id) })
+}
+
+func TestEnrollmentRejections(t *testing.T) {
+	e := newTestEnv(t)
+	ctx := ctxT(t)
+	client := e.bootstrapClient(e.pki.Fingerprint())
+	_, req := csr(t)
+
+	tok := e.token("dns")
+	_, err := client.Enroll(ctx, &agentv1.EnrollRequest{Token: tok, CsrDer: req, Role: agentv1.Role_ROLE_DHCP})
+	wantCode(t, err, codes.PermissionDenied)
+	_, err = client.Enroll(ctx, &agentv1.EnrollRequest{Token: tok, CsrDer: req})
+	wantCode(t, err, codes.InvalidArgument)
+	_, err = client.Enroll(ctx, &agentv1.EnrollRequest{Token: "garbage", CsrDer: req, Role: agentv1.Role_ROLE_DNS})
+	wantCode(t, err, codes.Unauthenticated)
+	_, err = client.Enroll(ctx, &agentv1.EnrollRequest{Token: tok, CsrDer: []byte("no csr"), Role: agentv1.Role_ROLE_DNS})
+	wantCode(t, err, codes.InvalidArgument)
+
+	// The failed attempts did not use up the token.
+	if _, err := client.Enroll(ctx, &agentv1.EnrollRequest{Token: tok, CsrDer: req, Role: agentv1.Role_ROLE_DNS}); err != nil {
+		t.Fatalf("enroll: %v", err)
+	}
+	_, err = client.Enroll(ctx, &agentv1.EnrollRequest{Token: tok, CsrDer: req, Role: agentv1.Role_ROLE_DNS})
+	wantCode(t, err, codes.Unauthenticated)
+
+	expired := e.token("dns")
+	if _, err := e.st.Pool().Exec(ctx, `UPDATE enrollment_tokens SET created_at = now() - interval '2 hours',
+		expires_at = now() - interval '1 hour' WHERE id = $1`, expired[:16]); err != nil {
+		t.Fatal(err)
+	}
+	_, err = client.Enroll(ctx, &agentv1.EnrollRequest{Token: expired, CsrDer: req, Role: agentv1.Role_ROLE_DNS})
+	wantCode(t, err, codes.Unauthenticated)
+}
+
+func TestEnrollmentRateLimit(t *testing.T) {
+	e := newTestEnv(t)
+	ctx := ctxT(t)
+	client := e.bootstrapClient(e.pki.Fingerprint())
+	for i := range 10 {
+		_, err := client.Enroll(ctx, &agentv1.EnrollRequest{Token: "0123456789abcdef.wrong", Role: agentv1.Role_ROLE_DNS})
+		if status.Code(err) != codes.Unauthenticated {
+			t.Fatalf("attempt %d: %v", i, err)
+		}
+	}
+	_, err := client.Enroll(ctx, &agentv1.EnrollRequest{Token: e.token("dns"), Role: agentv1.Role_ROLE_DNS})
+	wantCode(t, err, codes.ResourceExhausted)
+}
+
+func TestClientCertificateRequired(t *testing.T) {
+	e := newTestEnv(t)
+	ctx := ctxT(t)
+	a := e.enroll("dns")
+
+	noCert := e.bootstrapClient(e.pki.Fingerprint())
+	_, err := noCert.GetDesiredState(ctx, &agentv1.GetDesiredStateRequest{})
+	wantCode(t, err, codes.Unauthenticated)
+	_, err = noCert.FetchBundle(ctx, &agentv1.FetchBundleRequest{Revision: 1})
+	wantCode(t, err, codes.Unauthenticated)
+	stream, err := noCert.Connect(ctx)
+	if err == nil {
+		_, err = stream.Recv()
+	}
+	wantCode(t, err, codes.Unauthenticated)
+
+	// Same agent ID, but issued by another CA.
+	other, _ := pki.Generate(nil)
+	key, req := csr(t)
+	foreign, err := other.SignAgentCSR(req, a.id)
+	if err != nil {
+		t.Fatal(err)
+	}
+	forged := e.dial(agentpki.ClientTLSConfig(a.ca, tls.Certificate{Certificate: [][]byte{foreign.Raw}, PrivateKey: key}, serverName))
+	_, err = forged.GetDesiredState(ctx, &agentv1.GetDesiredStateRequest{})
+	wantCode(t, err, codes.Unauthenticated)
+
+	// A certificate of our CA for an agent that does not exist.
+	ghost, err := e.pki.SignAgentCSR(req, "00000000-0000-4000-8000-000000000001")
+	if err != nil {
+		t.Fatal(err)
+	}
+	ghostClient := e.dial(agentpki.ClientTLSConfig(a.ca, tls.Certificate{Certificate: [][]byte{ghost.Raw}, PrivateKey: key}, serverName))
+	_, err = ghostClient.GetDesiredState(ctx, &agentv1.GetDesiredStateRequest{})
+	wantCode(t, err, codes.Unauthenticated)
+}
+
+func TestRevocationClosesStream(t *testing.T) {
+	e := newTestEnv(t)
+	ctx := ctxT(t)
+	a := e.enroll("dns")
+	stream, err := a.client.Connect(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := stream.Send(&agentv1.ConnectRequest{Message: &agentv1.ConnectRequest_Hello{Hello: &agentv1.Hello{}}}); err != nil {
+		t.Fatal(err)
+	}
+	if msg, err := stream.Recv(); err != nil || msg.GetWelcome() == nil {
+		t.Fatalf("welcome %v %v", msg, err)
+	}
+	if _, err := e.svc.RevokeAgent(ctx, service.SystemActor, a.id); err != nil {
+		t.Fatal(err)
+	}
+	_, err = stream.Recv()
+	wantCode(t, err, codes.PermissionDenied)
+	_, err = a.client.GetDesiredState(ctx, &agentv1.GetDesiredStateRequest{})
+	wantCode(t, err, codes.PermissionDenied)
+	_, req := csr(t)
+	_, err = a.client.RenewCertificate(ctx, &agentv1.RenewCertificateRequest{CsrDer: req})
+	wantCode(t, err, codes.PermissionDenied)
+}
+
+func TestFirstMessageMustBeHello(t *testing.T) {
+	e := newTestEnv(t)
+	ctx := ctxT(t)
+	a := e.enroll("dns")
+	stream, err := a.client.Connect(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := stream.Send(&agentv1.ConnectRequest{Message: &agentv1.ConnectRequest_Heartbeat{Heartbeat: &agentv1.Heartbeat{}}}); err != nil {
+		t.Fatal(err)
+	}
+	_, err = stream.Recv()
+	wantCode(t, err, codes.InvalidArgument)
+}
+
+func TestCertificateRenewal(t *testing.T) {
+	e := newTestEnv(t)
+	ctx := ctxT(t)
+	day := 24 * time.Hour
+
+	// Enrolled 15 days ago: the certificate expired yesterday.
+	e.offset.Store(int64(-15 * day))
+	old := e.enroll("dns")
+	e.offset.Store(0)
+	if !old.cert.NotAfter.Before(time.Now()) {
+		t.Fatalf("certificate not expired: %v", old.cert.NotAfter)
+	}
+	_, err := old.client.GetDesiredState(ctx, &agentv1.GetDesiredStateRequest{})
+	wantCode(t, err, codes.Unauthenticated)
+
+	key, req := csr(t)
+	resp, err := old.client.RenewCertificate(ctx, &agentv1.RenewCertificateRequest{CsrDer: req})
+	if err != nil {
+		t.Fatalf("renew expired certificate: %v", err)
+	}
+	renewed := e.agentFrom(old.id, "dns", key, resp.GetCertificateDer(), resp.GetCaCertificatesDer(), old.bundlePub)
+	if !renewed.cert.NotAfter.After(time.Now().Add(13*day)) || renewed.cert.SerialNumber.Cmp(old.cert.SerialNumber) == 0 {
+		t.Errorf("renewed certificate %v", renewed.cert.NotAfter)
+	}
+	if _, err := renewed.client.GetDesiredState(ctx, &agentv1.GetDesiredStateRequest{}); err != nil {
+		t.Errorf("renewed certificate rejected: %v", err)
+	}
+	// The old certificate is worthless now, even for renewal.
+	_, err = old.client.RenewCertificate(ctx, &agentv1.RenewCertificateRequest{CsrDer: req})
+	wantCode(t, err, codes.Unauthenticated)
+
+	// Renewal with a valid certificate replaces it as well.
+	key2, req2 := csr(t)
+	resp, err = renewed.client.RenewCertificate(ctx, &agentv1.RenewCertificateRequest{CsrDer: req2})
+	if err != nil {
+		t.Fatal(err)
+	}
+	third := e.agentFrom(old.id, "dns", key2, resp.GetCertificateDer(), resp.GetCaCertificatesDer(), old.bundlePub)
+	_, err = renewed.client.GetDesiredState(ctx, &agentv1.GetDesiredStateRequest{})
+	wantCode(t, err, codes.Unauthenticated)
+	if _, err := third.client.GetDesiredState(ctx, &agentv1.GetDesiredStateRequest{}); err != nil {
+		t.Errorf("third certificate rejected: %v", err)
+	}
+	_, err = third.client.RenewCertificate(ctx, &agentv1.RenewCertificateRequest{CsrDer: []byte("garbage")})
+	wantCode(t, err, codes.InvalidArgument)
+
+	// Expired for more than 30 days: a new enrollment is needed.
+	e.offset.Store(int64(-45 * day))
+	stale := e.enroll("dns")
+	e.offset.Store(0)
+	_, req3 := csr(t)
+	_, err = stale.client.RenewCertificate(ctx, &agentv1.RenewCertificateRequest{CsrDer: req3})
+	wantCode(t, err, codes.Unauthenticated)
+}
+
+func TestLeaseReports(t *testing.T) {
+	e := newTestEnv(t)
+	ctx := ctxT(t)
+	dhcp := e.enroll("dhcp")
+	dns := e.enroll("dns")
+
+	lease := func(ip string, state agentv1.LeaseState) *agentv1.Lease {
+		return &agentv1.Lease{Ip: ip, Mac: "00:11:22:33:44:55", Hostname: "pc", State: state,
+			ExpiresAt: timestamppb.New(time.Now().Add(time.Hour)), SubnetId: 1}
+	}
+	report := func(seq uint64, full bool, ls ...*agentv1.Lease) uint64 {
+		t.Helper()
+		resp, err := dhcp.client.ReportLeases(ctx, &agentv1.ReportLeasesRequest{Sequence: seq, FullSync: full, Leases: ls})
+		if err != nil {
+			t.Fatalf("report %d: %v", seq, err)
+		}
+		return resp.GetAcknowledgedSequence()
+	}
+	count := func() int {
+		t.Helper()
+		page, err := e.svc.ListLeases(ctx, service.LeaseFilter{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		return len(page.Items)
+	}
+
+	if ack := report(1, true, lease("192.0.2.10", agentv1.LeaseState_LEASE_STATE_ACTIVE), lease("192.0.2.11", agentv1.LeaseState_LEASE_STATE_ACTIVE)); ack != 1 || count() != 2 {
+		t.Fatalf("full sync: ack %d, %d leases", ack, count())
+	}
+	// A repeated batch is dropped.
+	if ack := report(1, true); ack != 1 || count() != 2 {
+		t.Errorf("duplicate: ack %d, %d leases", ack, count())
+	}
+	if ack := report(3, false, lease("192.0.2.12", agentv1.LeaseState_LEASE_STATE_ACTIVE)); ack != 3 || count() != 3 {
+		t.Errorf("delta: ack %d, %d leases", ack, count())
+	}
+	if ack := report(2, true); ack != 3 || count() != 3 {
+		t.Errorf("old batch: ack %d, %d leases", ack, count())
+	}
+	if ack := report(4, true, lease("192.0.2.20", agentv1.LeaseState_LEASE_STATE_EXPIRED)); ack != 4 || count() != 1 {
+		t.Errorf("second full sync: ack %d, %d leases", ack, count())
+	}
+	_, err := dhcp.client.ReportLeases(ctx, &agentv1.ReportLeasesRequest{Sequence: 0})
+	wantCode(t, err, codes.InvalidArgument)
+	_, err = dns.client.ReportLeases(ctx, &agentv1.ReportLeasesRequest{Sequence: 1, FullSync: true})
+	wantCode(t, err, codes.PermissionDenied)
+}
+
+func TestShutdownClosesStreams(t *testing.T) {
+	e := newTestEnv(t)
+	ctx := ctxT(t)
+	a := e.enroll("dns")
+	stream, err := a.client.Connect(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := stream.Send(&agentv1.ConnectRequest{Message: &agentv1.ConnectRequest_Hello{Hello: &agentv1.Hello{}}}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := stream.Recv(); err != nil {
+		t.Fatal(err)
+	}
+	sctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	e.srv.Shutdown(sctx)
+	_, err = stream.Recv()
+	wantCode(t, err, codes.Unavailable)
+}

server/internal/agenthub/hub.go

@@ -23,6 +23,15 @@ const (
 	ModeManual   = "manual"
 )
 
+// Reasons for closing a stream from the server side.
+const (
+	ReasonRevoked  = "agent revoked"
+	ReasonDeleted  = "agent deleted"
+	ReasonReplaced = "replaced by a newer connection"
+	ReasonShutdown = "server shutting down"
+	reasonEnded    = "stream ended"
+)
+
 // Hub is the registry of open streams. It is safe for concurrent use.
 type Hub struct {
 	mu    sync.Mutex
@@ -72,11 +81,22 @@ func (h *Hub) Attach(agentID, role, mode string) *Conn {
 	h.conns[agentID] = c
 	h.mu.Unlock()
 	if old != nil {
-		old.close("replaced by a newer connection")
+		old.close(ReasonReplaced)
 	}
 	return c
 }
 
+// CloseAll closes every stream (shutdown).
+func (h *Hub) CloseAll(reason string) {
+	h.mu.Lock()
+	conns := h.conns
+	h.conns = map[string]*Conn{}
+	h.mu.Unlock()
+	for _, c := range conns {
+		c.close(reason)
+	}
+}
+
 // Detach removes c if it is still the registered stream of its agent.
 func (h *Hub) Detach(c *Conn) {
 	h.mu.Lock()
@@ -84,7 +104,7 @@ func (h *Hub) Detach(c *Conn) {
 		delete(h.conns, c.AgentID)
 	}
 	h.mu.Unlock()
-	c.close("stream ended")
+	c.close(reasonEnded)
 }
 
 // Connected reports whether the agent has an open stream.

server/internal/service/agents.go

@@ -19,6 +19,7 @@ import (
 	"unicode/utf8"
 
 	"git.beugsiud.de/beug.k/DDI/internal/brand"
+	"git.beugsiud.de/beug.k/DDI/server/internal/agenthub"
 	"git.beugsiud.de/beug.k/DDI/server/internal/store"
 )
 
@@ -227,7 +228,7 @@ func (s *Service) RevokeAgent(ctx context.Context, a *Actor, id string) (AgentDT
 		return AgentDTO{}, err
 	}
 	if s.opts.Hub != nil {
-		s.opts.Hub.Disconnect(id, "agent revoked")
+		s.opts.Hub.Disconnect(id, agenthub.ReasonRevoked)
 		out.Connected = false
 	}
 	return out, nil
@@ -256,7 +257,7 @@ func (s *Service) DeleteAgent(ctx context.Context, a *Actor, id string) error {
 		return err
 	}
 	if s.opts.Hub != nil {
-		s.opts.Hub.Disconnect(id, "agent deleted")
+		s.opts.Hub.Disconnect(id, agenthub.ReasonDeleted)
 	}
 	return nil
 }