diff --git a/controller/cmd/telemetry/main.go b/controller/cmd/telemetry/main.go index 08ecb5468..1c757d42e 100644 --- a/controller/cmd/telemetry/main.go +++ b/controller/cmd/telemetry/main.go @@ -18,6 +18,18 @@ limitations under the License. // via the PushLogs gRPC RPC and writes them to structured stdout for downstream // log shippers (Promtail, Grafana Alloy, Vector) to forward to Loki. // +// TLS: always enabled. Set EXTERNAL_CERT_PEM and EXTERNAL_KEY_PEM to file paths of +// operator-mounted cert/key (e.g. from a cert-manager Secret); when absent a +// self-signed certificate is generated. The self-signed cert PEM is logged at +// startup — copy it into the controller ConfigMap's telemetry.certificate field +// so exporters can verify the TLS connection. +// +// Endpoint: GRPC_TELEMETRY_ENDPOINT must be set on BOTH this pod and the controller +// pod to the same value (e.g. "jumpstarter-telemetry.jumpstarter.svc:9093"). +// The telemetry service uses it to generate the correct SAN in the self-signed +// certificate; the controller uses it to advertise the address to exporters via +// GetServiceEndpoints. A mismatch causes TLS hostname verification failures. +// // Future phases will add direct Loki push and MetricsStream for reverse-scrape // of exporter prometheus_client registries. package main @@ -79,14 +91,16 @@ func main() { Signer: signer, } + // Register signal handler before starting the service so no signal + // is missed in the window between goroutine start and Notify. + sigs := make(chan os.Signal, 1) + signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM) + errCh := make(chan error, 1) go func() { errCh <- svc.Start(ctx) }() - sigs := make(chan os.Signal, 1) - signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM) - select { case sig := <-sigs: logger.Info("received signal, shutting down", "signal", sig) diff --git a/controller/internal/config/config.go b/controller/internal/config/config.go index 6b0b2170e..e65d5b252 100644 --- a/controller/internal/config/config.go +++ b/controller/internal/config/config.go @@ -1,8 +1,11 @@ package config import ( + "cmp" "context" "fmt" + "net" + "os" "time" "github.com/jumpstarter-dev/jumpstarter/controller/internal/oidc" @@ -43,6 +46,34 @@ func LoadRouterConfiguration( return serverOptions, nil } +// resolveTelemetryConfig validates and resolves the telemetry endpoint for a +// Telemetry config block. The GRPC_TELEMETRY_ENDPOINT env var takes priority +// over the ConfigMap value, allowing operators to override at the pod level. +// Returns nil when t is nil or disabled. +func resolveTelemetryConfig(t *Telemetry) (*Telemetry, error) { + if t == nil || !t.Enabled { + return nil, nil + } + if err := t.Validate(); err != nil { + return nil, err + } + // Env var takes priority over ConfigMap, allowing operators to override + // at the pod level without modifying the ConfigMap. Resolving here ensures + // LoadedConfig.Telemetry.Endpoint is always the complete value — callers + // don't need to re-check the env var. + t.Endpoint = cmp.Or(os.Getenv("GRPC_TELEMETRY_ENDPOINT"), t.Endpoint) + if ep := t.Endpoint; ep != "" { + host, _, err := net.SplitHostPort(ep) + if err != nil { + return nil, fmt.Errorf("telemetry endpoint %q is not a valid host:port: %w", ep, err) + } + if host == "" { + return nil, fmt.Errorf("telemetry endpoint %q has no host", ep) + } + } + return t, nil +} + func LoadConfiguration( ctx context.Context, client client.Reader, @@ -122,18 +153,9 @@ func LoadConfiguration( return nil, err } - var telemetry *Telemetry - if config.Telemetry != nil && config.Telemetry.Enabled { - if err := config.Telemetry.Validate(); err != nil { - return nil, err - } - // Auto-derive the gRPC address when the operator has not overridden it. - // The well-known service name follows the same pattern as the controller - // and router: ..svc (in-cluster DNS). - if config.Telemetry.Endpoint == "" { - config.Telemetry.Endpoint = "jumpstarter-telemetry." + key.Namespace + ":9093" - } - telemetry = config.Telemetry + telemetry, err := resolveTelemetryConfig(config.Telemetry) + if err != nil { + return nil, err } return &LoadedConfig{ diff --git a/controller/internal/config/types.go b/controller/internal/config/types.go index 3a52f5041..890f10ac2 100644 --- a/controller/internal/config/types.go +++ b/controller/internal/config/types.go @@ -30,15 +30,22 @@ type Telemetry struct { // When true the controller advertises the endpoint returned by GetServiceEndpoints. Enabled bool `json:"enabled,omitempty" yaml:"enabled,omitempty"` - // Endpoint is an optional override for the telemetry gRPC address. - // When empty and Enabled is true, defaults to - // "jumpstarter-telemetry.:9093" derived from the controller namespace. + // Endpoint is an optional override for the telemetry gRPC address advertised + // to exporters. When empty the controller reads GRPC_TELEMETRY_ENDPOINT from its + // own environment (set by the operator on the controller Deployment). Endpoint string `json:"endpoint,omitempty" yaml:"endpoint,omitempty"` - // Certificate is reserved for a future phase where the telemetry service manages - // its own TLS credentials. Leave empty for Phase 1 deployments — the telemetry - // server listens on plaintext gRPC and exporters that receive a certificate here - // will fail to connect. + // Certificate is the PEM-encoded CA certificate that exporters use to verify + // the telemetry server's TLS certificate. + // + // When the operator provisions the telemetry service with a cert-manager-issued + // certificate, set this to the issuer's CA certificate. + // + // When the telemetry service runs in self-signed mode (no EXTERNAL_CERT_PEM / + // EXTERNAL_KEY_PEM set), it logs the generated certificate PEM at startup under + // the key "certPEM". Copy that value here so exporters can pin and verify it. + // A self-signed certificate is not trusted by the system CA pool, so leaving + // this field empty means exporters cannot establish a verified TLS connection. Certificate string `json:"certificate,omitempty" yaml:"certificate,omitempty"` // Logging configures the log ingestion path to the telemetry service. diff --git a/controller/internal/config/types_test.go b/controller/internal/config/types_test.go index 232bbb998..a8e6affa1 100644 --- a/controller/internal/config/types_test.go +++ b/controller/internal/config/types_test.go @@ -276,6 +276,97 @@ func TestDeprecatedLabelsOmitEmpty(t *testing.T) { } } +func TestTelemetryEndpointResolution(t *testing.T) { + tests := []struct { + name string + cfg *Telemetry + envValue string + wantNil bool + wantEndpoint string + wantErr bool + }{ + { + name: "nil config returns nil", + cfg: nil, + wantNil: true, + }, + { + name: "disabled config returns nil", + cfg: &Telemetry{Enabled: false, Endpoint: "telemetry:9093"}, + wantNil: true, + }, + { + name: "env var takes precedence over ConfigMap", + cfg: &Telemetry{Enabled: true, Endpoint: "telemetry.ns.svc:9093"}, + envValue: "env-telemetry:9093", + wantEndpoint: "env-telemetry:9093", + }, + { + name: "ConfigMap fallback when env var is empty", + cfg: &Telemetry{Enabled: true, Endpoint: "telemetry.ns.svc:9093"}, + wantEndpoint: "telemetry.ns.svc:9093", + }, + { + name: "both empty yields empty endpoint (no error)", + cfg: &Telemetry{Enabled: true}, + wantEndpoint: "", + }, + { + name: "malformed ConfigMap value is rejected", + cfg: &Telemetry{Enabled: true, Endpoint: "no-port"}, + wantErr: true, + }, + { + name: "malformed env var is rejected", + cfg: &Telemetry{Enabled: true}, + envValue: "garbage-no-port", + wantErr: true, + }, + { + name: "port-only ConfigMap value is rejected", + cfg: &Telemetry{Enabled: true, Endpoint: ":9093"}, + wantErr: true, + }, + { + name: "port-only env var is rejected", + cfg: &Telemetry{Enabled: true}, + envValue: ":9093", + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Setenv("GRPC_TELEMETRY_ENDPOINT", tt.envValue) + + resolved, err := resolveTelemetryConfig(tt.cfg) + + if tt.wantErr { + if err == nil { + t.Fatalf("expected validation error, got nil (resolved=%+v)", resolved) + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if tt.wantNil { + if resolved != nil { + t.Fatalf("expected nil, got %+v", resolved) + } + return + } + var gotEndpoint string + if resolved != nil { + gotEndpoint = resolved.Endpoint + } + if gotEndpoint != tt.wantEndpoint { + t.Errorf("resolved.Endpoint = %q, want %q", gotEndpoint, tt.wantEndpoint) + } + }) + } +} + func TestParseDuration(t *testing.T) { tests := []struct { input string diff --git a/controller/internal/service/controller_service.go b/controller/internal/service/controller_service.go index 2e42adb9b..bbbf0b841 100644 --- a/controller/internal/service/controller_service.go +++ b/controller/internal/service/controller_service.go @@ -327,6 +327,11 @@ func (s *ControllerService) GetServiceEndpoints( resp := &pb.GetServiceEndpointsResponse{} if s.TelemetryConfig != nil && s.TelemetryConfig.Enabled { + // Endpoint is resolved at config-load time (ConfigMap value or GRPC_TELEMETRY_ENDPOINT + // env var fallback), so TelemetryConfig.Endpoint is always the complete value here. + if s.TelemetryConfig.Endpoint == "" { + return nil, status.Error(codes.FailedPrecondition, "telemetry is enabled but no endpoint is configured; set telemetry.endpoint in the ConfigMap or GRPC_TELEMETRY_ENDPOINT on the controller pod") + } resp.TelemetryEndpoints = append(resp.TelemetryEndpoints, &pb.TelemetryEndpoint{ Endpoint: s.TelemetryConfig.Endpoint, Certificate: s.TelemetryConfig.Certificate, @@ -1196,32 +1201,9 @@ func (s *ControllerService) Start(ctx context.Context) error { return err } - // Load external certificate if provided via environment variables. - // Environment variables EXTERNAL_CERT_PEM and EXTERNAL_KEY_PEM should contain the PEM-encoded - // certificate and private key respectively. If both are set, they are used; otherwise - // a self-signed certificate is generated. - var cert *tls.Certificate - certPEMPath := os.Getenv("EXTERNAL_CERT_PEM") - keyPEMPath := os.Getenv("EXTERNAL_KEY_PEM") - if certPEMPath != "" && keyPEMPath != "" { - certPEMBytes, err := os.ReadFile(certPEMPath) - if err != nil { - return fmt.Errorf("failed to read external certificate file: %w", err) - } - keyPEMBytes, err := os.ReadFile(keyPEMPath) - if err != nil { - return fmt.Errorf("failed to read external key file: %w", err) - } - parsedCert, err := tls.X509KeyPair(certPEMBytes, keyPEMBytes) - if err != nil { - return fmt.Errorf("failed to parse external certificate: %w", err) - } - cert = &parsedCert - } else { - cert, err = NewSelfSignedCertificate("jumpstarter controller", dnsnames, ipaddresses) - if err != nil { - return err - } + cert, _, err := LoadTLSCertificate("jumpstarter controller", dnsnames, ipaddresses) + if err != nil { + return err } opts := append(s.ServerOptions, @@ -1264,8 +1246,11 @@ func (s *ControllerService) Start(ctx context.Context) error { // Register gRPC gateway gwmux := gwruntime.NewServeMux() + // The controller multiplexes gRPC (h2) and REST (http/1.1) on a single port, + // so it needs NextProtos — which LoadTLSCredentials doesn't expose. listener, err := tls.Listen("tcp", ":8082", &tls.Config{ Certificates: []tls.Certificate{*cert}, + MinVersion: tls.VersionTLS12, NextProtos: []string{"http/1.1", "h2"}, }) if err != nil { diff --git a/controller/internal/service/controller_service_test.go b/controller/internal/service/controller_service_test.go index cd48d1a8d..2848d3cbd 100644 --- a/controller/internal/service/controller_service_test.go +++ b/controller/internal/service/controller_service_test.go @@ -30,6 +30,7 @@ import ( "github.com/golang-jwt/jwt/v5" jumpstarterdevv1alpha1 "github.com/jumpstarter-dev/jumpstarter/controller/api/v1alpha1" + "github.com/jumpstarter-dev/jumpstarter/controller/internal/config" jlog "github.com/jumpstarter-dev/jumpstarter/controller/internal/log" pb "github.com/jumpstarter-dev/jumpstarter/controller/internal/protocol/jumpstarter/v1" "google.golang.org/grpc" @@ -39,9 +40,11 @@ import ( "google.golang.org/grpc/status" "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + k8sruntime "k8s.io/apimachinery/pkg/runtime" "k8s.io/apiserver/pkg/authentication/authenticator" "k8s.io/apiserver/pkg/authentication/user" "k8s.io/apiserver/pkg/authorization/authorizer" + "sigs.k8s.io/controller-runtime/pkg/client/fake" logf "sigs.k8s.io/controller-runtime/pkg/log" ctrlzap "sigs.k8s.io/controller-runtime/pkg/log/zap" ) @@ -2061,6 +2064,57 @@ func (noopAuthorizer) Authorize(_ context.Context, _ authorizer.Attributes) (aut return authorizer.DecisionNoOpinion, "", nil } +// passingAuthenticator always authenticates successfully with a fixed user name. +type passingAuthenticator struct{ userName string } + +func (p *passingAuthenticator) AuthenticateContext(_ context.Context) (*authenticator.Response, bool, error) { + return &authenticator.Response{User: &user.DefaultInfo{Name: p.userName}}, true, nil +} + +// exporterAttributesGetter returns attributes that identify a fixed Exporter object. +type exporterAttributesGetter struct{ namespace, name string } + +func (e *exporterAttributesGetter) ContextAttributes(_ context.Context, u user.Info) (authorizer.Attributes, error) { + return authorizer.AttributesRecord{ + User: u, + Namespace: e.namespace, + Resource: "Exporter", + Name: e.name, + }, nil +} + +// passingAuthorizer always allows. +type passingAuthorizer struct{} + +func (passingAuthorizer) Authorize(_ context.Context, _ authorizer.Attributes) (authorizer.Decision, string, error) { + return authorizer.DecisionAllow, "", nil +} + +// authSuccessServiceCtx builds a ControllerService whose authentication always +// succeeds. A pre-populated Exporter object is stored in the fake client so +// that VerifyExporterObjectToken can fetch it. +func authSuccessServiceCtx(t *testing.T, cfg *config.Telemetry) (*ControllerService, context.Context) { + t.Helper() + + scheme := k8sruntime.NewScheme() + if err := jumpstarterdevv1alpha1.AddToScheme(scheme); err != nil { + t.Fatalf("failed to add scheme: %v", err) + } + exporter := &jumpstarterdevv1alpha1.Exporter{ + ObjectMeta: metav1.ObjectMeta{Name: "test-exporter", Namespace: "default"}, + } + fakeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(exporter).Build() + + svc := &ControllerService{ + Client: fakeClient, + Authn: &passingAuthenticator{userName: "test-user"}, + Authz: passingAuthorizer{}, + Attr: &exporterAttributesGetter{namespace: "default", name: "test-exporter"}, + TelemetryConfig: cfg, + } + return svc, context.Background() +} + // authFailureServiceCtx builds a ControllerService whose authentication always // fails, plus a context carrying a peer address, a captured logger, and the // jlog.LogContext enrichment applied by the gRPC interceptors in production. diff --git a/controller/internal/service/endpoints.go b/controller/internal/service/endpoints.go index 93d0cc379..237de5b36 100644 --- a/controller/internal/service/endpoints.go +++ b/controller/internal/service/endpoints.go @@ -1,6 +1,7 @@ package service import ( + "fmt" "net" "os" ) @@ -21,11 +22,36 @@ func routerEndpoint() string { return ep } -func endpointToSAN(endpoint string) ([]string, []net.IP, error) { - host, _, err := net.SplitHostPort(endpoint) +// telemetryEndpoint returns the GRPC_TELEMETRY_ENDPOINT env var value. +// Returns ("", nil) if unset, or an error if set but malformed. +func telemetryEndpoint() (string, error) { + ep := os.Getenv("GRPC_TELEMETRY_ENDPOINT") + if ep == "" { + return "", nil + } + if err := validateHostPort(ep); err != nil { + return "", fmt.Errorf("GRPC_TELEMETRY_ENDPOINT %q is not a valid host:port: %w", ep, err) + } + return ep, nil +} + +// validateHostPort checks that s is a valid "host:port" with a non-empty host. +func validateHostPort(s string) error { + host, _, err := net.SplitHostPort(s) if err != nil { + return err + } + if host == "" { + return fmt.Errorf("endpoint %q has no host", s) + } + return nil +} + +func endpointToSAN(endpoint string) ([]string, []net.IP, error) { + if err := validateHostPort(endpoint); err != nil { return nil, nil, err } + host, _, _ := net.SplitHostPort(endpoint) ip := net.ParseIP(host) if ip != nil { return []string{}, []net.IP{ip}, nil diff --git a/controller/internal/service/router_service.go b/controller/internal/service/router_service.go index 2684e2046..0fa8e2b29 100644 --- a/controller/internal/service/router_service.go +++ b/controller/internal/service/router_service.go @@ -18,8 +18,6 @@ package service import ( "context" - "crypto/tls" - "fmt" "net" "os" "sync" @@ -31,7 +29,6 @@ import ( pb "github.com/jumpstarter-dev/jumpstarter/controller/internal/protocol/jumpstarter/v1" "google.golang.org/grpc" "google.golang.org/grpc/codes" - "google.golang.org/grpc/credentials" "google.golang.org/grpc/reflection" "google.golang.org/grpc/status" ctrl "sigs.k8s.io/controller-runtime" @@ -127,36 +124,13 @@ func (s *RouterService) Start(ctx context.Context) error { return err } - // Handle external certificate if provided via environment variables. - // Environment variables EXTERNAL_CERT_PEM and EXTERNAL_KEY_PEM should contain the PEM-encoded - // certificate and private key respectively. If both are set, they are used; otherwise - // a self-signed certificate is generated. - var cert *tls.Certificate - certPEMPath := os.Getenv("EXTERNAL_CERT_PEM") - keyPEMPath := os.Getenv("EXTERNAL_KEY_PEM") - if certPEMPath != "" && keyPEMPath != "" { - certPEMBytes, err := os.ReadFile(certPEMPath) - if err != nil { - return fmt.Errorf("failed to read external certificate file: %w", err) - } - keyPEMBytes, err := os.ReadFile(keyPEMPath) - if err != nil { - return fmt.Errorf("failed to read external key file: %w", err) - } - parsedCert, err := tls.X509KeyPair(certPEMBytes, keyPEMBytes) - if err != nil { - return fmt.Errorf("failed to parse external certificate: %w", err) - } - cert = &parsedCert - } else { - cert, err = NewSelfSignedCertificate("jumpstarter router", dnsnames, ipaddresses) - if err != nil { - return err - } + tlsCreds, _, err := LoadTLSCredentials("jumpstarter router", dnsnames, ipaddresses) + if err != nil { + return err } opts := []grpc.ServerOption{ - grpc.Creds(credentials.NewServerTLSFromCert(cert)), + grpc.Creds(tlsCreds), grpc.ChainUnaryInterceptor(recovery.UnaryServerInterceptor()), grpc.ChainStreamInterceptor(recovery.StreamServerInterceptor()), } diff --git a/controller/internal/service/telemetry_service.go b/controller/internal/service/telemetry_service.go index fb4ebafac..076f044c9 100644 --- a/controller/internal/service/telemetry_service.go +++ b/controller/internal/service/telemetry_service.go @@ -24,11 +24,13 @@ import ( "strings" "time" + "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/recovery" "github.com/jumpstarter-dev/jumpstarter/controller/internal/authentication" "github.com/jumpstarter-dev/jumpstarter/controller/internal/oidc" pb "github.com/jumpstarter-dev/jumpstarter/controller/internal/protocol/jumpstarter/v1" "google.golang.org/grpc" "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials" "google.golang.org/grpc/reflection" "google.golang.org/grpc/status" ctrl "sigs.k8s.io/controller-runtime" @@ -56,12 +58,11 @@ var reservedExtraFieldKeys = map[string]struct{}{ // TelemetryService receives structured log entries from exporters and clients, // logs them via structured stdout, and will forward them to Loki in a future phase. // -// Phase 1 design: the server listens on plaintext gRPC only. TLS termination is -// expected to be handled by a sidecar (e.g. Envoy) or service mesh in production -// deployments. The Certificate field advertised via GetServiceEndpoints is reserved -// for a future phase where the telemetry binary manages its own TLS credentials. -// Do NOT configure the Certificate field in the ConfigMap for Phase 1 deployments — -// exporters that receive a certificate will attempt a TLS handshake that will fail. +// TLS: the server always uses TLS. When EXTERNAL_CERT_PEM and EXTERNAL_KEY_PEM +// env vars point to certificate/key files (mounted by the operator from a Secret), +// those are loaded. Otherwise a self-signed certificate is generated — traffic is +// still encrypted, but clients cannot verify the server identity without the CA cert +// in the ConfigMap telemetry.certificate field. type TelemetryService struct { pb.UnimplementedTelemetryServiceServer @@ -203,16 +204,63 @@ func truncate(s string, n int) string { return s[:b] } +// loadTLSCredentials loads TLS credentials for the telemetry gRPC server. +// It derives the self-signed certificate SAN from the advertised endpoint +// (GRPC_TELEMETRY_ENDPOINT) so that TLS hostname verification succeeds when +// exporters connect. Falls back to "localhost" in development/local mode. +// Returns an error if GRPC_TELEMETRY_ENDPOINT is set but malformed. +// Delegates cert loading to the shared LoadTLSCredentials helper. +func (s *TelemetryService) loadTLSCredentials() (credentials.TransportCredentials, string, error) { + // Derive SANs from the advertised endpoint (what clients connect to), + // not from the bind address (which is a local port like ":9093"). + // IMPORTANT: GRPC_TELEMETRY_ENDPOINT must be set on the telemetry pod itself + // so the SAN matches the endpoint the controller advertises to exporters. + advertised, err := telemetryEndpoint() + if err != nil { + return nil, "", err + } + var dnsnames []string + var ipaddresses []net.IP + if advertised != "" { + var sanErr error + dnsnames, ipaddresses, sanErr = endpointToSAN(advertised) + if sanErr != nil { + ctrl.Log.WithName("telemetry").Error(sanErr, "failed to derive SAN from advertised endpoint; falling back to localhost", + "endpoint", advertised) + dnsnames = []string{"localhost"} + } + } else { + // No advertised endpoint configured — development/local mode. + dnsnames = []string{"localhost"} + } + return LoadTLSCredentials("jumpstarter telemetry", dnsnames, ipaddresses) +} + // Start starts the TelemetryService gRPC server and blocks until ctx is cancelled. func (s *TelemetryService) Start(ctx context.Context) error { logger := ctrl.Log.WithName("telemetry").WithValues("component", "telemetry") + creds, selfSignedPEM, err := s.loadTLSCredentials() + if err != nil { + return fmt.Errorf("telemetry: load TLS credentials: %w", err) + } + if selfSignedPEM != "" { + // Log the self-signed cert so the operator can copy it into the controller + // ConfigMap's telemetry.certificate field. Exporters need this PEM to verify + // the TLS connection — a self-signed cert is not trusted by the system CA pool. + logger.Info("Using self-signed TLS certificate; copy certPEM into the controller ConfigMap telemetry.certificate so exporters can verify TLS", + "certPEM", selfSignedPEM) + } + lis, err := net.Listen("tcp", s.BindAddr) if err != nil { return fmt.Errorf("telemetry: listen %s: %w", s.BindAddr, err) } - srv := grpc.NewServer() + srv := grpc.NewServer( + grpc.Creds(creds), + grpc.ChainUnaryInterceptor(recovery.UnaryServerInterceptor()), + ) pb.RegisterTelemetryServiceServer(srv, s) reflection.Register(srv) @@ -231,6 +279,7 @@ func (s *TelemetryService) Start(ctx context.Context) error { } return nil case err := <-errCh: + srv.Stop() return err } } diff --git a/controller/internal/service/telemetry_service_test.go b/controller/internal/service/telemetry_service_test.go index f3677ea64..20740f26f 100644 --- a/controller/internal/service/telemetry_service_test.go +++ b/controller/internal/service/telemetry_service_test.go @@ -19,14 +19,19 @@ package service import ( "bytes" "context" + "crypto/x509" + "encoding/pem" "fmt" + "os" "strings" "testing" "github.com/jumpstarter-dev/jumpstarter/controller/internal/config" "github.com/jumpstarter-dev/jumpstarter/controller/internal/oidc" pb "github.com/jumpstarter-dev/jumpstarter/controller/internal/protocol/jumpstarter/v1" + "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" logf "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/log/zap" ) @@ -85,40 +90,32 @@ func TestGetServiceEndpoints_RequiresAuthentication(t *testing.T) { } } -// buildTelemetryEndpointsResponse exercises the response-building logic -// without going through the auth gate, for isolated unit testing. -func buildTelemetryEndpointsResponse(cfg *config.Telemetry) *pb.GetServiceEndpointsResponse { - resp := &pb.GetServiceEndpointsResponse{} - if cfg != nil && cfg.Enabled { - minSev := cfg.Logging.Filter.MinSeverity - if minSev == "" { - minSev = "info" - } - resp.TelemetryEndpoints = append(resp.TelemetryEndpoints, &pb.TelemetryEndpoint{ - Endpoint: cfg.Endpoint, - Certificate: cfg.Certificate, - MinSeverity: minSev, - }) - } - return resp -} - func TestGetServiceEndpoints_NilConfig_ReturnsEmptyList(t *testing.T) { - resp := buildTelemetryEndpointsResponse(nil) + svc, ctx := authSuccessServiceCtx(t, nil) + + resp, err := svc.GetServiceEndpoints(ctx, &pb.GetServiceEndpointsRequest{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } if len(resp.TelemetryEndpoints) != 0 { t.Errorf("expected empty telemetry_endpoints, got %d", len(resp.TelemetryEndpoints)) } } func TestGetServiceEndpoints_DisabledConfig_ReturnsEmptyList(t *testing.T) { - resp := buildTelemetryEndpointsResponse(&config.Telemetry{Enabled: false, Endpoint: "telemetry:9093"}) + svc, ctx := authSuccessServiceCtx(t, &config.Telemetry{Enabled: false, Endpoint: "telemetry:9093"}) + + resp, err := svc.GetServiceEndpoints(ctx, &pb.GetServiceEndpointsRequest{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } if len(resp.TelemetryEndpoints) != 0 { t.Errorf("expected empty telemetry_endpoints when disabled, got %d", len(resp.TelemetryEndpoints)) } } func TestGetServiceEndpoints_WithEndpoint_ReturnsEndpoint(t *testing.T) { - resp := buildTelemetryEndpointsResponse(&config.Telemetry{ + svc, ctx := authSuccessServiceCtx(t, &config.Telemetry{ Enabled: true, Endpoint: "telemetry.jumpstarter.svc:9093", Certificate: "--- PEM ---", @@ -127,6 +124,11 @@ func TestGetServiceEndpoints_WithEndpoint_ReturnsEndpoint(t *testing.T) { }, }) + resp, err := svc.GetServiceEndpoints(ctx, &pb.GetServiceEndpointsRequest{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(resp.TelemetryEndpoints) != 1 { t.Fatalf("expected 1 telemetry endpoint, got %d", len(resp.TelemetryEndpoints)) } @@ -143,11 +145,16 @@ func TestGetServiceEndpoints_WithEndpoint_ReturnsEndpoint(t *testing.T) { } func TestGetServiceEndpoints_DefaultsMinSeverityToInfo(t *testing.T) { - resp := buildTelemetryEndpointsResponse(&config.Telemetry{ + svc, ctx := authSuccessServiceCtx(t, &config.Telemetry{ Enabled: true, Endpoint: "telemetry:9093", }) + resp, err := svc.GetServiceEndpoints(ctx, &pb.GetServiceEndpointsRequest{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(resp.TelemetryEndpoints) != 1 { t.Fatalf("expected 1 endpoint, got %d", len(resp.TelemetryEndpoints)) } @@ -156,6 +163,269 @@ func TestGetServiceEndpoints_DefaultsMinSeverityToInfo(t *testing.T) { } } +func TestGetServiceEndpoints_EnabledNoEndpoint_FailedPrecondition(t *testing.T) { + // This test exercises the empty-endpoint guard in the real GetServiceEndpoints + // handler — after authentication succeeds. Previously all GetServiceEndpoints + // tests used a failing authenticator, so this branch was never reached. + svc, ctx := authSuccessServiceCtx(t, &config.Telemetry{Enabled: true, Endpoint: ""}) + + _, err := svc.GetServiceEndpoints(ctx, &pb.GetServiceEndpointsRequest{}) + if err == nil { + t.Fatal("expected error when telemetry enabled but endpoint is empty, got nil") + } + if status.Code(err) != codes.FailedPrecondition { + t.Errorf("expected codes.FailedPrecondition, got %v", status.Code(err)) + } +} + +// writeTLSPEMFiles generates a self-signed cert, marshals it to PEM, and writes +// cert and key to temporary files. Returns (certPath, keyPath). +func writeTLSPEMFiles(t *testing.T) (certPath, keyPath string) { + t.Helper() + + tlsCert, err := NewSelfSignedCertificate("test", []string{"localhost"}, nil) + if err != nil { + t.Fatalf("NewSelfSignedCertificate: %v", err) + } + + certPEM := pem.EncodeToMemory(&pem.Block{ + Type: "CERTIFICATE", + Bytes: tlsCert.Certificate[0], + }) + + keyDER, err := x509.MarshalPKCS8PrivateKey(tlsCert.PrivateKey) + if err != nil { + t.Fatalf("MarshalPKCS8PrivateKey: %v", err) + } + keyPEM := pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: keyDER}) + + dir := t.TempDir() + certPath = dir + "/tls.crt" + keyPath = dir + "/tls.key" + if err := os.WriteFile(certPath, certPEM, 0o600); err != nil { + t.Fatalf("write cert: %v", err) + } + if err := os.WriteFile(keyPath, keyPEM, 0o600); err != nil { + t.Fatalf("write key: %v", err) + } + return certPath, keyPath +} + +func TestTelemetryService_LoadTLSCredentials_SelfSigned(t *testing.T) { + t.Setenv("EXTERNAL_CERT_PEM", "") + t.Setenv("EXTERNAL_KEY_PEM", "") + + svc := &TelemetryService{BindAddr: ":9093", Signer: testSigner(t)} + creds, selfSignedPEM, err := svc.loadTLSCredentials() + if err != nil { + t.Fatalf("loadTLSCredentials() with self-signed cert failed: %v", err) + } + if creds == nil { + t.Fatal("expected non-nil credentials") + } + if selfSignedPEM == "" { + t.Error("expected non-empty selfSignedPEM when no external cert is configured") + } + // Must be parseable PEM. + block, _ := pem.Decode([]byte(selfSignedPEM)) + if block == nil { + t.Errorf("selfSignedPEM is not valid PEM: %s", selfSignedPEM) + } +} + +func TestTelemetryService_LoadTLSCredentials_SelfSignedUsesAdvertisedEndpointForSAN(t *testing.T) { + // The self-signed cert SAN must match the advertised endpoint hostname so + // that TLS hostname verification succeeds when exporters connect. + t.Setenv("EXTERNAL_CERT_PEM", "") + t.Setenv("EXTERNAL_KEY_PEM", "") + t.Setenv("GRPC_TELEMETRY_ENDPOINT", "telemetry.jumpstarter.svc:9093") + + svc := &TelemetryService{BindAddr: ":9093", Signer: testSigner(t)} + _, selfSignedPEM, err := svc.loadTLSCredentials() + if err != nil { + t.Fatalf("loadTLSCredentials() failed: %v", err) + } + if selfSignedPEM == "" { + t.Fatal("expected non-empty selfSignedPEM") + } + + block, _ := pem.Decode([]byte(selfSignedPEM)) + if block == nil { + t.Fatal("selfSignedPEM is not valid PEM") + } + leaf, err := x509.ParseCertificate(block.Bytes) + if err != nil { + t.Fatalf("ParseCertificate: %v", err) + } + if len(leaf.DNSNames) != 1 || leaf.DNSNames[0] != "telemetry.jumpstarter.svc" { + t.Errorf("expected SAN [telemetry.jumpstarter.svc], got %v", leaf.DNSNames) + } +} + +func TestTelemetryService_LoadTLSCredentials_SelfSignedFallsBackToLocalhostWhenNoEndpoint(t *testing.T) { + // When GRPC_TELEMETRY_ENDPOINT is unset, the self-signed cert SAN defaults to "localhost". + t.Setenv("EXTERNAL_CERT_PEM", "") + t.Setenv("EXTERNAL_KEY_PEM", "") + t.Setenv("GRPC_TELEMETRY_ENDPOINT", "") + + svc := &TelemetryService{BindAddr: ":9093", Signer: testSigner(t)} + _, selfSignedPEM, err := svc.loadTLSCredentials() + if err != nil { + t.Fatalf("loadTLSCredentials() failed: %v", err) + } + + block, _ := pem.Decode([]byte(selfSignedPEM)) + if block == nil { + t.Fatal("selfSignedPEM is not valid PEM") + } + leaf, err := x509.ParseCertificate(block.Bytes) + if err != nil { + t.Fatalf("ParseCertificate: %v", err) + } + if len(leaf.DNSNames) != 1 || leaf.DNSNames[0] != "localhost" { + t.Errorf("expected SAN [localhost], got %v", leaf.DNSNames) + } +} + +func TestTelemetryService_LoadTLSCredentials_OnlyCertEnvVarReturnsError(t *testing.T) { + certPath, _ := writeTLSPEMFiles(t) + // Only cert set, key missing — must fail to avoid hiding a broken Secret mount. + t.Setenv("EXTERNAL_CERT_PEM", certPath) + t.Setenv("EXTERNAL_KEY_PEM", "") + + svc := &TelemetryService{BindAddr: ":9093", Signer: testSigner(t)} + _, _, err := svc.loadTLSCredentials() + if err == nil { + t.Fatal("expected error when only EXTERNAL_CERT_PEM is set") + } + if !strings.Contains(err.Error(), "EXTERNAL_CERT_PEM and EXTERNAL_KEY_PEM must be set together") { + t.Errorf("unexpected error message: %v", err) + } +} + +func TestTelemetryService_LoadTLSCredentials_OnlyKeyEnvVarReturnsError(t *testing.T) { + _, keyPath := writeTLSPEMFiles(t) + // Only key set, cert missing — must fail. + t.Setenv("EXTERNAL_CERT_PEM", "") + t.Setenv("EXTERNAL_KEY_PEM", keyPath) + + svc := &TelemetryService{BindAddr: ":9093", Signer: testSigner(t)} + _, _, err := svc.loadTLSCredentials() + if err == nil { + t.Fatal("expected error when only EXTERNAL_KEY_PEM is set") + } + if !strings.Contains(err.Error(), "EXTERNAL_CERT_PEM and EXTERNAL_KEY_PEM must be set together") { + t.Errorf("unexpected error message: %v", err) + } +} + +func TestTelemetryService_LoadTLSCredentials_WithValidPEMFiles(t *testing.T) { + certPath, keyPath := writeTLSPEMFiles(t) + t.Setenv("EXTERNAL_CERT_PEM", certPath) + t.Setenv("EXTERNAL_KEY_PEM", keyPath) + + svc := &TelemetryService{BindAddr: ":9093", Signer: testSigner(t)} + creds, selfSignedPEM, err := svc.loadTLSCredentials() + if err != nil { + t.Fatalf("loadTLSCredentials() with valid PEM files failed: %v", err) + } + if creds == nil { + t.Fatal("expected non-nil credentials") + } + // External cert provided — selfSignedPEM must be empty. + if selfSignedPEM != "" { + t.Errorf("expected empty selfSignedPEM when external cert is provided, got non-empty") + } +} + +func TestTelemetryService_LoadTLSCredentials_BadCertFileReturnsError(t *testing.T) { + certFile, err := os.CreateTemp(t.TempDir(), "tls-*.crt") + if err != nil { + t.Fatalf("CreateTemp: %v", err) + } + if err := certFile.Close(); err != nil { + t.Fatalf("close: %v", err) + } + _, keyPath := writeTLSPEMFiles(t) + + t.Setenv("EXTERNAL_CERT_PEM", certFile.Name()) + t.Setenv("EXTERNAL_KEY_PEM", keyPath) + + svc := &TelemetryService{BindAddr: ":9093", Signer: testSigner(t)} + _, _, err = svc.loadTLSCredentials() + if err == nil { + t.Fatal("expected error parsing empty cert file") + } + // Must be a parse error, not a file-not-found. + if strings.Contains(err.Error(), "no such file") { + t.Errorf("expected parse error, got: %v", err) + } +} + +func TestTelemetryService_LoadTLSCredentials_MissingCertFileReturnsError(t *testing.T) { + _, keyPath := writeTLSPEMFiles(t) + t.Setenv("EXTERNAL_CERT_PEM", "/does/not/exist/tls.crt") + t.Setenv("EXTERNAL_KEY_PEM", keyPath) + + svc := &TelemetryService{BindAddr: ":9093", Signer: testSigner(t)} + _, _, err := svc.loadTLSCredentials() + if err == nil { + t.Fatal("expected error reading missing cert file") + } +} + +func TestTelemetryService_LoadTLSCredentials_MissingKeyFileReturnsError(t *testing.T) { + certPath, _ := writeTLSPEMFiles(t) + t.Setenv("EXTERNAL_CERT_PEM", certPath) + t.Setenv("EXTERNAL_KEY_PEM", "/does/not/exist/tls.key") + + svc := &TelemetryService{BindAddr: ":9093", Signer: testSigner(t)} + _, _, err := svc.loadTLSCredentials() + if err == nil { + t.Fatal("expected error reading missing key file") + } + if !strings.Contains(err.Error(), "key") { + t.Errorf("expected 'key' in error message, got: %v", err) + } +} + +func TestTelemetryService_LoadTLSCredentials_ValidCertInvalidKeyReturnsError(t *testing.T) { + certPath, _ := writeTLSPEMFiles(t) + + keyFile, err := os.CreateTemp(t.TempDir(), "tls-*.key") + if err != nil { + t.Fatalf("CreateTemp: %v", err) + } + if err := keyFile.Close(); err != nil { + t.Fatalf("close: %v", err) + } + + t.Setenv("EXTERNAL_CERT_PEM", certPath) + t.Setenv("EXTERNAL_KEY_PEM", keyFile.Name()) + + svc := &TelemetryService{BindAddr: ":9093", Signer: testSigner(t)} + _, _, err = svc.loadTLSCredentials() + if err == nil { + t.Fatal("expected error parsing mismatched cert/key pair") + } +} + +func TestTelemetryService_Start_FailsWhenExternalCertFileMissing(t *testing.T) { + t.Setenv("EXTERNAL_CERT_PEM", "/no/such/cert.pem") + t.Setenv("EXTERNAL_KEY_PEM", "/no/such/key.pem") + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + svc := &TelemetryService{BindAddr: ":0", Signer: testSigner(t)} + err := svc.Start(ctx) + if err == nil { + t.Fatal("expected Start to fail with missing cert files") + } + if !strings.Contains(err.Error(), "TLS") { + t.Errorf("expected 'TLS' in error, got: %v", err) + } +} + func TestTelemetryService_PushLogs_RequiresAuthentication(t *testing.T) { svc := &TelemetryService{BindAddr: ":0", Signer: testSigner(t)} @@ -458,6 +728,82 @@ func TestTelemetryService_PushLogs_StripsReservedExtraFieldKeys(t *testing.T) { } } +func TestTelemetryEndpoint(t *testing.T) { + tests := []struct { + name string + env string + want string + wantErr bool + }{ + {"empty", "", "", false}, + {"valid host:port", "telemetry.svc:9093", "telemetry.svc:9093", false}, + {"valid IP:port", "10.0.0.1:9093", "10.0.0.1:9093", false}, + {"missing port", "telemetry.svc", "", true}, + {"just port", ":9093", "", true}, + {"garbage", "not a valid endpoint!", "", true}, + {"has scheme", "http://host:9093", "", true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Setenv("GRPC_TELEMETRY_ENDPOINT", tt.env) + got, err := telemetryEndpoint() + if tt.wantErr { + if err == nil { + t.Fatalf("telemetryEndpoint() expected error for %q, got %q", tt.env, got) + } + return + } + if err != nil { + t.Fatalf("telemetryEndpoint() unexpected error: %v", err) + } + if got != tt.want { + t.Errorf("telemetryEndpoint() = %q, want %q", got, tt.want) + } + }) + } +} + +func TestEndpointToSAN(t *testing.T) { + tests := []struct { + name string + endpoint string + wantDNS []string + wantIPs int + wantErr bool + }{ + {"valid hostname:port", "telemetry.svc:9093", []string{"telemetry.svc"}, 0, false}, + {"valid IP:port", "10.0.0.1:9093", nil, 1, false}, + {"port-only is rejected", ":9093", nil, 0, true}, + {"missing port", "telemetry.svc", nil, 0, true}, + {"empty string", "", nil, 0, true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + dns, ips, err := endpointToSAN(tt.endpoint) + if tt.wantErr { + if err == nil { + t.Fatalf("expected error for %q, got dns=%v ips=%v", tt.endpoint, dns, ips) + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(dns) != len(tt.wantDNS) { + t.Errorf("dns = %v, want %v", dns, tt.wantDNS) + } + for i := range tt.wantDNS { + if i < len(dns) && dns[i] != tt.wantDNS[i] { + t.Errorf("dns[%d] = %q, want %q", i, dns[i], tt.wantDNS[i]) + } + } + if len(ips) != tt.wantIPs { + t.Errorf("got %d IPs, want %d", len(ips), tt.wantIPs) + } + }) + } +} + func TestTelemetryService_truncate(t *testing.T) { cases := []struct { input string diff --git a/controller/internal/service/tls_credentials.go b/controller/internal/service/tls_credentials.go new file mode 100644 index 000000000..0328e82d5 --- /dev/null +++ b/controller/internal/service/tls_credentials.go @@ -0,0 +1,88 @@ +/* +Copyright 2026. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package service + +import ( + "crypto/tls" + "encoding/pem" + "fmt" + "net" + "os" + + "google.golang.org/grpc/credentials" +) + +// LoadTLSCertificate loads a TLS certificate from operator-mounted files or +// generates a self-signed one as fallback. +// +// If EXTERNAL_CERT_PEM and EXTERNAL_KEY_PEM env vars both point to valid PEM +// files (set by the operator via Secret volume mounts), those are used. +// Otherwise a self-signed certificate is generated for the given SANs. +// +// selfSignedPEM is non-empty only when a self-signed certificate was generated. +// Callers that advertise the certificate to clients (e.g. the telemetry service) +// should log it so the operator can pin it in the relevant ConfigMap field. +func LoadTLSCertificate(commonName string, dnsnames []string, ipaddresses []net.IP) (*tls.Certificate, string, error) { + certPEMPath := os.Getenv("EXTERNAL_CERT_PEM") + keyPEMPath := os.Getenv("EXTERNAL_KEY_PEM") + + // Require both or neither — a partial configuration means a broken Secret + // mount that would cause clients trusting the intended certificate to fail TLS. + if (certPEMPath == "") != (keyPEMPath == "") { + return nil, "", fmt.Errorf("EXTERNAL_CERT_PEM and EXTERNAL_KEY_PEM must be set together; got cert=%q key=%q", certPEMPath, keyPEMPath) + } + + if certPEMPath != "" { + certPEMBytes, err := os.ReadFile(certPEMPath) + if err != nil { + return nil, "", fmt.Errorf("failed to read external certificate file: %w", err) + } + keyPEMBytes, err := os.ReadFile(keyPEMPath) + if err != nil { + return nil, "", fmt.Errorf("failed to read external key file: %w", err) + } + cert, err := tls.X509KeyPair(certPEMBytes, keyPEMBytes) + if err != nil { + return nil, "", fmt.Errorf("failed to parse external certificate: %w", err) + } + return &cert, "", nil + } + + cert, err := NewSelfSignedCertificate(commonName, dnsnames, ipaddresses) + if err != nil { + return nil, "", err + } + selfSignedPEM := string(pem.EncodeToMemory(&pem.Block{ + Type: "CERTIFICATE", + Bytes: cert.Certificate[0], + })) + return cert, selfSignedPEM, nil +} + +// LoadTLSCredentials returns gRPC server TLS credentials built from +// LoadTLSCertificate, enforcing a minimum TLS version of 1.2. +func LoadTLSCredentials(commonName string, dnsnames []string, ipaddresses []net.IP) (credentials.TransportCredentials, string, error) { + cert, selfSignedPEM, err := LoadTLSCertificate(commonName, dnsnames, ipaddresses) + if err != nil { + return nil, "", err + } + creds := credentials.NewTLS(&tls.Config{ + Certificates: []tls.Certificate{*cert}, + MinVersion: tls.VersionTLS12, + }) + return creds, selfSignedPEM, nil +}