From 455cf57101735e67c0ec9ab5de988d2108a7e222 Mon Sep 17 00:00:00 2001 From: kerthcet Date: Wed, 19 Aug 2026 23:49:10 +0100 Subject: [PATCH] Add support for kubelet cert Signed-off-by: kerthcet --- cmd/main.go | 30 +- config/manager/kustomization.yaml | 2 +- config/manager/manager.yaml | 5 + config/rbac/kubelet_serving_role.yaml | 46 ++ config/rbac/kubelet_serving_role_binding.yaml | 17 + config/rbac/kustomization.yaml | 6 + config/samples/deployment.yaml | 2 +- pkg/vnode/kubelet.go | 50 ++- pkg/vnode/kubelet_test.go | 6 +- pkg/vnode/servingcert.go | 282 +++++++++++++ pkg/vnode/servingcert_test.go | 395 ++++++++++++++++++ 11 files changed, 821 insertions(+), 20 deletions(-) create mode 100644 config/rbac/kubelet_serving_role.yaml create mode 100644 config/rbac/kubelet_serving_role_binding.yaml create mode 100644 pkg/vnode/servingcert.go create mode 100644 pkg/vnode/servingcert_test.go diff --git a/cmd/main.go b/cmd/main.go index 204bb4e..ed57687 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -84,6 +84,7 @@ func main() { var secureMetrics bool var enableHTTP2 bool var kubeletAddr, kubeletClientCA string + var kubeletServingCSR bool var tlsOpts []func(*tls.Config) flag.StringVar(&metricsAddr, "metrics-bind-address", "0", "The address the metrics endpoint binds to. "+ "Use :8443 for HTTPS or :8080 for HTTP, or leave as 0 to disable the metrics service.") @@ -110,6 +111,13 @@ func main() { "serves TLS without client verification, because which CA signs the API server's kubelet "+ "client certificate is not portable across distributions; restrict the port with a "+ "NetworkPolicy, or set this to your API server's kubelet client CA.") + flag.BoolVar(&kubeletServingCSR, "kubelet-serving-csr", false, + "Request the kubelet API's serving certificate from the cluster's kubernetes.io/kubelet-serving "+ + "signer instead of self-signing it. Set this if `kubectl logs` fails with "+ + "\"certificate signed by unknown authority\", which means the API server runs with "+ + "--kubelet-certificate-authority. Needs the RBAC in config/rbac/kubelet_serving_role.yaml "+ + "(it self-approves its own CSR); without it, and on a control plane whose signer is "+ + "disabled, it logs and falls back to self-signed.") opts := zap.Options{ Development: true, } @@ -270,7 +278,7 @@ func main() { // The kubelet endpoint for `kubectl logs` — one listener shared by every provider's // node, hence built here rather than in setupVirtualNodes. Nil is supported: the // nodes then advertise no address, and logs report NotFound. - kubeletSrv := setupKubeletServer(mgr, kubeletAddr, kubeletClientCA) + kubeletSrv := setupKubeletServer(mgr, kubeletAddr, kubeletClientCA, kubeletServingCSR) // Controller and webhook registration is deferred until the cert exists, so it // runs in a goroutine: the cert cannot be minted until the manager is STARTED @@ -417,7 +425,7 @@ func setupControllers(mgr ctrl.Manager, blocklist *failover.Blocklist, kubeletSr // what the API server dials and nothing substitutes for it: a Service would balance to // a non-leader replica, which holds no tracked Pods. Either way only logs degrade, so // it is logged loudly and the manager carries on. -func setupKubeletServer(mgr ctrl.Manager, addr, clientCA string) *vnode.KubeletServer { +func setupKubeletServer(mgr ctrl.Manager, addr, clientCA string, servingCSR bool) *vnode.KubeletServer { if addr == "" { setupLog.Info("kubelet API disabled by configuration; `kubectl logs` will not work for Nebula pods") return nil @@ -435,11 +443,27 @@ func setupKubeletServer(mgr ctrl.Manager, addr, clientCA string) *vnode.KubeletS setupLog.Error(err, "unable to set up the kubelet API; `kubectl logs` will not work for Nebula pods") return nil } + if servingCSR { + // Only reachable because registerProviders ran first; without a provider there is no + // virtual node, and nothing to serve logs for. The name is only the CSR's subject — + // one issued cert covers every node here, since it is the IP SAN that matters. + names := provider.Names() + clientset, err := kubernetes.NewForConfig(mgr.GetConfig()) + switch { + case err != nil: + setupLog.Error(err, "unable to build a clientset for the kubelet serving CSR; self-signing instead") + case len(names) == 0: + setupLog.Info("no provider registered; skipping the kubelet serving CSR") + default: + srv.EnableServingCSR(clientset, vnode.NodeName(names[0])) + } + } if err := mgr.Add(srv); err != nil { setupLog.Error(err, "unable to add the kubelet API to the manager") return nil } - setupLog.Info("kubelet API enabled", "addr", addr, "advertisedIP", podIP, "clientCertRequired", clientCA != "") + setupLog.Info("kubelet API enabled", "addr", addr, "advertisedIP", podIP, + "clientCertRequired", clientCA != "", "servingCSR", servingCSR) return srv } diff --git a/config/manager/kustomization.yaml b/config/manager/kustomization.yaml index 086bc1e..798472d 100644 --- a/config/manager/kustomization.yaml +++ b/config/manager/kustomization.yaml @@ -5,4 +5,4 @@ kind: Kustomization images: - name: controller newName: inftyai/nebula-controller - newTag: latest + newTag: 0819-02 diff --git a/config/manager/manager.yaml b/config/manager/manager.yaml index fbe0e7f..8c1dbb5 100644 --- a/config/manager/manager.yaml +++ b/config/manager/manager.yaml @@ -128,6 +128,11 @@ spec: # portable — requiring it would break logs on managed control planes. So # anything able to reach this port can read any Nebula pod's logs: restrict it # with a NetworkPolicy, or set --kubelet-client-ca to require mTLS. + # + # The self-signed cert is only accepted by an API server that does not verify it. + # If logs fail with "certificate signed by unknown authority", the control plane + # sets --kubelet-certificate-authority: add --kubelet-serving-csr and uncomment + # kubelet_serving_role.yaml in config/rbac/kustomization.yaml. - name: kubelet-api containerPort: 10250 protocol: TCP diff --git a/config/rbac/kubelet_serving_role.yaml b/config/rbac/kubelet_serving_role.yaml new file mode 100644 index 0000000..4eb26f7 --- /dev/null +++ b/config/rbac/kubelet_serving_role.yaml @@ -0,0 +1,46 @@ +# Opt-in RBAC for --kubelet-serving-csr. Not generated from kubebuilder markers and not +# applied by default, on purpose: `approve` on the kubelet-serving signer lets the holder +# obtain a serving certificate for ANY node's kubelet endpoint, so granting it has to be a +# deliberate act. +# +# Needed only where the API server runs with --kubelet-certificate-authority, which makes +# it verify the certificate the kubelet API presents. Without these rules the manager logs +# the failure and serves a self-signed certificate, which such an API server rejects — so +# `kubectl logs` and `kubectl exec` on Nebula pods keep failing. +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + labels: + app.kubernetes.io/name: nebula + app.kubernetes.io/managed-by: kustomize + name: kubelet-serving-role +rules: +# No update: the request is immutable once created, and a stale one (previous pod IP) is +# replaced by delete-then-create. +- apiGroups: + - certificates.k8s.io + resources: + - certificatesigningrequests + verbs: + - get + - create + - delete +# kube-controller-manager auto-approves node CLIENT certificates only, never serving ones, +# because an approver cannot verify that a requester owns the SANs it asks for. This is the +# manager asserting that about itself. +- apiGroups: + - certificates.k8s.io + resources: + - certificatesigningrequests/approval + verbs: + - update +# Approval is scoped to one signer: it authorizes nothing about client certificates, which +# are identities the API server would authenticate. +- apiGroups: + - certificates.k8s.io + resources: + - signers + resourceNames: + - kubernetes.io/kubelet-serving + verbs: + - approve diff --git a/config/rbac/kubelet_serving_role_binding.yaml b/config/rbac/kubelet_serving_role_binding.yaml new file mode 100644 index 0000000..1cf8909 --- /dev/null +++ b/config/rbac/kubelet_serving_role_binding.yaml @@ -0,0 +1,17 @@ +# Binds the opt-in kubelet-serving permissions to the manager. Applied together with +# kubelet_serving_role.yaml — see the rationale there. +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRoleBinding +metadata: + labels: + app.kubernetes.io/name: nebula + app.kubernetes.io/managed-by: kustomize + name: kubelet-serving-rolebinding +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: ClusterRole + name: kubelet-serving-role +subjects: +- kind: ServiceAccount + name: controller-manager + namespace: system diff --git a/config/rbac/kustomization.yaml b/config/rbac/kustomization.yaml index f15d94b..fc86cde 100644 --- a/config/rbac/kustomization.yaml +++ b/config/rbac/kustomization.yaml @@ -9,6 +9,12 @@ resources: - role_binding.yaml - leader_election_role.yaml - leader_election_role_binding.yaml +# Uncomment these two together with the manager's --kubelet-serving-csr flag, which is +# needed when the API server runs with --kubelet-certificate-authority. Left out by +# default because approving on the kubelet-serving signer is a privileged grant — see +# kubelet_serving_role.yaml. +#- kubelet_serving_role.yaml +#- kubelet_serving_role_binding.yaml # The following RBAC configurations are used to protect # the metrics endpoint with authn/authz. These configurations # ensure that only authorized users and service accounts diff --git a/config/samples/deployment.yaml b/config/samples/deployment.yaml index 931e5ee..0ec1d9d 100644 --- a/config/samples/deployment.yaml +++ b/config/samples/deployment.yaml @@ -29,7 +29,7 @@ metadata: labels: app.kubernetes.io/managed-by: nebula spec: - replicas: 30 + replicas: 3 selector: matchLabels: app: gpu-workload-sample diff --git a/pkg/vnode/kubelet.go b/pkg/vnode/kubelet.go index ebec767..6e3fb7f 100644 --- a/pkg/vnode/kubelet.go +++ b/pkg/vnode/kubelet.go @@ -38,6 +38,7 @@ import ( "github.com/virtual-kubelet/virtual-kubelet/errdefs" vkapi "github.com/virtual-kubelet/virtual-kubelet/node/api" corev1 "k8s.io/api/core/v1" + "k8s.io/client-go/kubernetes" logf "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/manager" ) @@ -48,7 +49,9 @@ import ( const DefaultKubeletAddr = ":10250" // certValidity is the lifetime of the self-signed serving cert. Rotation is a process -// restart — the key never leaves memory, so there is no Secret to keep in step. +// restart — the key never leaves memory, so there is no Secret to keep in step. A +// signer-issued cert is different: the cluster picks its lifetime, so it is renewed in +// place (see rotateServingCert). const certValidity = 365 * 24 * time.Hour // kubeletShutdownGrace is how long a `kubectl logs -f` stream may finish after shutdown @@ -81,10 +84,10 @@ const ( // is resolved by asking each registered Handler whether it tracks that Pod — at most one // can. Cheaper than a port per provider, and than reading the Pod to learn its node. // -// TLS uses a self-signed in-memory cert, which is what the API server expects: it does -// not verify a kubelet's serving cert unless --kubelet-certificate-authority is set. The -// webhook cert rotator cannot help, since it mints for a Service DNS name and this -// endpoint is dialed by Pod IP. +// TLS defaults to a self-signed in-memory cert, which is enough for an API server that +// does not set --kubelet-certificate-authority. Where it does, EnableServingCSR asks the +// cluster's kubelet-serving signer instead. The webhook cert rotator cannot help either +// way: it mints for a Service DNS name and this endpoint is dialed by Pod IP. // // Client certs are verified only when ClientCAPath is set. Off by default because which CA // signs the API server's kubelet client cert is not portable (kubeadm uses the cluster CA, @@ -105,6 +108,18 @@ type KubeletServer struct { // others are refused at the TLS layer. Empty disables verification — see above. clientCAPath string + // csrClient, when set by EnableServingCSR, asks the cluster's kubelet-serving signer + // for the serving cert instead of self-signing it. nil keeps the self-signed default. + csrClient kubernetes.Interface + csrNodeName string + // csrTimeout overrides csrIssueTimeout, so tests need not wait out the real one. + csrTimeout time.Duration + + // certMu guards the served cert, which rotation swaps under live connections. + certMu sync.RWMutex + cert *tls.Certificate + certIssued bool + mu sync.RWMutex handlers map[string]*Handler } @@ -170,10 +185,13 @@ func (s *KubeletServer) daemonEndpoints() corev1.NodeDaemonEndpoints { func (s *KubeletServer) Start(ctx context.Context) error { log := logf.FromContext(ctx).WithName("kubelet-api") - tlsCfg, err := s.tlsConfig() + tlsCfg, err := s.tlsConfig(ctx) if err != nil { return err } + if s.csrClient != nil { + go s.rotateServingCert(ctx) + } mux := http.NewServeMux() // Logs and exec are wired; the nil funcs make VK answer NotImplemented on @@ -211,8 +229,11 @@ func (s *KubeletServer) Start(ctx context.Context) error { } }() + // signerIssued is the field to read when logs fail with x509 errors: false means the + // fallback cert, which only a non-verifying API server accepts. log.Info("serving kubelet api (container logs, exec)", - "addr", s.addr, "advertisedIP", s.nodeIP, "clientCertRequired", s.clientCAPath != "") + "addr", s.addr, "advertisedIP", s.nodeIP, "clientCertRequired", s.clientCAPath != "", + "signerIssued", s.signerIssued()) // The cert and key are already in TLSConfig, hence the empty paths. if err := srv.ServeTLS(ln, "", ""); err != nil && !errors.Is(err, http.ErrServerClosed) { return fmt.Errorf("kubelet api: serve: %w", err) @@ -260,15 +281,20 @@ func (s *KubeletServer) runInContainer( return h.RunInContainer(ctx, namespace, podName, containerName, cmd, attach) } -// tlsConfig: a fresh self-signed keypair, plus client verification if a CA is set. -func (s *KubeletServer) tlsConfig() (*tls.Config, error) { - cert, err := selfSignedCert(s.nodeIP) +// tlsConfig mints the first serving cert — signer-issued or self-signed, see servingCert — +// and adds client verification if a CA is set. +// +// GetCertificate rather than Certificates, so rotation is a field swap that new handshakes +// pick up without rebuilding the server. +func (s *KubeletServer) tlsConfig(ctx context.Context) (*tls.Config, error) { + cert, issued, err := s.servingCert(ctx) if err != nil { return nil, err } + s.setCert(cert, issued) cfg := &tls.Config{ - Certificates: []tls.Certificate{cert}, - MinVersion: tls.VersionTLS12, + GetCertificate: func(*tls.ClientHelloInfo) (*tls.Certificate, error) { return s.currentCert(), nil }, + MinVersion: tls.VersionTLS12, // http/1.1 only, like a real kubelet: logs need nothing HTTP/2 offers, and this is // the streaming path every kubelet client already exercises. NextProtos: []string{"http/1.1"}, diff --git a/pkg/vnode/kubelet_test.go b/pkg/vnode/kubelet_test.go index 4f099e1..d02dbad 100644 --- a/pkg/vnode/kubelet_test.go +++ b/pkg/vnode/kubelet_test.go @@ -301,7 +301,7 @@ func TestKubeletServer_ClientCAErrorsAreFatal(t *testing.T) { if err != nil { t.Fatalf("NewKubeletServer: %v", err) } - if _, err := s.tlsConfig(); err == nil { + if _, err := s.tlsConfig(context.Background()); err == nil { t.Fatal("expected an error for a client CA path that does not exist") } }) @@ -315,7 +315,7 @@ func TestKubeletServer_ClientCAErrorsAreFatal(t *testing.T) { if err != nil { t.Fatalf("NewKubeletServer: %v", err) } - if _, err := s.tlsConfig(); err == nil { + if _, err := s.tlsConfig(context.Background()); err == nil { t.Fatal("expected an error for a client CA file containing no certificates") } }) @@ -327,7 +327,7 @@ func TestKubeletServer_ClientCAErrorsAreFatal(t *testing.T) { if err != nil { t.Fatalf("NewKubeletServer: %v", err) } - cfg, err := s.tlsConfig() + cfg, err := s.tlsConfig(context.Background()) if err != nil { t.Fatalf("tlsConfig: %v", err) } diff --git a/pkg/vnode/servingcert.go b/pkg/vnode/servingcert.go new file mode 100644 index 0000000..a5ddfd7 --- /dev/null +++ b/pkg/vnode/servingcert.go @@ -0,0 +1,282 @@ +/* +Copyright 2026 The InftyAI Team. + +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 vnode + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "fmt" + "net" + "time" + + certificatesv1 "k8s.io/api/certificates/v1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/client-go/kubernetes" + logf "sigs.k8s.io/controller-runtime/pkg/log" +) + +const ( + // csrIssueTimeout bounds the whole request → approve → issue round trip. Short on + // purpose: an approved CSR that is never signed is the normal outcome on a control + // plane whose kubelet-serving signer is off, and waiting longer only delays the + // fallback. + csrIssueTimeout = 30 * time.Second + // csrPollInterval is how often the pending CSR is re-read while waiting. + csrPollInterval = time.Second + // csrRetryInterval is how long to wait before asking again after a failed request, + // i.e. while the fallback cert is being served. Long enough not to hammer the API + // server, short enough that granting the missing RBAC takes effect without a restart. + csrRetryInterval = 10 * time.Minute + // csrRenewFloor keeps the rotation loop from spinning on a cert that is already close + // to expiry, or expired. + csrRenewFloor = time.Minute +) + +// EnableServingCSR makes the endpoint ask the cluster's kubelet-serving signer for its +// serving cert instead of self-signing it. Needed on an API server started with +// --kubelet-certificate-authority, which rejects a self-signed kubelet cert; pointless +// elsewhere, and the approval it needs is privileged, hence opt-in. +// +// nodeName only has to be one of our virtual node names: the signer requires the subject +// to be system:node:, while the API server verifies the SAN and not the CN, so one +// cert serves every node this process hosts. Call before Start. +func (s *KubeletServer) EnableServingCSR(cs kubernetes.Interface, nodeName string) { + s.csrClient = cs + s.csrNodeName = nodeName +} + +// servingCert is what the endpoint presents: signer-issued when that is enabled and works, +// self-signed otherwise. The bool reports which, since it decides how soon the rotation +// loop tries again. An error means even self-signing failed. +func (s *KubeletServer) servingCert(ctx context.Context) (tls.Certificate, bool, error) { + if s.csrClient == nil { + cert, err := selfSignedCert(s.nodeIP) + return cert, false, err + } + cert, err := s.requestServingCert(ctx) + if err == nil { + return cert, true, nil + } + // Degrade, never fail: a node that serves no logs still runs workloads, and self-signed + // is the working configuration on every cluster that does not verify. + logf.FromContext(ctx).Error(err, "no signer-issued serving cert; falling back to self-signed, "+ + "so `kubectl logs` and `kubectl exec` will fail if the API server sets "+ + "--kubelet-certificate-authority") + cert, err = selfSignedCert(s.nodeIP) + return cert, false, err +} + +// requestServingCert obtains a serving cert from the cluster's kubelet-serving signer, so +// the API server accepts this endpoint the way it accepts a real kubelet's. There is no +// alternative for a verifying API server: Node has no caBundle field, unlike APIService +// and the webhook configurations, so our own CA cannot be published anywhere it would be +// trusted. +// +// The private key is generated here and never leaves this process; only the CSR is written +// to the API. +func (s *KubeletServer) requestServingCert(ctx context.Context) (tls.Certificate, error) { + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + return tls.Certificate{}, fmt.Errorf("kubelet serving cert: generate key: %w", err) + } + // The signer enforces this subject. The SANs mirror selfSignedCert: the advertised IP + // because that is what the API server dials, loopback for curl'ing inside the pod. + csrDER, err := x509.CreateCertificateRequest(rand.Reader, &x509.CertificateRequest{ + Subject: pkix.Name{ + CommonName: "system:node:" + s.csrNodeName, + Organization: []string{"system:nodes"}, + }, + IPAddresses: []net.IP{net.ParseIP(s.nodeIP), net.IPv4(127, 0, 0, 1)}, + DNSNames: []string{"localhost"}, + }, key) + if err != nil { + return tls.Certificate{}, fmt.Errorf("kubelet serving cert: create request: %w", err) + } + + csrs := s.csrClient.CertificatesV1().CertificateSigningRequests() + name := s.csrName() + // Delete first: the name is deterministic, so a restart finds the CSR of the PREVIOUS + // pod IP, and the API rejects reusing a name with different content. + if err := csrs.Delete(ctx, name, metav1.DeleteOptions{}); err != nil && !apierrors.IsNotFound(err) { + return tls.Certificate{}, fmt.Errorf("kubelet serving cert: delete stale CSR %s: %w", name, err) + } + + created, err := csrs.Create(ctx, &certificatesv1.CertificateSigningRequest{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Spec: certificatesv1.CertificateSigningRequestSpec{ + Request: pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE REQUEST", Bytes: csrDER}), + SignerName: certificatesv1.KubeletServingSignerName, + // No key encipherment: it means nothing for an ECDSA key, and the signer accepts + // this pair. + Usages: []certificatesv1.KeyUsage{ + certificatesv1.UsageDigitalSignature, + certificatesv1.UsageServerAuth, + }, + }, + }, metav1.CreateOptions{}) + if err != nil { + return tls.Certificate{}, fmt.Errorf("kubelet serving cert: create CSR %s: %w", name, err) + } + + // Self-approve. kube-controller-manager auto-approves node CLIENT certs only, never + // serving ones, because an approver cannot verify that a requester owns the SANs it + // asks for — which is exactly what we are asserting about ourselves here, and why the + // RBAC for it ships separately. + created.Status.Conditions = append(created.Status.Conditions, certificatesv1.CertificateSigningRequestCondition{ + Type: certificatesv1.CertificateApproved, + Status: corev1.ConditionTrue, + Reason: "NebulaSelfApproved", + Message: "requested by the Nebula manager for its own kubelet endpoint", + }) + if _, err := csrs.UpdateApproval(ctx, name, created, metav1.UpdateOptions{}); err != nil { + return tls.Certificate{}, fmt.Errorf("kubelet serving cert: approve CSR %s: %w", name, err) + } + + // Polling, not a watch: one object, seconds of waiting, and a watch here would need its + // own reconnect handling to be no more reliable. + var issued []byte + err = wait.PollUntilContextTimeout(ctx, csrPollInterval, s.issueTimeout(), true, + func(ctx context.Context) (bool, error) { + cur, err := csrs.Get(ctx, name, metav1.GetOptions{}) + if err != nil { + return false, err + } + for _, c := range cur.Status.Conditions { + if c.Status != corev1.ConditionTrue { + continue + } + switch c.Type { + case certificatesv1.CertificateDenied: + return false, fmt.Errorf("denied: %s", c.Message) + case certificatesv1.CertificateFailed: + return false, fmt.Errorf("failed: %s", c.Message) + } + } + issued = cur.Status.Certificate + return len(issued) > 0, nil + }) + if err != nil { + // Approved but never signed is the signature of a control plane whose + // kubelet-serving signer is disabled — common on managed offerings. + return tls.Certificate{}, fmt.Errorf("kubelet serving cert: CSR %s not issued: %w", name, err) + } + + keyDER, err := x509.MarshalECPrivateKey(key) + if err != nil { + return tls.Certificate{}, fmt.Errorf("kubelet serving cert: marshal key: %w", err) + } + cert, err := tls.X509KeyPair(issued, pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER})) + if err != nil { + return tls.Certificate{}, fmt.Errorf("kubelet serving cert: load issued cert: %w", err) + } + // Leaf is what the rotation loop reads NotAfter from: the signer decides the lifetime + // (--cluster-signing-duration), not us. + if cert.Leaf == nil { + if cert.Leaf, err = x509.ParseCertificate(cert.Certificate[0]); err != nil { + return tls.Certificate{}, fmt.Errorf("kubelet serving cert: parse issued cert: %w", err) + } + } + return cert, nil +} + +// rotateServingCert replaces the cert before it expires, and keeps trying while the +// fallback is in use. Runs until ctx is cancelled. +// +// It exists because the signer's lifetime is the cluster's to choose: a control plane with +// a short --cluster-signing-duration would otherwise lose logs mid-run, silently, hours +// after a start that looked fine. +func (s *KubeletServer) rotateServingCert(ctx context.Context) { + log := logf.FromContext(ctx) + for { + delay := s.renewAfter() + log.V(1).Info("kubelet serving cert renewal scheduled", "in", delay.String()) + select { + case <-ctx.Done(): + return + case <-time.After(delay): + } + cert, issued, err := s.servingCert(ctx) + if err != nil { + // Keep the old cert: an expiring cert still serves, a missing one serves nothing. + log.Error(err, "kubelet serving cert renewal failed; keeping the current cert") + continue + } + s.setCert(cert, issued) + log.Info("kubelet serving cert renewed", "signerIssued", issued) + } +} + +// renewAfter is how long to wait before asking again: two thirds of the current cert's +// remaining life when the signer issued it, a fixed retry while on the fallback. +func (s *KubeletServer) renewAfter() time.Duration { + s.certMu.RLock() + cert, issued := s.cert, s.certIssued + s.certMu.RUnlock() + + if !issued || cert == nil || cert.Leaf == nil { + return csrRetryInterval + } + if d := time.Until(cert.Leaf.NotAfter) * 2 / 3; d > csrRenewFloor { + return d + } + return csrRenewFloor +} + +// csrName is deterministic, so a restart replaces its own CSR rather than leaving one +// behind per pod IP. One name for the whole deployment is safe because Start is +// leader-scoped: at most one replica requests, and a new leader replacing the outgoing +// one's CSR is the intended outcome. +func (s *KubeletServer) csrName() string { + return "nebula-kubelet-serving-" + s.csrNodeName +} + +func (s *KubeletServer) issueTimeout() time.Duration { + if s.csrTimeout > 0 { + return s.csrTimeout + } + return csrIssueTimeout +} + +// setCert swaps in the cert served to new connections. Established ones keep the old one, +// which is correct: renewal must not sever a `kubectl logs -f`. +func (s *KubeletServer) setCert(cert tls.Certificate, issued bool) { + s.certMu.Lock() + defer s.certMu.Unlock() + s.cert, s.certIssued = &cert, issued +} + +func (s *KubeletServer) currentCert() *tls.Certificate { + s.certMu.RLock() + defer s.certMu.RUnlock() + return s.cert +} + +// signerIssued reports whether the served cert came from the cluster signer. +func (s *KubeletServer) signerIssued() bool { + s.certMu.RLock() + defer s.certMu.RUnlock() + return s.certIssued +} diff --git a/pkg/vnode/servingcert_test.go b/pkg/vnode/servingcert_test.go new file mode 100644 index 0000000..abe901e --- /dev/null +++ b/pkg/vnode/servingcert_test.go @@ -0,0 +1,395 @@ +/* +Copyright 2026 The InftyAI Team. + +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 vnode + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "math/big" + "net" + "strings" + "testing" + "time" + + certificatesv1 "k8s.io/api/certificates/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/kubernetes/fake" + k8stesting "k8s.io/client-go/testing" +) + +const ( + testNodeIP = "10.244.1.7" + testNodeName = "nebula-modal" + testCASubj = "test-cluster-signing-ca" +) + +// testCA stands in for the cluster's kubelet-serving signer, so tests can assert on a +// certificate the fake control plane actually issued rather than on a canned blob. +type testCA struct { + cert *x509.Certificate + key *ecdsa.PrivateKey +} + +func newTestCA(t *testing.T) *testCA { + t.Helper() + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatalf("ca key: %v", err) + } + tmpl := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: testCASubj}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(24 * time.Hour), + KeyUsage: x509.KeyUsageCertSign, + BasicConstraintsValid: true, + IsCA: true, + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key) + if err != nil { + t.Fatalf("ca cert: %v", err) + } + cert, err := x509.ParseCertificate(der) + if err != nil { + t.Fatalf("parse ca: %v", err) + } + return &testCA{cert: cert, key: key} +} + +// sign issues a leaf for a CSR, carrying its SANs over the way a real signer does. +func (ca *testCA) sign(t *testing.T, csrPEM []byte, lifetime time.Duration) []byte { + t.Helper() + csr := parseCSR(t, csrPEM) + tmpl := &x509.Certificate{ + SerialNumber: big.NewInt(2), + Subject: csr.Subject, + NotBefore: time.Now().Add(-time.Minute), + NotAfter: time.Now().Add(lifetime), + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + IPAddresses: csr.IPAddresses, + DNSNames: csr.DNSNames, + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, ca.cert, csr.PublicKey, ca.key) + if err != nil { + t.Fatalf("sign: %v", err) + } + return pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}) +} + +func parseCSR(t *testing.T, csrPEM []byte) *x509.CertificateRequest { + t.Helper() + block, _ := pem.Decode(csrPEM) + if block == nil { + t.Fatal("CSR request is not PEM") + } + csr, err := x509.ParseCertificateRequest(block.Bytes) + if err != nil { + t.Fatalf("parse CSR: %v", err) + } + return csr +} + +// issuingClient is a control plane whose kubelet-serving signer works: it signs on +// approval, which is the ordering the real one enforces. +func issuingClient(t *testing.T, ca *testCA, lifetime time.Duration) *fake.Clientset { + t.Helper() + cs := fake.NewSimpleClientset() + cs.PrependReactor("update", "certificatesigningrequests", + func(action k8stesting.Action) (bool, runtime.Object, error) { + csr := approvedCSR(action) + if csr == nil { + return false, nil, nil + } + // Mutate and fall through: the default reactor then persists it, so the next Get + // sees the issued certificate. + csr.Status.Certificate = ca.sign(t, csr.Spec.Request, lifetime) + return false, nil, nil + }) + return cs +} + +// approvedCSR returns the CSR an approval update carries, or nil for any other action. +// The fake control planes react on approval because the real signer only acts after it. +func approvedCSR(action k8stesting.Action) *certificatesv1.CertificateSigningRequest { + if action.GetSubresource() != "approval" { + return nil + } + csr, ok := action.(k8stesting.UpdateAction).GetObject().(*certificatesv1.CertificateSigningRequest) + if !ok { + return nil + } + return csr +} + +func serverWithCSR(t *testing.T, cs *fake.Clientset, timeout time.Duration) *KubeletServer { + t.Helper() + s, err := NewKubeletServer(testNodeIP, freeAddr(t), "") + if err != nil { + t.Fatalf("NewKubeletServer: %v", err) + } + s.EnableServingCSR(cs, testNodeName) + s.csrTimeout = timeout + return s +} + +func storedCSR(t *testing.T, cs *fake.Clientset, name string) *certificatesv1.CertificateSigningRequest { + t.Helper() + csr, err := cs.CertificatesV1().CertificateSigningRequests().Get(context.Background(), name, metav1.GetOptions{}) + if err != nil { + t.Fatalf("get CSR %s: %v", name, err) + } + return csr +} + +// The signer rejects anything but this shape, and the API server verifies the SAN rather +// than the subject — so a wrong CN is rejected at issue time and a missing IP SAN only +// fails later, when logs are dialed. Both are pinned here. +func TestRequestServingCert_RequestShape(t *testing.T) { + ca := newTestCA(t) + cs := issuingClient(t, ca, time.Hour) + s := serverWithCSR(t, cs, 5*time.Second) + + if _, err := s.requestServingCert(context.Background()); err != nil { + t.Fatalf("requestServingCert: %v", err) + } + + stored := storedCSR(t, cs, "nebula-kubelet-serving-"+testNodeName) + if got, want := stored.Spec.SignerName, certificatesv1.KubeletServingSignerName; got != want { + t.Errorf("SignerName = %q, want %q", got, want) + } + wantUsages := []certificatesv1.KeyUsage{certificatesv1.UsageDigitalSignature, certificatesv1.UsageServerAuth} + if got := stored.Spec.Usages; len(got) != len(wantUsages) || got[0] != wantUsages[0] || got[1] != wantUsages[1] { + t.Errorf("Usages = %v, want %v", got, wantUsages) + } + + csr := parseCSR(t, stored.Spec.Request) + if got, want := csr.Subject.CommonName, "system:node:"+testNodeName; got != want { + t.Errorf("CommonName = %q, want %q", got, want) + } + if got := csr.Subject.Organization; len(got) != 1 || got[0] != "system:nodes" { + t.Errorf("Organization = %v, want [system:nodes]", got) + } + if !hasIP(csr.IPAddresses, testNodeIP) { + t.Errorf("IPAddresses = %v, want the advertised IP %s (what the API server dials)", csr.IPAddresses, testNodeIP) + } + if !hasIP(csr.IPAddresses, "127.0.0.1") { + t.Errorf("IPAddresses = %v, want loopback", csr.IPAddresses) + } +} + +func hasIP(ips []net.IP, want string) bool { + for _, ip := range ips { + if ip.Equal(net.ParseIP(want)) { + return true + } + } + return false +} + +// Nothing else will approve a serving CSR — kube-controller-manager approves node client +// certs only — so without this the request would sit pending until the timeout. +func TestRequestServingCert_SelfApproves(t *testing.T) { + ca := newTestCA(t) + cs := issuingClient(t, ca, time.Hour) + s := serverWithCSR(t, cs, 5*time.Second) + + if _, err := s.requestServingCert(context.Background()); err != nil { + t.Fatalf("requestServingCert: %v", err) + } + + stored := storedCSR(t, cs, "nebula-kubelet-serving-"+testNodeName) + var approved bool + for _, c := range stored.Status.Conditions { + if c.Type == certificatesv1.CertificateApproved && c.Status == corev1.ConditionTrue { + approved = true + } + } + if !approved { + t.Fatalf("conditions = %v, want an Approved=True condition", stored.Status.Conditions) + } +} + +// The point of the whole path: the endpoint must present the CA-issued cert, since that is +// the only thing an API server with --kubelet-certificate-authority accepts. +func TestTLSConfig_ServesIssuedCert(t *testing.T) { + ca := newTestCA(t) + cs := issuingClient(t, ca, time.Hour) + s := serverWithCSR(t, cs, 5*time.Second) + + if _, err := s.tlsConfig(context.Background()); err != nil { + t.Fatalf("tlsConfig: %v", err) + } + if !s.signerIssued() { + t.Fatal("signerIssued = false, want true when the signer issued") + } + cert := s.currentCert() + if cert == nil || cert.Leaf == nil { + t.Fatal("no leaf on the served cert; rotation reads NotAfter from it") + } + if got := cert.Leaf.Issuer.CommonName; got != testCASubj { + t.Errorf("issuer = %q, want the signing CA %q", got, testCASubj) + } + if !hasIP(cert.Leaf.IPAddresses, testNodeIP) { + t.Errorf("served cert SANs = %v, want %s", cert.Leaf.IPAddresses, testNodeIP) + } +} + +// A control plane whose kubelet-serving signer is disabled leaves the CSR approved and +// unsigned forever. That must degrade to self-signed, not fail startup: the node still +// runs workloads, and self-signed is what every non-verifying cluster accepts. +func TestTLSConfig_FallsBackWhenNeverIssued(t *testing.T) { + cs := fake.NewSimpleClientset() // no signing reactor + s := serverWithCSR(t, cs, 300*time.Millisecond) + + if _, err := s.tlsConfig(context.Background()); err != nil { + t.Fatalf("tlsConfig: %v", err) + } + if s.signerIssued() { + t.Fatal("signerIssued = true, want false on the fallback") + } + cert := s.currentCert() + if cert == nil { + t.Fatal("no cert served") + } + leaf, err := x509.ParseCertificate(cert.Certificate[0]) + if err != nil { + t.Fatalf("parse: %v", err) + } + if got := leaf.Subject.CommonName; got != "nebula-virtual-kubelet" { + t.Errorf("subject = %q, want the self-signed fallback", got) + } +} + +// Denial is terminal, so it must return at once instead of polling out the timeout — the +// difference between a fast degrade and a stalled startup. +func TestRequestServingCert_DeniedReturnsImmediately(t *testing.T) { + cs := fake.NewSimpleClientset() + cs.PrependReactor("update", "certificatesigningrequests", + func(action k8stesting.Action) (bool, runtime.Object, error) { + csr := approvedCSR(action) + if csr == nil { + return false, nil, nil + } + csr.Status.Conditions = append(csr.Status.Conditions, certificatesv1.CertificateSigningRequestCondition{ + Type: certificatesv1.CertificateDenied, + Status: corev1.ConditionTrue, + Reason: "TestDenied", + Message: "denied by the test", + }) + return false, nil, nil + }) + s := serverWithCSR(t, cs, time.Minute) + + start := time.Now() + _, err := s.requestServingCert(context.Background()) + if err == nil { + t.Fatal("expected an error for a denied CSR") + } + if !strings.Contains(err.Error(), "denied") { + t.Errorf("error = %v, want it to name the denial", err) + } + if elapsed := time.Since(start); elapsed > 10*time.Second { + t.Errorf("took %v; a denied CSR must not be polled until the timeout", elapsed) + } +} + +// The CSR name is deterministic, so a restart finds its own request from the previous pod +// IP. The API rejects reusing a name with different content, so the stale one is deleted. +func TestRequestServingCert_ReplacesStaleCSR(t *testing.T) { + ca := newTestCA(t) + cs := issuingClient(t, ca, time.Hour) + name := "nebula-kubelet-serving-" + testNodeName + if _, err := cs.CertificatesV1().CertificateSigningRequests().Create(context.Background(), + &certificatesv1.CertificateSigningRequest{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Spec: certificatesv1.CertificateSigningRequestSpec{ + Request: []byte("stale, for an IP this pod no longer has"), + SignerName: certificatesv1.KubeletServingSignerName, + }, + }, metav1.CreateOptions{}); err != nil { + t.Fatalf("seed stale CSR: %v", err) + } + + s := serverWithCSR(t, cs, 5*time.Second) + if _, err := s.requestServingCert(context.Background()); err != nil { + t.Fatalf("requestServingCert: %v", err) + } + + // The stored request must be ours, not the seeded junk. + csr := parseCSR(t, storedCSR(t, cs, name).Spec.Request) + if !hasIP(csr.IPAddresses, testNodeIP) { + t.Errorf("stored CSR SANs = %v, want the current pod IP %s", csr.IPAddresses, testNodeIP) + } + var deleted bool + for _, a := range cs.Actions() { + if a.GetVerb() == "delete" && a.GetResource().Resource == "certificatesigningrequests" { + deleted = true + } + } + if !deleted { + t.Error("no delete recorded; a stale CSR would make Create fail with AlreadyExists") + } +} + +// Renewal timing is the difference between rotating quietly and losing logs mid-run on a +// cluster with a short --cluster-signing-duration. +func TestRenewAfter(t *testing.T) { + s := &KubeletServer{} + + t.Run("fallback retries soon", func(t *testing.T) { + s.setCert(mustSelfSigned(t), false) + if got := s.renewAfter(); got != csrRetryInterval { + t.Errorf("renewAfter = %v, want the retry interval %v while self-signed", got, csrRetryInterval) + } + }) + + t.Run("two thirds of the remaining life", func(t *testing.T) { + cert := mustSelfSigned(t) + cert.Leaf = &x509.Certificate{NotAfter: time.Now().Add(3 * time.Hour)} + s.setCert(cert, true) + if got := s.renewAfter(); got < 110*time.Minute || got > 2*time.Hour { + t.Errorf("renewAfter = %v, want ~2h for a 3h cert", got) + } + }) + + t.Run("expired does not spin", func(t *testing.T) { + cert := mustSelfSigned(t) + cert.Leaf = &x509.Certificate{NotAfter: time.Now().Add(-time.Hour)} + s.setCert(cert, true) + if got := s.renewAfter(); got != csrRenewFloor { + t.Errorf("renewAfter = %v, want the floor %v", got, csrRenewFloor) + } + }) +} + +func mustSelfSigned(t *testing.T) tls.Certificate { + t.Helper() + cert, err := selfSignedCert(testNodeIP) + if err != nil { + t.Fatalf("selfSignedCert: %v", err) + } + return cert +}