diff --git a/server/internal/agentapi/rpcs.go b/server/internal/agentapi/rpcs.go new file mode 100644 index 0000000..059a5a0 --- /dev/null +++ b/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)) + } +} diff --git a/server/internal/agentapi/server.go b/server/internal/agentapi/server.go new file mode 100644 index 0000000..24c18d0 --- /dev/null +++ b/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") +} diff --git a/server/internal/agentapi/server_test.go b/server/internal/agentapi/server_test.go new file mode 100644 index 0000000..972ac02 --- /dev/null +++ b/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) +} diff --git a/server/internal/agenthub/hub.go b/server/internal/agenthub/hub.go index 5eec1f1..cc33853 100644 --- a/server/internal/agenthub/hub.go +++ b/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. diff --git a/server/internal/service/agents.go b/server/internal/service/agents.go index 8b76c4b..a3ec57b 100644 --- a/server/internal/service/agents.go +++ b/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 }