From 185bb8750eb2c4e8e5f1c605c4c616fc18abaaa4 Mon Sep 17 00:00:00 2001 From: Yang Le Date: Fri, 3 Apr 2020 18:06:52 +0800 Subject: [PATCH] Add agent bootstrap & cert-rotation --- go.mod | 3 + pkg/cmd/spoke/agent.go | 4 +- pkg/spoke/agent.go | 58 +++- pkg/spoke/bootstrap/bootstrap.go | 423 ++++++++++++++++++++++++++++ pkg/spoke/bootstrap/certificate.go | 354 +++++++++++++++++++++++ pkg/spoke/bootstrap/clientconfig.go | 237 ++++++++++++++++ pkg/spoke/bootstrap/csr.go | 222 +++++++++++++++ pkg/spoke/bootstrap/options.go | 60 ++++ pkg/spoke/bootstrap/transport.go | 140 +++++++++ 9 files changed, 1499 insertions(+), 2 deletions(-) create mode 100644 pkg/spoke/bootstrap/bootstrap.go create mode 100644 pkg/spoke/bootstrap/certificate.go create mode 100644 pkg/spoke/bootstrap/clientconfig.go create mode 100644 pkg/spoke/bootstrap/csr.go create mode 100644 pkg/spoke/bootstrap/options.go create mode 100644 pkg/spoke/bootstrap/transport.go diff --git a/go.mod b/go.mod index 9cb7905fc..af1878296 100644 --- a/go.mod +++ b/go.mod @@ -7,6 +7,9 @@ require ( github.com/openshift/library-go v0.0.0-20200401114229-ffab8c6e83a9 github.com/spf13/cobra v0.0.5 github.com/spf13/pflag v1.0.5 + k8s.io/api v0.18.0 k8s.io/apimachinery v0.18.0 + k8s.io/client-go v0.18.0 k8s.io/component-base v0.18.0 + k8s.io/klog v1.0.0 ) diff --git a/pkg/cmd/spoke/agent.go b/pkg/cmd/spoke/agent.go index 34a88aa80..0d0e5574b 100644 --- a/pkg/cmd/spoke/agent.go +++ b/pkg/cmd/spoke/agent.go @@ -10,11 +10,13 @@ import ( ) func NewAgent() *cobra.Command { + agent := spoke.NewAgent() cmd := controllercmd. - NewControllerCommandConfig("agent", version.Get(), spoke.RunAgent). + NewControllerCommandConfig("agent", version.Get(), agent.RunAgent). NewCommand() cmd.Use = "agent" cmd.Short = "Start the Cluster Registration Agent" + agent.AddFlags(cmd.Flags()) return cmd } diff --git a/pkg/spoke/agent.go b/pkg/spoke/agent.go index 1cd0182c4..9b1fedc88 100644 --- a/pkg/spoke/agent.go +++ b/pkg/spoke/agent.go @@ -3,11 +3,67 @@ package spoke import ( "context" + "github.com/open-cluster-management/registration/pkg/spoke/bootstrap" "github.com/openshift/library-go/pkg/controller/controllercmd" + "github.com/spf13/pflag" + "k8s.io/client-go/kubernetes" + "k8s.io/klog" ) +// Agent holds configuration for spoke cluster agent +type Agent struct { + bootstrapOptions *bootstrap.Options +} + +// NewAgent returns an Agent +func NewAgent() *Agent { + return &Agent{ + bootstrapOptions: bootstrap.NewOptions(), + } +} + // RunAgent starts the controllers on agent to register to hub. -func RunAgent(ctx context.Context, controllerContext *controllercmd.ControllerContext) error { +func (a *Agent) RunAgent(ctx context.Context, controllerContext *controllercmd.ControllerContext) error { + if err := a.bootstrapOptions.Validate(); err != nil { + klog.Fatal(err) + } + + if err := a.bootstrapOptions.Complete(); err != nil { + klog.Fatal(err) + } + + kubeClient, err := kubernetes.NewForConfig(controllerContext.KubeConfig) + if err != nil { + return err + } + + // bootstrap the agent and register the spoke cluster + _, _, err = bootstrap.Bootstrap(kubeClient.CoreV1(), a.bootstrapOptions, ctx.Done()) + if err != nil { + return err + } + <-ctx.Done() return nil } + +// AddFlags registers flags for Agent +func (a *Agent) AddFlags(fs *pflag.FlagSet) { + a.bootstrapOptions.AddFlags(fs) +} + +// Validate verifies the inputs. +func (a *Agent) Validate() error { + if err := a.bootstrapOptions.Validate(); err != nil { + return err + } + return nil +} + +// Complete fills in missing values. +func (a *Agent) Complete() error { + if err := a.bootstrapOptions.Complete(); err != nil { + return err + } + return nil +} diff --git a/pkg/spoke/bootstrap/bootstrap.go b/pkg/spoke/bootstrap/bootstrap.go new file mode 100644 index 000000000..dec269d92 --- /dev/null +++ b/pkg/spoke/bootstrap/bootstrap.go @@ -0,0 +1,423 @@ +package bootstrap + +import ( + "context" + "errors" + "fmt" + "reflect" + "strings" + "time" + + corev1 "k8s.io/api/core/v1" + kerrors "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/scheme" + certificatesv1beta1 "k8s.io/client-go/kubernetes/typed/certificates/v1beta1" + corev1client "k8s.io/client-go/kubernetes/typed/core/v1" + restclient "k8s.io/client-go/rest" + "k8s.io/client-go/tools/cache" + "k8s.io/client-go/tools/clientcmd" + "k8s.io/client-go/util/keyutil" + "k8s.io/klog" +) + +const ( + certPairNamePrefix = "spoke-cluster-client" + kubeconfigSecretDataKey = "kubeconfig" + agentNameSecretDataKey = "agent-name" +) + +var ( + splitMetaNamespaceKey = cache.SplitMetaNamespaceKey +) + +// Bootstrap registers the spoke cluster and bootstraps the agent +func Bootstrap(coreClient corev1client.CoreV1Interface, o *Options, stopCh <-chan struct{}) (*restclient.Config, func(), error) { + agentName, bootstrapped, err := recoverAgentState(coreClient, o) + if err != nil { + return nil, nil, err + } + + if !bootstrapped { + if agentName == "" { + agentName, err = generateAgentName("") + if err != nil { + return nil, nil, fmt.Errorf("unable to generate agent name: %v", err) + } + klog.V(4).Infof("Agent name is generated: %s", agentName) + err = writeAgentName(agentName, o.CertStoreSecret, coreClient) + if err != nil { + return nil, nil, err + } + } + + err = bootstrapAgent(agentName, coreClient, o) + if err != nil { + return nil, nil, fmt.Errorf("unable to bootstrap agent: %v", err) + } + } + + clientConfig, err := loadRESTClientConfig(o.HubKubeconfigSecret, coreClient) + if err != nil { + return nil, nil, fmt.Errorf("invalid kubeconfig: %v", err) + } + + clusterName, err := getClusterName(agentName) + if err != nil { + return nil, nil, err + } + + // block until the spoke cluster on hub is approved + err = waitForSpokeClusterApproval(clientConfig, clusterName, stopCh) + if err != nil { + return nil, nil, err + } + + return rotateCertificates(agentName, clientConfig, coreClient, o, stopCh) +} + +// recoverAgentState recovers the current state of the cluster agent +func recoverAgentState(coreClient corev1client.CoreV1Interface, o *Options) (string, bool, error) { + agentName, err := resolveAgentName(o.CertStoreSecret, o.ClusterNameOverride, coreClient) + if err != nil { + return "", false, fmt.Errorf("unable to resolve agent name: %v", err) + } + if agentName != "" { + klog.V(4).Infof("Agent name is resolved: %s", agentName) + } + + kubeconfigData, exists, err := loadKubeconfig(o.HubKubeconfigSecret, coreClient) + if err != nil { + return agentName, false, fmt.Errorf("unable to load kubeconfig from secret %q: %v", o.HubKubeconfigSecret, err) + } + + if !exists { + klog.Info("No kubeconfig is found, bootstrap is required") + return agentName, false, nil + } + + ok, err := isClientConfigStillValid(kubeconfigData, agentName) + if err != nil { + return agentName, false, err + } + if !ok { + klog.Info("Kubeconfig is no long valid, bootstrap is required") + return agentName, false, nil + } + + klog.Info("Kubeconfig exists and is valid, skipping bootstrap") + if agentName == "" { + agentName, err = getAgentNameFromKubeconfig(kubeconfigData) + if err != nil { + return agentName, false, fmt.Errorf("unable to get agent name from kubeconfig: %v", err) + } + klog.V(4).Infof("Agent name is detected in certification from kubeconfig: %s", agentName) + err = writeAgentName(agentName, o.CertStoreSecret, coreClient) + if err != nil { + return agentName, false, err + } + } + + // register spoke cluster in case it does not exists + clientConfig, err := clientcmd.RESTConfigFromKubeConfig(kubeconfigData) + if err != nil { + return agentName, false, err + } + clusterName, err := getClusterName(agentName) + if err != nil { + return agentName, false, err + } + err = registerSpokeCluster(clientConfig, clusterName) + if err != nil { + return agentName, false, fmt.Errorf("unable to register spoke cluster %q: %v", clusterName, err) + } + + return agentName, true, nil +} + +// bootstrapAgent bootstraps cluster agent with the bootstrap kubeconfig +func bootstrapAgent(agentName string, coreClient corev1client.CoreV1Interface, o *Options) error { + klog.Info("Start bootstrapping") + store, err := NewSecretStore(o.CertStoreSecret, certPairNamePrefix, coreClient, nil, nil, nil) + if err != nil { + return fmt.Errorf("unable to create cert store") + } + + bootstrapClientConfig, err := loadRESTClientConfig(o.BootstrapKubeconfigSecret, coreClient) + if err != nil { + return fmt.Errorf("unable to load kubeconfig from secret %q: %v", o.BootstrapKubeconfigSecret, err) + } + + clusterName, err := getClusterName(agentName) + if err != nil { + return err + } + // register spoke cluster + err = registerSpokeCluster(bootstrapClientConfig, clusterName) + if err != nil { + return fmt.Errorf("unable to register spoke cluster %q: %v", clusterName, err) + } + + bootstrapClient, err := certificatesv1beta1.NewForConfig(bootstrapClientConfig) + if err != nil { + return fmt.Errorf("unable to create certificates signing request client: %v", err) + } + + var keyData []byte + if cert, err := store.Current(); err == nil { + if cert.PrivateKey != nil { + keyData, err = keyutil.MarshalPrivateKeyToPEM(cert.PrivateKey) + if err != nil { + keyData = nil + } + } + } + + // Cache the private key in cert store until CSR succeeds. + if !verifyKeyData(keyData) { + // Note: always call GetOrGenerateTmpPrivateKey so that private key is + // reused on next startup if CSR request fails. + keyData, err = store.GetOrGenerateTmpPrivateKey() + if err != nil { + return err + } + } + + if err := waitForServer(*bootstrapClientConfig, 1*time.Minute); err != nil { + klog.Errorf("Error waiting for apiserver to come up: %v", err) + } + + certData, err := requestClusterCertificate(bootstrapClient.CertificateSigningRequests(), keyData, clusterName, agentName) + if err != nil { + return err + } + klog.V(4).Info("Client certificate issued") + + if _, err := store.Update(certData, keyData); err != nil { + return err + } + if err := store.RemoveTmpPrivateKey(); err != nil { + klog.V(4).Infof("Failed cleaning up private key in cert store: %v", err) + } + + err = writeKubeconfig(bootstrapClientConfig, certData, keyData, o.HubKubeconfigSecret, coreClient) + if err != nil { + return err + } + + klog.Info("Bootstrap done") + return nil +} + +// rotateCertificates rotates certificates before client certificate becomes expired. +func rotateCertificates(agentName string, clientConfig *restclient.Config, coreClient corev1client.CoreV1Interface, o *Options, stopCh <-chan struct{}) (*restclient.Config, func(), error) { + store, err := NewSecretStore(o.CertStoreSecret, certPairNamePrefix, coreClient, clientConfig.CertData, clientConfig.KeyData, + func(certData, keyData []byte) error { + return writeKubeconfig(restclient.AnonymousClientConfig(clientConfig), + certData, keyData, o.HubKubeconfigSecret, coreClient) + }) + if err != nil { + return nil, nil, fmt.Errorf("unable to create cert store") + } + + clientCertificateManager, err := NewManager(clientConfig, agentName, store) + if err != nil { + return nil, nil, fmt.Errorf("unable to create certificate manager") + } + + // the rotating transport will use the cert from the cert manager + transportConfig := restclient.AnonymousClientConfig(clientConfig) + closeAllConns, err := UpdateTransport(wait.NeverStop, transportConfig, clientCertificateManager, 5*time.Minute) + if err != nil { + return nil, nil, err + } + + klog.Info("Starting client certificate rotation.") + clientCertificateManager.Start() + + return transportConfig, closeAllConns, nil +} + +// registerSpokeCluster register a spoke cluster on hub with the given name if it does not exists yet +func registerSpokeCluster(clientConfig *restclient.Config, clusterName string) error { + // TODO register the spoke cluster + return nil +} + +// waitForSpokeClusterApproval waits until the spoke cluster is approved on hub +func waitForSpokeClusterApproval(clientConfig *restclient.Config, clusterName string, stopCh <-chan struct{}) error { + // TODO wait until the spoke cluster is approved on hub + return nil +} + +// verifyKeyData returns true if the provided data appears to be a valid private key. +func verifyKeyData(data []byte) bool { + if len(data) == 0 { + return false + } + _, err := keyutil.ParsePrivateKeyPEM(data) + return err == nil +} + +func waitForServer(cfg restclient.Config, deadline time.Duration) error { + cfg.NegotiatedSerializer = scheme.Codecs.WithoutConversion() + cfg.Timeout = 1 * time.Second + cli, err := restclient.UnversionedRESTClientFor(&cfg) + if err != nil { + return fmt.Errorf("couldn't create client: %v", err) + } + + ctx, cancel := context.WithTimeout(context.TODO(), deadline) + defer cancel() + + var connected bool + wait.JitterUntil(func() { + if _, err := cli.Get().AbsPath("/healthz").Do(context.Background()).Raw(); err != nil { + klog.Errorf("Failed to connect to apiserver: %v", err) + return + } + cancel() + connected = true + }, 2*time.Second, 0.2, true, ctx.Done()) + + if !connected { + return errors.New("timed out waiting to connect to apiserver") + } + return nil +} + +// resolveAgentName resolves the name of the agent. +func resolveAgentName(secretKey, clusterNameOverride string, coreClient corev1client.CoreV1Interface) (string, error) { + agentName, err := getAgentName(secretKey, coreClient) + if err != nil { + return "", err + } + + clusterName, err := getClusterName(agentName) + if err != nil { + return "", err + } + + if clusterNameOverride != "" && clusterNameOverride != clusterName { + agentName, err = generateAgentName(clusterNameOverride) + if err != nil { + return "", err + } + + err := writeAgentName(agentName, secretKey, coreClient) + if err != nil { + return "", err + } + return agentName, nil + } + + return agentName, nil +} + +// generateClusterName generates a random name for cluster or return cluster UID if it's an openshift cluster +func generateClusterName() (string, error) { + // TODO add logic to generate random cluster name + return "cluster0", nil +} + +// generateAgentName generates a random name for cluster agent +func generateAgentName(clusterName string) (string, error) { + // TODO add logic to generate random agent name + if clusterName == "" { + var err error + clusterName, err = generateClusterName() + if err != nil { + return "", err + } + } + return clusterName + ":agent0", nil +} + +// getClusterName returns cluster name by parsing agent name +func getClusterName(agentName string) (string, error) { + if agentName == "" { + return "", nil + } + + names := strings.Split(agentName, ":") + if len(names) != 2 { + return "", fmt.Errorf("invalid agent name %q", agentName) + } + + return names[0], nil +} + +// getAgentName return agent name stored in secret +func getAgentName(secretKey string, coreClient corev1client.CoreV1Interface) (string, error) { + namespace, name, err := splitMetaNamespaceKey(secretKey) + if err != nil { + return "", fmt.Errorf("invalid secret name %q: %v", secretKey, err) + } + + secret, err := coreClient.Secrets(namespace).Get(context.Background(), name, metav1.GetOptions{}) + if err != nil { + if kerrors.IsNotFound(err) { + return "", nil + } else { + return "", fmt.Errorf("unable to get secret %q: %v", secretKey, err) + } + } + + if secret.Data == nil { + return "", nil + } + if value, ok := secret.Data[agentNameSecretDataKey]; ok { + return string(value), nil + } + + return "", nil +} + +// writeAgentName saves agent name in secret +func writeAgentName(agentName, secretKey string, coreClient corev1client.CoreV1Interface) error { + namespace, name, err := splitMetaNamespaceKey(secretKey) + if err != nil { + return fmt.Errorf("invalid secret name %q: %v", secretKey, err) + } + + found := true + secret, err := coreClient.Secrets(namespace).Get(context.Background(), name, metav1.GetOptions{}) + if err != nil { + if kerrors.IsNotFound(err) { + found = false + secret = &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: namespace, + Name: name, + }, + } + } else { + return fmt.Errorf("unable to get secret %q: %v", secretKey, err) + } + } + + if secret.Data == nil { + secret.Data = make(map[string][]byte) + } + agentNameData := []byte(agentName) + if value, ok := secret.Data[agentNameSecretDataKey]; ok { + if reflect.DeepEqual(value, agentNameData) { + return nil + } + } + secret.Data[agentNameSecretDataKey] = agentNameData + + // create/update secret + if found { + _, err = coreClient.Secrets(namespace).Update(context.Background(), secret, metav1.UpdateOptions{}) + + } else { + _, err = coreClient.Secrets(namespace).Create(context.Background(), secret, metav1.CreateOptions{}) + } + if err != nil { + return fmt.Errorf("unable to save agent name in secret %q: %v", secretKey, err) + } + klog.V(4).Infof("Save agent name in secret %q", secretKey) + return nil +} diff --git a/pkg/spoke/bootstrap/certificate.go b/pkg/spoke/bootstrap/certificate.go new file mode 100644 index 000000000..0916d7955 --- /dev/null +++ b/pkg/spoke/bootstrap/certificate.go @@ -0,0 +1,354 @@ +package bootstrap + +import ( + "bytes" + "context" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "errors" + "fmt" + "time" + + certificates "k8s.io/api/certificates/v1beta1" + corev1 "k8s.io/api/core/v1" + kerrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" + restclient "k8s.io/client-go/rest" + "k8s.io/client-go/util/certificate" + "k8s.io/client-go/util/keyutil" + "k8s.io/klog" + + certificatesclient "k8s.io/client-go/kubernetes/typed/certificates/v1beta1" + corev1client "k8s.io/client-go/kubernetes/typed/core/v1" +) + +const ( + keyExtension = ".key" + certExtension = ".crt" + pemExtension = ".pem" + currentPair = "current" + updatedPair = "updated" + + currentAnnotation = "certification.open-cluster-management.io/current" + tmpPrivateKeyFile = "spoke-cluster-client.key.tmp" + subjectPrefix = "system:open-cluster-management:" +) + +// NewManager creates a certificate manager. +func NewManager(clientConfig *restclient.Config, agentName string, store certificate.Store) (certificate.Manager, error) { + clusterName, err := getClusterName(agentName) + if err != nil { + return nil, err + } + + newClientFn := func(current *tls.Certificate) (certificatesclient.CertificateSigningRequestInterface, error) { + client, err := kubernetes.NewForConfig(clientConfig) + if err != nil { + return nil, err + } + return client.CertificatesV1beta1().CertificateSigningRequests(), nil + } + + return certificate.NewManager(&certificate.Config{ + ClientFn: newClientFn, + Template: &x509.CertificateRequest{ + Subject: pkix.Name{ + CommonName: fmt.Sprintf("%s%s", subjectPrefix, agentName), + Organization: []string{fmt.Sprintf("%s%s", subjectPrefix, clusterName)}, + }, + }, + Usages: []certificates.KeyUsage{ + // https://tools.ietf.org/html/rfc5280#section-4.2.1.3 + // + // DigitalSignature allows the certificate to be used to verify + // digital signatures including signatures used during TLS + // negotiation. + certificates.UsageDigitalSignature, + // KeyEncipherment allows the cert/key pair to be used to encrypt + // keys, including the symmetric keys negotiated during TLS setup + // and used for data transfer.. + certificates.UsageKeyEncipherment, + // ClientAuth allows the cert to be used by a TLS client to + // authenticate itself to the TLS server. + certificates.UsageClientAuth, + }, + CertificateStore: store, + }) +} + +// OnCertUpdateFunc is the calllback function which is invoked once certificate in store is updated +type OnCertUpdateFunc func(certData, keyData []byte) error + +// SecretStore extends certificate.Store and supports persistance of temporary private key +type SecretStore interface { + certificate.Store + + GetOrGenerateTmpPrivateKey() ([]byte, error) + RemoveTmpPrivateKey() error +} + +// secretStore is a concrete implementation of a Store that is based on +// storing the cert/key pairs in the designated secret. +type secretStore struct { + secretNamespace string + secretName string + pairNamePrefix string + certData []byte + keyData []byte + coreClient corev1client.CoreV1Interface + + onCertUpdateFunc OnCertUpdateFunc +} + +// NewSecretStore returns a concrete implementation of SecretStore. +func NewSecretStore( + secretKey string, + pairNamePrefix string, + coreClient corev1client.CoreV1Interface, + certData, keyData []byte, + onCertUpdateFunc OnCertUpdateFunc) (SecretStore, error) { + namespace, name, err := splitMetaNamespaceKey(secretKey) + if err != nil { + return nil, fmt.Errorf("invalid secret name %q: %v", secretKey, err) + } + + s := secretStore{ + secretNamespace: namespace, + secretName: name, + pairNamePrefix: pairNamePrefix, + certData: certData, + keyData: keyData, + coreClient: coreClient, + onCertUpdateFunc: onCertUpdateFunc, + } + + return &s, nil +} + +// Current returns the current certificate in the store +func (s *secretStore) Current() (*tls.Certificate, error) { + found := true + secret, err := s.coreClient.Secrets(s.secretNamespace).Get(context.Background(), s.secretName, metav1.GetOptions{}) + if err != nil { + if kerrors.IsNotFound(err) { + found = false + } else { + return nil, fmt.Errorf("unable to get secret %q: %v", s.secretNamespace+"/"+s.secretName, err) + } + } + + if found { + cert, exists, err := getCurrentCertificate(secret) + if err != nil { + return nil, err + } + if exists { + return cert, nil + } + } + + if s.certData != nil && s.keyData != nil { + klog.V(4).Info("Loading cert/key pair from PEM blocks.") + return parseCertFromPEMBlocks(s.certData, s.keyData) + } + + c := s.pairNamePrefix + certExtension + k := s.pairNamePrefix + keyExtension + certData, certExists := secret.Data[c] + keyData, keyExists := secret.Data[k] + if certExists && keyExists { + klog.V(4).Infof("Loading cert/key pair: %s, %s.", c, k) + return parseCertFromPEMBlocks(certData, keyData) + } + + noKeyErr := certificate.NoCertKeyError(fmt.Sprintf("no cert/key available in secret: %s/%s", + s.secretNamespace, s.secretName)) + return nil, &noKeyErr +} + +// getCurrentCertificate returns the current certificate stored in secret +func getCurrentCertificate(secret *corev1.Secret) (*tls.Certificate, bool, error) { + current, ok := secret.Annotations[currentAnnotation] + if !ok { + return nil, false, nil + } + + if pemBlock, exists := secret.Data[current]; exists { + klog.Infof("Loading cert/key pair: %s.", current) + cert, err := parseCertFromPEMBlocks(pemBlock, pemBlock) + return cert, true, err + } + + return nil, false, fmt.Errorf("unable to find PEM Block for current certification: %s", current) +} + +// parseCertFromPEMBlocks parses and returns tls certificate from cert/key pem blocks +func parseCertFromPEMBlocks(certPEMBlock, keyPEMBlock []byte) (*tls.Certificate, error) { + cert, err := tls.X509KeyPair(certPEMBlock, keyPEMBlock) + if err != nil { + return nil, fmt.Errorf("could not convert PEM block into cert/key pair: %v", err) + } + certs, err := x509.ParseCertificates(cert.Certificate[0]) + if err != nil { + return nil, fmt.Errorf("unable to parse certificate data: %v", err) + } + cert.Leaf = certs[0] + return &cert, nil +} + +// Update updates the current certificate in store +func (s *secretStore) Update(certData, keyData []byte) (*tls.Certificate, error) { + found := true + secret, err := s.coreClient.Secrets(s.secretNamespace).Get(context.Background(), s.secretName, metav1.GetOptions{}) + if err != nil { + if kerrors.IsNotFound(err) { + found = false + secret = &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: s.secretNamespace, + Name: s.secretName, + }, + } + } else if err != nil { + return nil, err + } + } + + // build pem block for certificate + var buffer bytes.Buffer + certBlock, _ := pem.Decode(certData) + if certBlock == nil { + return nil, errors.New("invalid certificate data") + } + pem.Encode(&buffer, certBlock) + keyBlock, _ := pem.Decode(keyData) + if keyBlock == nil { + return nil, errors.New("invalid key data") + } + pem.Encode(&buffer, keyBlock) + + pemBlock := buffer.Bytes() + cert, err := parseCertFromPEMBlocks(pemBlock, pemBlock) + if err != nil { + return nil, err + } + + // save the certificate in secret and make the annotation 'current' point to it + ts := time.Now().Format("2006-01-02-15-04-05") + pemBlockKey := s.filename(ts) + if secret.Data == nil { + secret.Data = make(map[string][]byte) + } + secret.Data[pemBlockKey] = pemBlock + + if secret.Annotations == nil { + secret.Annotations = make(map[string]string) + } + secret.Annotations[currentAnnotation] = pemBlockKey + + if found { + // update secret + _, err = s.coreClient.Secrets(s.secretNamespace).Update(context.Background(), secret, metav1.UpdateOptions{}) + if err != nil { + return nil, fmt.Errorf("unable to update certificate in secret %q: %v", s.secretNamespace+"/"+s.secretName, err) + } + } else { + // create seccret + _, err = s.coreClient.Secrets(s.secretNamespace).Create(context.Background(), secret, metav1.CreateOptions{}) + if err != nil { + return nil, fmt.Errorf("unable to create secret %q: %v", s.secretNamespace+"/"+s.secretName, err) + } + } + + // call callback function once cert is updated + if s.onCertUpdateFunc == nil { + return cert, nil + } + + return cert, s.onCertUpdateFunc(certData, keyData) +} + +// GetOrGenerateTmpPrivateKey returns temporary private key if it exists; otherwise +// generates a new private key and returns. +func (s *secretStore) GetOrGenerateTmpPrivateKey() ([]byte, error) { + found := true + secret, err := s.coreClient.Secrets(s.secretNamespace).Get(context.Background(), s.secretName, metav1.GetOptions{}) + if err != nil { + if kerrors.IsNotFound(err) { + found = false + secret = &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: s.secretNamespace, + Name: s.secretName, + }, + } + } else if err != nil { + return nil, err + } + } + + // reuse existing private key if exists + if tmpPrivateKey, ok := secret.Data[tmpPrivateKeyFile]; ok { + klog.V(4).Info("Reuse existing private key") + return tmpPrivateKey, nil + } + + // otherwise, create a new private key + tmpPrivateKey, err := keyutil.MakeEllipticPrivateKeyPEM() + if err != nil { + return nil, err + } + + // save private key in secret + if secret.Data == nil { + secret.Data = make(map[string][]byte) + } + secret.Data[tmpPrivateKeyFile] = tmpPrivateKey + + if found { + // update secret + _, err = s.coreClient.Secrets(s.secretNamespace).Update(context.Background(), secret, metav1.UpdateOptions{}) + if err != nil { + return nil, fmt.Errorf("unable to save temporary private key in secret %q: %v", s.secretNamespace+"/"+s.secretName, err) + } + } else { + // create seccret + _, err = s.coreClient.Secrets(s.secretNamespace).Create(context.Background(), secret, metav1.CreateOptions{}) + if err != nil { + return nil, fmt.Errorf("unable to create secret %q: %v", s.secretNamespace+"/"+s.secretName, err) + } + } + + klog.V(4).Info("Create a new private key") + return tmpPrivateKey, err +} + +// RemoveTmpPrivateKey remove the temporary private key from store if it exists +func (s *secretStore) RemoveTmpPrivateKey() error { + secret, err := s.coreClient.Secrets(s.secretNamespace).Get(context.Background(), s.secretName, metav1.GetOptions{}) + if err != nil { + if kerrors.IsNotFound(err) { + return nil + } + return fmt.Errorf("unable to get secret %q: %v", s.secretNamespace+"/"+s.secretName, err) + } + + // remove the temporary private key if exists + if _, ok := secret.Data[tmpPrivateKeyFile]; ok { + delete(secret.Data, tmpPrivateKeyFile) + _, err = s.coreClient.Secrets(s.secretNamespace).Update(context.Background(), secret, metav1.UpdateOptions{}) + if err != nil { + return fmt.Errorf("unable to remove temporary private key from secret %q: %v", s.secretNamespace+"/"+s.secretName, err) + } + } + + return nil +} + +// filename returns the file name of certificate as data key in secret +func (s *secretStore) filename(qualifier string) string { + return s.pairNamePrefix + "-" + qualifier + pemExtension +} diff --git a/pkg/spoke/bootstrap/clientconfig.go b/pkg/spoke/bootstrap/clientconfig.go new file mode 100644 index 000000000..923d66570 --- /dev/null +++ b/pkg/spoke/bootstrap/clientconfig.go @@ -0,0 +1,237 @@ +package bootstrap + +import ( + "context" + "crypto/x509" + "errors" + "fmt" + "reflect" + "strings" + "time" + + corev1 "k8s.io/api/core/v1" + kerrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + utilruntime "k8s.io/apimachinery/pkg/util/runtime" + corev1client "k8s.io/client-go/kubernetes/typed/core/v1" + restclient "k8s.io/client-go/rest" + "k8s.io/client-go/tools/clientcmd" + clientcmdapi "k8s.io/client-go/tools/clientcmd/api" + "k8s.io/client-go/transport" + certutil "k8s.io/client-go/util/cert" + "k8s.io/klog" +) + +// writeKubeconfig writes client config as kubeconfig into the given secret +func writeKubeconfig(clientConfig *restclient.Config, certData, keyData []byte, kubeconfigSecret string, coreClient corev1client.CoreV1Interface) error { + namespace, name, err := splitMetaNamespaceKey(kubeconfigSecret) + if err != nil { + return fmt.Errorf("invalid secret name: %s", kubeconfigSecret) + } + + kubeconfigData, err := buildKubeconfig(clientConfig, certData, keyData) + if err != nil { + return err + } + + found := true + secret, err := coreClient.Secrets(namespace).Get(context.Background(), name, metav1.GetOptions{}) + if err != nil { + if kerrors.IsNotFound(err) { + found = false + secret = &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: namespace, + Name: name, + }, + } + } else { + return err + } + } + + if secret.Data == nil { + secret.Data = make(map[string][]byte) + } + + // do nothing if kubeconfigData is not changed at all + if data, exists := secret.Data[kubeconfigSecretDataKey]; exists { + if reflect.DeepEqual(data, kubeconfigData) { + return nil + } + } + + secret.Data[kubeconfigSecretDataKey] = kubeconfigData + + if found { + // update secret + _, err = coreClient.Secrets(namespace).Update(context.Background(), secret, metav1.UpdateOptions{}) + if err != nil { + return fmt.Errorf("unable to write kubeconfig into secret %q: %v", kubeconfigSecret, err) + } + } else { + // create seccret + _, err = coreClient.Secrets(namespace).Create(context.Background(), secret, metav1.CreateOptions{}) + if err != nil { + return fmt.Errorf("unable to create secret %q: %v", kubeconfigSecret, err) + } + } + + klog.V(4).Infof("write kubeconfig into secret %q", kubeconfigSecret) + return nil +} + +// buildKubeconfig builds kubeconfig based on rest config and a cert/key pair +func buildKubeconfig(clientConfig *restclient.Config, certData, keyData []byte) ([]byte, error) { + // Get the CA data from the bootstrap client config. + caFile, caData := clientConfig.CAFile, []byte{} + if len(caFile) == 0 { + caData = clientConfig.CAData + } + + // Build kubeconfig. + kubeconfig := clientcmdapi.Config{ + // Define a cluster stanza based on the bootstrap kubeconfig. + Clusters: map[string]*clientcmdapi.Cluster{"default-cluster": { + Server: clientConfig.Host, + InsecureSkipTLSVerify: clientConfig.Insecure, + CertificateAuthority: caFile, + CertificateAuthorityData: caData, + }}, + // Define auth based on the obtained client cert. + AuthInfos: map[string]*clientcmdapi.AuthInfo{"default-auth": { + ClientCertificateData: certData, + ClientKeyData: keyData, + }}, + // Define a context that connects the auth info and cluster, and set it as the default + Contexts: map[string]*clientcmdapi.Context{"default-context": { + Cluster: "default-cluster", + AuthInfo: "default-auth", + Namespace: "default", + }}, + CurrentContext: "default-context", + } + + return clientcmd.Write(kubeconfig) +} + +// loadKubeconfig loads kubeconfig from given secret +func loadKubeconfig(secretKey string, coreClient corev1client.CoreV1Interface) ([]byte, bool, error) { + namespace, name, err := splitMetaNamespaceKey(secretKey) + if err != nil { + return nil, false, fmt.Errorf("invalid secret name: %s", secretKey) + } + + secret, err := coreClient.Secrets(namespace).Get(context.Background(), name, metav1.GetOptions{}) + if err != nil { + if kerrors.IsNotFound(err) { + return nil, false, nil + } + return nil, false, err + } + + if secret.Data == nil { + return nil, false, nil + } + + if data, exists := secret.Data[kubeconfigSecretDataKey]; exists { + return data, true, nil + } + + return nil, false, nil +} + +// loadRESTClientConfig loads client config from given secret +func loadRESTClientConfig(secretKey string, coreClient corev1client.CoreV1Interface) (*restclient.Config, error) { + kubeconfigData, exists, err := loadKubeconfig(secretKey, coreClient) + if err != nil { + return nil, err + } + + if exists { + return clientcmd.RESTConfigFromKubeConfig(kubeconfigData) + } + + return nil, fmt.Errorf("either secret %q or data key %q not found", secretKey, kubeconfigSecretDataKey) +} + +// isClientConfigStillValid checks the provided kubeconfig to see if it has a valid +// client certificate. It returns true if the kubeconfig is valid, or an error if bootstrapping +// should stop immediately. +func isClientConfigStillValid(kubeconfigData []byte, agentName string) (bool, error) { + certs, err := getCertsFromKubeconfig(kubeconfigData) + if err != nil { + utilruntime.HandleError(err) + return false, nil + } + + now := time.Now() + for _, cert := range certs { + if now.After(cert.NotAfter) { + utilruntime.HandleError(fmt.Errorf("part of the client certificate in kubeconfig is expired: %s", cert.NotAfter)) + return false, nil + } + + if agentName == "" { + continue + } + + if !strings.HasPrefix(cert.Subject.CommonName, subjectPrefix) { + continue + } + + commonName := fmt.Sprintf("%s%s", subjectPrefix, agentName) + if cert.Subject.CommonName != commonName { + utilruntime.HandleError(fmt.Errorf("part of the client certificate has wrong common name: %s", cert.Subject.CommonName)) + return false, nil + } + } + return true, nil +} + +// getAgentNameFromKubeconfig returns agent name in the client certificate from kubeconfig +func getAgentNameFromKubeconfig(kubeconfigData []byte) (string, error) { + certs, err := getCertsFromKubeconfig(kubeconfigData) + if err != nil { + return "", err + } + + for _, cert := range certs { + if !strings.HasPrefix(cert.Subject.CommonName, subjectPrefix) { + continue + } + + return cert.Subject.CommonName[len(subjectPrefix):], nil + } + + return "", errors.New("no agent name found in certificate from kubeconfig") +} + +// getCertsFromKubeconfig returns all certificates found in tls configuration of client configuration +func getCertsFromKubeconfig(kubeconfigData []byte) ([]*x509.Certificate, error) { + clientConfig, err := clientcmd.RESTConfigFromKubeConfig(kubeconfigData) + if err != nil { + return nil, fmt.Errorf("unable to create client config from kubeconfig: %v", err) + } + + transportConfig, err := clientConfig.TransportConfig() + if err != nil { + return nil, fmt.Errorf("unable to load transport configuration from kubeconfig: %v", err) + } + + // has side effect of populating transport config data fields + if _, err := transport.TLSConfigFor(transportConfig); err != nil { + return nil, fmt.Errorf("unable to load TLS configuration from kubeconfig: %v", err) + } + + certs, err := certutil.ParseCertsPEM(transportConfig.TLS.CertData) + if err != nil { + return nil, fmt.Errorf("unable to load TLS certificates from kubeconfig: %v", err) + } + + if len(certs) == 0 { + return nil, errors.New("unable to read TLS certificates from kubeconfig") + } + + return certs, nil +} diff --git a/pkg/spoke/bootstrap/csr.go b/pkg/spoke/bootstrap/csr.go new file mode 100644 index 000000000..f975e291e --- /dev/null +++ b/pkg/spoke/bootstrap/csr.go @@ -0,0 +1,222 @@ +package bootstrap + +import ( + "context" + "crypto" + "crypto/sha512" + "crypto/x509" + "crypto/x509/pkix" + "encoding/base64" + "encoding/pem" + "fmt" + "reflect" + "time" + + certificates "k8s.io/api/certificates/v1beta1" + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + certificatesclient "k8s.io/client-go/kubernetes/typed/certificates/v1beta1" + certificatesv1beta1 "k8s.io/client-go/kubernetes/typed/certificates/v1beta1" + certutil "k8s.io/client-go/util/cert" + "k8s.io/client-go/util/certificate/csr" + "k8s.io/client-go/util/keyutil" + "k8s.io/klog" +) + +const ( + clusterNameAnnotation = "open-cluster-management.io/cluster-name" +) + +// requestNodeCertificate will create a certificate signing request for a node +// (Organization and CommonName for the CSR will be set as expected for node +// certificates) and send it to API server, then it will watch the object's +// status, once approved by API server, it will return the API server's issued +// certificate (pem-encoded). If there is any errors, or the watch timeouts, it +// will return an error. This is intended for use on nodes (kubelet and +// kubeadm). +func requestClusterCertificate(client certificatesv1beta1.CertificateSigningRequestInterface, privateKeyData []byte, clusterName, agentName string) (certData []byte, err error) { + subject := &pkix.Name{ + Organization: []string{fmt.Sprintf("%s%s", subjectPrefix, clusterName)}, + CommonName: fmt.Sprintf("%s%s", subjectPrefix, agentName), + } + + privateKey, err := keyutil.ParsePrivateKeyPEM(privateKeyData) + if err != nil { + return nil, fmt.Errorf("invalid private key for certificate request: %v", err) + } + csrData, err := certutil.MakeCSR(privateKey, subject, nil, nil) + if err != nil { + return nil, fmt.Errorf("unable to generate certificate request: %v", err) + } + + usages := []certificates.KeyUsage{ + certificates.UsageDigitalSignature, + certificates.UsageKeyEncipherment, + certificates.UsageClientAuth, + } + + // The Signer interface contains the Public() method to get the public key. + signer, ok := privateKey.(crypto.Signer) + if !ok { + return nil, fmt.Errorf("private key does not implement crypto.Signer") + } + + name, err := digestedName(signer.Public(), subject, usages) + if err != nil { + return nil, err + } + + req, err := RequestCertificate(client, csrData, name, certificates.KubeAPIServerClientKubeletSignerName, usages, privateKey, clusterName) + if err != nil { + return nil, err + } + klog.V(4).Infof("CSR %q created", name) + + ctx, cancel := context.WithTimeout(context.Background(), 3600*time.Second) + defer cancel() + + klog.Info("Waiting for client certificate to be issued") + return csr.WaitForCertificate(ctx, client, req) +} + +// This digest should include all the relevant pieces of the CSR we care about. +// We can't directly hash the serialized CSR because of random padding that we +// regenerate every loop and we include usages which are not contained in the +// CSR. This needs to be kept up to date as we add new fields to the node +// certificates and with ensureCompatible. +func digestedName(publicKey interface{}, subject *pkix.Name, usages []certificates.KeyUsage) (string, error) { + hash := sha512.New512_256() + + // Here we make sure two different inputs can't write the same stream + // to the hash. This delimiter is not in the base64.URLEncoding + // alphabet so there is no way to have spill over collisions. Without + // it 'CN:foo,ORG:bar' hashes to the same value as 'CN:foob,ORG:ar' + const delimiter = '|' + encode := base64.RawURLEncoding.EncodeToString + + write := func(data []byte) { + hash.Write([]byte(encode(data))) + hash.Write([]byte{delimiter}) + } + + publicKeyData, err := x509.MarshalPKIXPublicKey(publicKey) + if err != nil { + return "", err + } + write(publicKeyData) + + write([]byte(subject.CommonName)) + for _, v := range subject.Organization { + write([]byte(v)) + } + for _, v := range usages { + write([]byte(v)) + } + + return fmt.Sprintf("cluster-csr-%s", encode(hash.Sum(nil))), nil +} + +// RequestCertificate will either use an existing (if this process has run +// before but not to completion) or create a certificate signing request using the +// PEM encoded CSR and send it to API server, then it will watch the object's +// status, once approved by API server, it will return the API server's issued +// certificate (pem-encoded). If there is any errors, or the watch timeouts, it +// will return an error. +func RequestCertificate(client certificatesclient.CertificateSigningRequestInterface, csrData []byte, name string, signerName string, usages []certificates.KeyUsage, privateKey interface{}, clusterName string) (req *certificates.CertificateSigningRequest, err error) { + csr := &certificates.CertificateSigningRequest{ + // Username, UID, Groups will be injected by API server. + TypeMeta: metav1.TypeMeta{Kind: "CertificateSigningRequest"}, + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Labels: map[string]string{ + clusterNameAnnotation: clusterName, + }, + }, + Spec: certificates.CertificateSigningRequestSpec{ + Request: csrData, + Usages: usages, + SignerName: &signerName, + }, + } + if len(csr.Name) == 0 { + csr.GenerateName = "csr-" + } + + req, err = client.Create(context.TODO(), csr, metav1.CreateOptions{}) + switch { + case err == nil: + case errors.IsAlreadyExists(err) && len(name) > 0: + klog.Infof("csr for this cluster already exists, reusing") + req, err = client.Get(context.TODO(), name, metav1.GetOptions{}) + if err != nil { + return nil, formatError("cannot retrieve certificate signing request: %v", err) + } + if err := ensureCompatible(req, csr, privateKey); err != nil { + return nil, fmt.Errorf("retrieved csr is not compatible: %v", err) + } + klog.Infof("csr for this cluster is still valid") + default: + return nil, formatError("cannot create certificate signing request: %v", err) + } + return req, nil +} + +// ensureCompatible ensures that a CSR object is compatible with an original CSR +func ensureCompatible(new, orig *certificates.CertificateSigningRequest, privateKey interface{}) error { + newCSR, err := parseCSR(new) + if err != nil { + return fmt.Errorf("unable to parse new csr: %v", err) + } + origCSR, err := parseCSR(orig) + if err != nil { + return fmt.Errorf("unable to parse original csr: %v", err) + } + if !reflect.DeepEqual(newCSR.Subject, origCSR.Subject) { + return fmt.Errorf("csr subjects differ: new: %#v, orig: %#v", newCSR.Subject, origCSR.Subject) + } + if new.Spec.SignerName != nil && orig.Spec.SignerName != nil && *new.Spec.SignerName != *orig.Spec.SignerName { + return fmt.Errorf("csr signerNames differ: new %q, orig: %q", *new.Spec.SignerName, *orig.Spec.SignerName) + } + signer, ok := privateKey.(crypto.Signer) + if !ok { + return fmt.Errorf("privateKey is not a signer") + } + newCSR.PublicKey = signer.Public() + if err := newCSR.CheckSignature(); err != nil { + return fmt.Errorf("error validating signature new CSR against old key: %v", err) + } + if len(new.Status.Certificate) > 0 { + certs, err := certutil.ParseCertsPEM(new.Status.Certificate) + if err != nil { + return fmt.Errorf("error parsing signed certificate for CSR: %v", err) + } + now := time.Now() + for _, cert := range certs { + if now.After(cert.NotAfter) { + return fmt.Errorf("one of the certificates for the CSR has expired: %s", cert.NotAfter) + } + } + } + return nil +} + +// parseCSR extracts the CSR from the API object and decodes it. +func parseCSR(obj *certificates.CertificateSigningRequest) (*x509.CertificateRequest, error) { + // extract PEM from request object + block, _ := pem.Decode(obj.Spec.Request) + if block == nil || block.Type != "CERTIFICATE REQUEST" { + return nil, fmt.Errorf("PEM block type must be CERTIFICATE REQUEST") + } + return x509.ParseCertificateRequest(block.Bytes) +} + +// formatError preserves the type of an API message but alters the message. Expects +// a single argument format string, and returns the wrapped error. +func formatError(format string, err error) error { + if s, ok := err.(errors.APIStatus); ok { + se := &errors.StatusError{ErrStatus: s.Status()} + se.ErrStatus.Message = fmt.Sprintf(format, se.ErrStatus.Message) + return se + } + return fmt.Errorf(format, err) +} diff --git a/pkg/spoke/bootstrap/options.go b/pkg/spoke/bootstrap/options.go new file mode 100644 index 000000000..236c928e2 --- /dev/null +++ b/pkg/spoke/bootstrap/options.go @@ -0,0 +1,60 @@ +package bootstrap + +import ( + "io/ioutil" + + "github.com/spf13/pflag" +) + +const ( + defaultComponentNamespace = "open-cluster-management" +) + +// Options holds the arguments for bootstrapping +type Options struct { + ClusterNameOverride string + HubKubeconfigSecret string + BootstrapKubeconfigSecret string + CertStoreSecret string +} + +// NewOptions return a bootstrap Options with default value +func NewOptions() *Options { + return &Options{ + HubKubeconfigSecret: "hub-kubeconfig-secret", + BootstrapKubeconfigSecret: "default/bootstrap-kubeconfig-secret", + CertStoreSecret: "cert-store-secret", + } +} + +// AddFlags registers flags for bootstrap +func (o *Options) AddFlags(fs *pflag.FlagSet) { + fs.StringVar(&o.ClusterNameOverride, "cluster-name-override", o.ClusterNameOverride, "If non-empty, will use this string as cluster name instead of generated random name.") + fs.StringVar(&o.HubKubeconfigSecret, "hub-kubeconfig-secret", o.HubKubeconfigSecret, + "The name of secret in component namespace storing kubeconfig for hub connection.") + fs.StringVar(&o.BootstrapKubeconfigSecret, "bootstrap-kubeconfig-secret", o.BootstrapKubeconfigSecret, + "The name of secret containing kubeconfig for spoke agent bootstrap in the format of namespace/name.") + fs.StringVar(&o.CertStoreSecret, "cert-store-secret", o.CertStoreSecret, + "The name of secret in component namespace storing keys/client certificates against hub.") +} + +// Validate verifies the inputs. +func (o *Options) Validate() error { + return nil +} + +// Complete fills in missing values. +func (o *Options) Complete() error { + componentNamespace := getComponentNamespace() + o.CertStoreSecret = componentNamespace + "/" + o.CertStoreSecret + o.HubKubeconfigSecret = componentNamespace + "/" + o.HubKubeconfigSecret + return nil +} + +func getComponentNamespace() string { + nsBytes, err := ioutil.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/namespace") + if err != nil { + return defaultComponentNamespace + } + return string(nsBytes) +} diff --git a/pkg/spoke/bootstrap/transport.go b/pkg/spoke/bootstrap/transport.go new file mode 100644 index 000000000..97ca17144 --- /dev/null +++ b/pkg/spoke/bootstrap/transport.go @@ -0,0 +1,140 @@ +package bootstrap + +import ( + "crypto/tls" + "fmt" + "net" + "net/http" + "time" + + utilnet "k8s.io/apimachinery/pkg/util/net" + "k8s.io/apimachinery/pkg/util/wait" + restclient "k8s.io/client-go/rest" + "k8s.io/client-go/util/certificate" + "k8s.io/client-go/util/connrotation" + "k8s.io/klog" +) + +// UpdateTransport instruments a restconfig with a transport that dynamically uses +// certificates provided by the manager for TLS client auth. +// +// The config must not already provide an explicit transport. +// +// The returned function allows forcefully closing all active connections. +// +// The returned transport periodically checks the manager to determine if the +// certificate has changed. If it has, the transport shuts down all existing client +// connections, forcing the client to re-handshake with the server and use the +// new certificate. +// +// The exitAfter duration, if set, will terminate the current process if a certificate +// is not available from the store (because it has been deleted on disk or is corrupt) +// or if the certificate has expired and the server is responsive. This allows the +// process parent or the bootstrap credentials an opportunity to retrieve a new initial +// certificate. +// +// stopCh should be used to indicate when the transport is unused and doesn't need +// to continue checking the manager. +func UpdateTransport(stopCh <-chan struct{}, clientConfig *restclient.Config, clientCertificateManager certificate.Manager, exitAfter time.Duration) (func(), error) { + return updateTransport(stopCh, 10*time.Second, clientConfig, clientCertificateManager, exitAfter) +} + +// updateTransport is an internal method that exposes how often this method checks that the +// client cert has changed. +func updateTransport(stopCh <-chan struct{}, period time.Duration, clientConfig *restclient.Config, clientCertificateManager certificate.Manager, exitAfter time.Duration) (func(), error) { + if clientConfig.Transport != nil || clientConfig.Dial != nil { + return nil, fmt.Errorf("there is already a transport or dialer configured") + } + + d := connrotation.NewDialer((&net.Dialer{Timeout: 30 * time.Second, KeepAlive: 30 * time.Second}).DialContext) + + if clientCertificateManager != nil { + if err := addCertRotation(stopCh, period, clientConfig, clientCertificateManager, exitAfter, d); err != nil { + return nil, err + } + } else { + clientConfig.Dial = d.DialContext + } + + return d.CloseAll, nil +} + +func addCertRotation(stopCh <-chan struct{}, period time.Duration, clientConfig *restclient.Config, clientCertificateManager certificate.Manager, exitAfter time.Duration, d *connrotation.Dialer) error { + tlsConfig, err := restclient.TLSConfigFor(clientConfig) + if err != nil { + return fmt.Errorf("unable to configure TLS for the rest client: %v", err) + } + if tlsConfig == nil { + tlsConfig = &tls.Config{} + } + + tlsConfig.Certificates = nil + tlsConfig.GetClientCertificate = func(requestInfo *tls.CertificateRequestInfo) (*tls.Certificate, error) { + cert := clientCertificateManager.Current() + if cert == nil { + return &tls.Certificate{Certificate: nil}, nil + } + return cert, nil + } + + lastCertAvailable := time.Now() + lastCert := clientCertificateManager.Current() + go wait.Until(func() { + curr := clientCertificateManager.Current() + + if exitAfter > 0 { + now := time.Now() + if curr == nil { + // the certificate has been deleted from disk or is otherwise corrupt + if now.After(lastCertAvailable.Add(exitAfter)) { + if clientCertificateManager.ServerHealthy() { + klog.Fatalf("It has been %s since a valid client cert was found and the server is responsive, exiting.", exitAfter) + } else { + klog.Errorf("It has been %s since a valid client cert was found, but the server is not responsive. A restart may be necessary to retrieve new initial credentials.", exitAfter) + } + } + } else { + // the certificate is expired + if now.After(curr.Leaf.NotAfter) { + if clientCertificateManager.ServerHealthy() { + klog.Fatal("The currently active client certificate has expired and the server is responsive, exiting.") + } else { + klog.Error("The currently active client certificate has expired, but the server is not responsive. A restart may be necessary to retrieve new initial credentials.") + } + } + lastCertAvailable = now + } + } + + if curr == nil || lastCert == curr { + // Cert hasn't been rotated. + return + } + lastCert = curr + + klog.Info("certificate rotation detected, shutting down client connections to start using new credentials") + // The cert has been rotated. Close all existing connections to force the client + // to reperform its TLS handshake with new cert. + d.CloseAll() + }, period, stopCh) + + clientConfig.Transport = utilnet.SetTransportDefaults(&http.Transport{ + Proxy: http.ProxyFromEnvironment, + TLSHandshakeTimeout: 10 * time.Second, + TLSClientConfig: tlsConfig, + MaxIdleConnsPerHost: 25, + DialContext: d.DialContext, + }) + + // Zero out all existing TLS options since our new transport enforces them. + clientConfig.CertData = nil + clientConfig.KeyData = nil + clientConfig.CertFile = "" + clientConfig.KeyFile = "" + clientConfig.CAData = nil + clientConfig.CAFile = "" + clientConfig.Insecure = false + clientConfig.NextProtos = nil + + return nil +}