github
/
k3s
mirror of https://github.com/k3s-io/k3s
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Restore "Make bootstrap client cert loading part of rotation"" 

This reverts the revert of commit 3464222267.
pull/564/head
Clayton Coleman 2018-11-17 13:44:58 -05:00
parent 8996fc1639
commit 486577df17
8 changed files with 604 additions and 139 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
cmd/kubelet/app/BUILD
View File

@ -8,8 +8,22 @@ load(
go_test(
name = "go_default_test",
srcs = ["server_test.go"],
srcs = [
"server_bootstrap_test.go",
"server_test.go",
],
embed = [":go_default_library"],
deps = [
"//staging/src/k8s.io/api/certificates/v1beta1:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
"//staging/src/k8s.io/client-go/rest:go_default_library",
"//staging/src/k8s.io/client-go/util/cert:go_default_library",
"//vendor/github.com/cloudflare/cfssl/config:go_default_library",
"//vendor/github.com/cloudflare/cfssl/signer:go_default_library",
"//vendor/github.com/cloudflare/cfssl/signer/local:go_default_library",
],
)
go_library(
@ -113,6 +127,7 @@ go_library(
"//staging/src/k8s.io/client-go/kubernetes:go_default_library",
"//staging/src/k8s.io/client-go/kubernetes/typed/authentication/v1beta1:go_default_library",
"//staging/src/k8s.io/client-go/kubernetes/typed/authorization/v1beta1:go_default_library",
"//staging/src/k8s.io/client-go/kubernetes/typed/certificates/v1beta1:go_default_library",
"//staging/src/k8s.io/client-go/kubernetes/typed/core/v1:go_default_library",
"//staging/src/k8s.io/client-go/rest:go_default_library",
"//staging/src/k8s.io/client-go/tools/clientcmd:go_default_library",

156
cmd/kubelet/app/server.go
View File

@ -50,6 +50,7 @@ import (
"k8s.io/apiserver/pkg/util/flag"
"k8s.io/client-go/dynamic"
clientset "k8s.io/client-go/kubernetes"
certificatesclient "k8s.io/client-go/kubernetes/typed/certificates/v1beta1"
v1core "k8s.io/client-go/kubernetes/typed/core/v1"
restclient "k8s.io/client-go/rest"
"k8s.io/client-go/tools/clientcmd"
@ -537,66 +538,39 @@ func run(s *options.KubeletServer, kubeDeps *kubelet.Dependencies, stopCh <-chan
return err
}
if s.BootstrapKubeconfig != "" {
if err := bootstrap.LoadClientCert(s.KubeConfig, s.BootstrapKubeconfig, s.CertDirectory, nodeName); err != nil {
return err
}
}
// if in standalone mode, indicate as much by setting all clients to nil
if standaloneMode {
switch {
case standaloneMode:
kubeDeps.KubeClient = nil
kubeDeps.DynamicKubeClient = nil
kubeDeps.EventClient = nil
kubeDeps.HeartbeatClient = nil
klog.Warningf("standalone mode, no API client")
} else if kubeDeps.KubeClient == nil || kubeDeps.EventClient == nil || kubeDeps.HeartbeatClient == nil || kubeDeps.DynamicKubeClient == nil {
// initialize clients if not standalone mode and any of the clients are not provided
var kubeClient clientset.Interface
var eventClient v1core.EventsGetter
var heartbeatClient clientset.Interface
var dynamicKubeClient dynamic.Interface
clientConfig, err := createAPIServerClientConfig(s)
if err != nil {
return fmt.Errorf("invalid kubeconfig: %v", err)
}
var clientCertificateManager certificate.Manager
if s.RotateCertificates && utilfeature.DefaultFeatureGate.Enabled(features.RotateKubeletClientCertificate) {
clientCertificateManager, err = kubeletcertificate.NewKubeletClientCertificateManager(s.CertDirectory, nodeName, clientConfig.CertData, clientConfig.KeyData, clientConfig.CertFile, clientConfig.KeyFile)
if err != nil {
return err
}
}
// we set exitAfter to five minutes because we use this client configuration to request new certs - if we are unable
// to request new certs, we will be unable to continue normal operation. Exiting the process allows a wrapper
// or the bootstrapping credentials to potentially lay down new initial config.
closeAllConns, err := kubeletcertificate.UpdateTransport(wait.NeverStop, clientConfig, clientCertificateManager, 5*time.Minute)
case kubeDeps.KubeClient == nil, kubeDeps.EventClient == nil, kubeDeps.HeartbeatClient == nil, kubeDeps.DynamicKubeClient == nil:
clientConfig, closeAllConns, err := buildKubeletClientConfig(s, nodeName)
if err != nil {
return err
}
kubeDeps.OnHeartbeatFailure = closeAllConns
kubeClient, err = clientset.NewForConfig(clientConfig)
kubeDeps.KubeClient, err = clientset.NewForConfig(clientConfig)
if err != nil {
klog.Warningf("New kubeClient from clientConfig error: %v", err)
} else if kubeClient.CertificatesV1beta1() != nil && clientCertificateManager != nil {
klog.V(2).Info("Starting client certificate rotation.")
clientCertificateManager.SetCertificateSigningRequestClient(kubeClient.CertificatesV1beta1().CertificateSigningRequests())
clientCertificateManager.Start()
return fmt.Errorf("failed to initialize kubelet client: %v", err)
}
dynamicKubeClient, err = dynamic.NewForConfig(clientConfig)
kubeDeps.DynamicKubeClient, err = dynamic.NewForConfig(clientConfig)
if err != nil {
klog.Warningf("Failed to initialize dynamic KubeClient: %v", err)
return fmt.Errorf("failed to initialize kubelet dynamic client: %v", err)
}
// make a separate client for events
eventClientConfig := *clientConfig
eventClientConfig.QPS = float32(s.EventRecordQPS)
eventClientConfig.Burst = int(s.EventBurst)
eventClient, err = v1core.NewForConfig(&eventClientConfig)
kubeDeps.EventClient, err = v1core.NewForConfig(&eventClientConfig)
if err != nil {
klog.Warningf("Failed to create API Server client for Events: %v", err)
return fmt.Errorf("failed to initialize kubelet event client: %v", err)
}
// make a separate client for heartbeat with throttling disabled and a timeout attached
@ -610,28 +584,18 @@ func run(s *options.KubeletServer, kubeDeps *kubelet.Dependencies, stopCh <-chan
}
}
heartbeatClientConfig.QPS = float32(-1)
heartbeatClient, err = clientset.NewForConfig(&heartbeatClientConfig)
kubeDeps.HeartbeatClient, err = clientset.NewForConfig(&heartbeatClientConfig)
if err != nil {
klog.Warningf("Failed to create API Server client for heartbeat: %v", err)
return fmt.Errorf("failed to initialize kubelet heartbeat client: %v", err)
}
// csiClient works with CRDs that support json only
clientConfig.ContentType = "application/json"
csiClient, err := csiclientset.NewForConfig(clientConfig)
// CRDs are JSON only, and client renegotiation for streaming is not correct as per #67803
csiClientConfig := restclient.CopyConfig(clientConfig)
csiClientConfig.ContentType = "application/json"
kubeDeps.CSIClient, err = csiclientset.NewForConfig(csiClientConfig)
if err != nil {
klog.Warningf("Failed to create CSI API client: %v", err)
return fmt.Errorf("failed to initialize kubelet storage client: %v", err)
}
kubeDeps.KubeClient = kubeClient
kubeDeps.DynamicKubeClient = dynamicKubeClient
if heartbeatClient != nil {
kubeDeps.HeartbeatClient = heartbeatClient
kubeDeps.OnHeartbeatFailure = closeAllConns
}
if eventClient != nil {
kubeDeps.EventClient = eventClient
}
kubeDeps.CSIClient = csiClient
}
// If the kubelet config controller is available, and dynamic config is enabled, start the config and status sync loops
@ -771,6 +735,81 @@ func run(s *options.KubeletServer, kubeDeps *kubelet.Dependencies, stopCh <-chan
return nil
}
// buildKubeletClientConfig constructs the appropriate client config for the kubelet depending on whether
// bootstrapping is enabled or client certificate rotation is enabled.
func buildKubeletClientConfig(s *options.KubeletServer, nodeName types.NodeName) (*restclient.Config, func(), error) {
if s.RotateCertificates && utilfeature.DefaultFeatureGate.Enabled(features.RotateKubeletClientCertificate) {
certConfig, clientConfig, err := bootstrap.LoadClientConfig(s.KubeConfig, s.BootstrapKubeconfig, s.CertDirectory)
if err != nil {
return nil, nil, err
}
clientCertificateManager, err := buildClientCertificateManager(certConfig, clientConfig, s.CertDirectory, nodeName)
if err != nil {
return nil, nil, err
}
// the rotating transport will use the cert from the cert manager instead of these files
transportConfig := restclient.CopyConfig(clientConfig)
transportConfig.CertFile = ""
transportConfig.KeyFile = ""
// we set exitAfter to five minutes because we use this client configuration to request new certs - if we are unable
// to request new certs, we will be unable to continue normal operation. Exiting the process allows a wrapper
// or the bootstrapping credentials to potentially lay down new initial config.
closeAllConns, err := kubeletcertificate.UpdateTransport(wait.NeverStop, transportConfig, clientCertificateManager, 5*time.Minute)
if err != nil {
return nil, nil, err
}
klog.V(2).Info("Starting client certificate rotation.")
clientCertificateManager.Start()
return transportConfig, closeAllConns, nil
}
if len(s.BootstrapKubeconfig) > 0 {
if err := bootstrap.LoadClientCert(s.KubeConfig, s.BootstrapKubeconfig, s.CertDirectory, nodeName); err != nil {
return nil, nil, err
}
}
clientConfig, err := createAPIServerClientConfig(s)
if err != nil {
return nil, nil, fmt.Errorf("invalid kubeconfig: %v", err)
}
return clientConfig, nil, nil
}
// buildClientCertificateManager creates a certificate manager that will use certConfig to request a client certificate
// if no certificate is available, or the most recent clientConfig (which is assumed to point to the cert that the manager will
// write out).
func buildClientCertificateManager(certConfig, clientConfig *restclient.Config, certDir string, nodeName types.NodeName) (certificate.Manager, error) {
newClientFn := func(current *tls.Certificate) (certificatesclient.CertificateSigningRequestInterface, error) {
// If we have a valid certificate, use that to fetch CSRs. Otherwise use the bootstrap
// credentials. In the future it would be desirable to change the behavior of bootstrap
// to always fall back to the external bootstrap credentials when such credentials are
// provided by a fundamental trust system like cloud VM identity or an HSM module.
config := certConfig
if current != nil {
config = clientConfig
}
client, err := clientset.NewForConfig(config)
if err != nil {
return nil, err
}
return client.CertificatesV1beta1().CertificateSigningRequests(), nil
}
return kubeletcertificate.NewKubeletClientCertificateManager(
certDir,
nodeName,
clientConfig.CertFile,
clientConfig.KeyFile,
newClientFn,
)
}
// getNodeName returns the node name according to the cloud provider
// if cloud provider is specified. Otherwise, returns the hostname of the node.
func getNodeName(cloud cloudprovider.Interface, hostname string) (types.NodeName, error) {
@ -869,11 +908,10 @@ func kubeconfigClientConfig(s *options.KubeletServer) (*restclient.Config, error
// createClientConfig creates a client configuration from the command line arguments.
// If --kubeconfig is explicitly set, it will be used.
func createClientConfig(s *options.KubeletServer) (*restclient.Config, error) {
if s.BootstrapKubeconfig != "" || len(s.KubeConfig) > 0 {
if len(s.BootstrapKubeconfig) > 0 || len(s.KubeConfig) > 0 {
return kubeconfigClientConfig(s)
} else {
return nil, fmt.Errorf("createClientConfig called in standalone mode")
}
return nil, fmt.Errorf("createClientConfig called in standalone mode")
}
// createAPIServerClientConfig generates a client.Config from command line flags

281
cmd/kubelet/app/server_bootstrap_test.go Normal file
View File

@ -0,0 +1,281 @@
/*
Copyright 2016 The Kubernetes Authors.
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 app
import (
"crypto/ecdsa"
"crypto/elliptic"
cryptorand "crypto/rand"
"crypto/x509"
"encoding/json"
"io/ioutil"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"sync"
"testing"
"time"
cfsslconfig "github.com/cloudflare/cfssl/config"
cfsslsigner "github.com/cloudflare/cfssl/signer"
cfssllocal "github.com/cloudflare/cfssl/signer/local"
certapi "k8s.io/api/certificates/v1beta1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
restclient "k8s.io/client-go/rest"
certutil "k8s.io/client-go/util/cert"
)
// Test_buildClientCertificateManager validates that we can build a local client cert
// manager that will use the bootstrap client until we get a valid cert, then use our
// provided identity on subsequent requests.
func Test_buildClientCertificateManager(t *testing.T) {
testDir, err := ioutil.TempDir("", "kubeletcert")
if err != nil {
t.Fatal(err)
}
defer func() { os.RemoveAll(testDir) }()
serverPrivateKey, err := ecdsa.GenerateKey(elliptic.P256(), cryptorand.Reader)
if err != nil {
t.Fatal(err)
}
serverCA, err := certutil.NewSelfSignedCACert(certutil.Config{
CommonName: "the-test-framework",
}, serverPrivateKey)
if err != nil {
t.Fatal(err)
}
server := &csrSimulator{
t: t,
serverPrivateKey: serverPrivateKey,
serverCA: serverCA,
}
s := httptest.NewServer(server)
defer s.Close()
config1 := &restclient.Config{
UserAgent: "FirstClient",
Host: s.URL,
}
config2 := &restclient.Config{
UserAgent: "SecondClient",
Host: s.URL,
}
nodeName := types.NodeName("test")
m, err := buildClientCertificateManager(config1, config2, testDir, nodeName)
if err != nil {
t.Fatal(err)
}
defer m.Stop()
r := m.(rotater)
// get an expired CSR (simulating historical output)
server.backdate = 2 * time.Hour
server.expectUserAgent = "FirstClient"
ok, err := r.RotateCerts()
if !ok || err != nil {
t.Fatalf("unexpected rotation err: %t %v", ok, err)
}
if cert := m.Current(); cert != nil {
t.Fatalf("Unexpected cert, should be expired: %#v", cert)
}
fi := getFileInfo(testDir)
if len(fi) != 2 {
t.Fatalf("Unexpected directory contents: %#v", fi)
}
// if m.Current() == nil, then we try again and get a valid
// client
server.backdate = 0
server.expectUserAgent = "FirstClient"
if ok, err := r.RotateCerts(); !ok || err != nil {
t.Fatalf("unexpected rotation err: %t %v", ok, err)
}
if cert := m.Current(); cert == nil {
t.Fatalf("Unexpected cert, should be valid: %#v", cert)
}
fi = getFileInfo(testDir)
if len(fi) != 2 {
t.Fatalf("Unexpected directory contents: %#v", fi)
}
// if m.Current() != nil, then we should use the second client
server.expectUserAgent = "SecondClient"
if ok, err := r.RotateCerts(); !ok || err != nil {
t.Fatalf("unexpected rotation err: %t %v", ok, err)
}
if cert := m.Current(); cert == nil {
t.Fatalf("Unexpected cert, should be valid: %#v", cert)
}
fi = getFileInfo(testDir)
if len(fi) != 2 {
t.Fatalf("Unexpected directory contents: %#v", fi)
}
}
func getFileInfo(dir string) map[string]os.FileInfo {
fi := make(map[string]os.FileInfo)
filepath.Walk(dir, func(path string, info os.FileInfo, err error) error {
if path == dir {
return nil
}
fi[path] = info
if !info.IsDir() {
os.Remove(path)
}
return nil
})
return fi
}
type rotater interface {
RotateCerts() (bool, error)
}
func getCSR(req *http.Request) (*certapi.CertificateSigningRequest, error) {
if req.Body == nil {
return nil, nil
}
body, err := ioutil.ReadAll(req.Body)
if err != nil {
return nil, err
}
csr := &certapi.CertificateSigningRequest{}
if err := json.Unmarshal(body, csr); err != nil {
return nil, err
}
return csr, nil
}
func mustMarshal(obj interface{}) []byte {
data, err := json.Marshal(obj)
if err != nil {
panic(err)
}
return data
}
type csrSimulator struct {
t *testing.T
serverPrivateKey *ecdsa.PrivateKey
serverCA *x509.Certificate
backdate time.Duration
expectUserAgent string
lock sync.Mutex
csr *certapi.CertificateSigningRequest
}
func (s *csrSimulator) ServeHTTP(w http.ResponseWriter, req *http.Request) {
s.lock.Lock()
defer s.lock.Unlock()
t := s.t
t.Logf("Request %s %s %s", req.Method, req.URL, req.UserAgent())
if len(s.expectUserAgent) > 0 && req.UserAgent() != s.expectUserAgent {
t.Errorf("Unexpected user agent: %s", req.UserAgent())
}
switch {
case req.Method == "POST" && req.URL.Path == "/apis/certificates.k8s.io/v1beta1/certificatesigningrequests":
csr, err := getCSR(req)
if err != nil {
t.Fatal(err)
}
if csr.Name == "" {
csr.Name = "test-csr"
}
csr.UID = types.UID("1")
csr.ResourceVersion = "1"
data := mustMarshal(csr)
w.Header().Set("Content-Type", "application/json")
w.Write(data)
csr = csr.DeepCopy()
csr.ResourceVersion = "2"
var usages []string
for _, usage := range csr.Spec.Usages {
usages = append(usages, string(usage))
}
policy := &cfsslconfig.Signing{
Default: &cfsslconfig.SigningProfile{
Usage: usages,
Expiry: time.Hour,
ExpiryString: time.Hour.String(),
Backdate: s.backdate,
},
}
cfs, err := cfssllocal.NewSigner(s.serverPrivateKey, s.serverCA, cfsslsigner.DefaultSigAlgo(s.serverPrivateKey), policy)
if err != nil {
t.Fatal(err)
}
csr.Status.Certificate, err = cfs.Sign(cfsslsigner.SignRequest{
Request: string(csr.Spec.Request),
})
if err != nil {
t.Fatal(err)
}
csr.Status.Conditions = []certapi.CertificateSigningRequestCondition{
{Type: certapi.CertificateApproved},
}
s.csr = csr
case req.Method == "GET" && req.URL.Path == "/apis/certificates.k8s.io/v1beta1/certificatesigningrequests" && req.URL.RawQuery == "fieldSelector=metadata.name%3Dtest-csr&limit=500":
if s.csr == nil {
t.Fatalf("no csr")
}
csr := s.csr.DeepCopy()
data := mustMarshal(&certapi.CertificateSigningRequestList{
ListMeta: metav1.ListMeta{
ResourceVersion: "2",
},
Items: []certapi.CertificateSigningRequest{
*csr,
},
})
w.Header().Set("Content-Type", "application/json")
w.Write(data)
case req.Method == "GET" && req.URL.Path == "/apis/certificates.k8s.io/v1beta1/certificatesigningrequests" && req.URL.RawQuery == "fieldSelector=metadata.name%3Dtest-csr&resourceVersion=2&watch=true":
if s.csr == nil {
t.Fatalf("no csr")
}
csr := s.csr.DeepCopy()
data := mustMarshal(&metav1.WatchEvent{
Type: "ADDED",
Object: runtime.RawExtension{
Raw: mustMarshal(csr),
},
})
w.Header().Set("Content-Type", "application/json")
w.Write(data)
default:
t.Fatalf("unexpected request: %s %s", req.Method, req.URL)
}
}

55
pkg/kubelet/certificate/bootstrap/bootstrap.go
View File

@ -47,11 +47,60 @@ import (
const tmpPrivateKeyFile = "kubelet-client.key.tmp"
// LoadClientConfig tries to load the appropriate client config for retrieving certs and for use by users.
// If bootstrapPath is empty, only kubeconfigPath is checked. If bootstrap path is set and the contents
// of kubeconfigPath are valid, both certConfig and userConfig will point to that file. Otherwise the
// kubeconfigPath on disk is populated based on bootstrapPath but pointing to the location of the client cert
// in certDir. This preserves the historical behavior of bootstrapping where on subsequent restarts the
// most recent client cert is used to request new client certs instead of the initial token.
func LoadClientConfig(kubeconfigPath, bootstrapPath, certDir string) (certConfig, userConfig *restclient.Config, err error) {
if len(bootstrapPath) == 0 {
clientConfig, err := loadRESTClientConfig(kubeconfigPath)
if err != nil {
return nil, nil, fmt.Errorf("unable to load kubeconfig: %v", err)
}
return clientConfig, clientConfig, nil
}
store, err := certificate.NewFileStore("kubelet-client", certDir, certDir, "", "")
if err != nil {
return nil, nil, fmt.Errorf("unable to build bootstrap cert store")
}
ok, err := verifyBootstrapClientConfig(kubeconfigPath)
if err != nil {
return nil, nil, err
}
// use the current client config
if ok {
clientConfig, err := loadRESTClientConfig(kubeconfigPath)
if err != nil {
return nil, nil, fmt.Errorf("unable to load kubeconfig: %v", err)
}
return clientConfig, clientConfig, nil
}
bootstrapClientConfig, err := loadRESTClientConfig(bootstrapPath)
if err != nil {
return nil, nil, fmt.Errorf("unable to load bootstrap kubeconfig: %v", err)
}
clientConfig := restclient.AnonymousClientConfig(bootstrapClientConfig)
pemPath := store.CurrentPath()
clientConfig.KeyFile = pemPath
clientConfig.CertFile = pemPath
if err := writeKubeconfigFromBootstrapping(clientConfig, kubeconfigPath, pemPath); err != nil {
return nil, nil, err
}
return bootstrapClientConfig, clientConfig, nil
}
// LoadClientCert requests a client cert for kubelet if the kubeconfigPath file does not exist.
// The kubeconfig at bootstrapPath is used to request a client certificate from the API server.
// On success, a kubeconfig file referencing the generated key and obtained certificate is written to kubeconfigPath.
// The certificate and key file are stored in certDir.
func LoadClientCert(kubeconfigPath string, bootstrapPath string, certDir string, nodeName types.NodeName) error {
func LoadClientCert(kubeconfigPath, bootstrapPath, certDir string, nodeName types.NodeName) error {
// Short-circuit if the kubeconfig file exists and is valid.
ok, err := verifyBootstrapClientConfig(kubeconfigPath)
if err != nil {
@ -117,8 +166,10 @@ func LoadClientCert(kubeconfigPath string, bootstrapPath string, certDir string,
klog.V(2).Infof("failed cleaning up private key file %q: %v", privKeyPath, err)
}
pemPath := store.CurrentPath()
return writeKubeconfigFromBootstrapping(bootstrapClientConfig, kubeconfigPath, store.CurrentPath())
}
func writeKubeconfigFromBootstrapping(bootstrapClientConfig *restclient.Config, kubeconfigPath, pemPath string) error {
// Get the CA data from the bootstrap client config.
caFile, caData := bootstrapClientConfig.CAFile, []byte{}
if len(caFile) == 0 {

25
pkg/kubelet/certificate/kubelet.go
View File

@ -17,6 +17,7 @@ limitations under the License.
package certificate
import (
"crypto/tls"
"crypto/x509"
"crypto/x509/pkix"
"fmt"
@ -29,7 +30,7 @@ import (
"k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/types"
clientset "k8s.io/client-go/kubernetes"
clientcertificates "k8s.io/client-go/kubernetes/typed/certificates/v1beta1"
certificatesclient "k8s.io/client-go/kubernetes/typed/certificates/v1beta1"
"k8s.io/client-go/util/certificate"
kubeletconfig "k8s.io/kubernetes/pkg/kubelet/apis/config"
"k8s.io/kubernetes/pkg/kubelet/metrics"
@ -38,7 +39,7 @@ import (
// NewKubeletServerCertificateManager creates a certificate manager for the kubelet when retrieving a server certificate
// or returns an error.
func NewKubeletServerCertificateManager(kubeClient clientset.Interface, kubeCfg *kubeletconfig.KubeletConfiguration, nodeName types.NodeName, getAddresses func() []v1.NodeAddress, certDirectory string) (certificate.Manager, error) {
var certSigningRequestClient clientcertificates.CertificateSigningRequestInterface
var certSigningRequestClient certificatesclient.CertificateSigningRequestInterface
if kubeClient != nil && kubeClient.CertificatesV1beta1() != nil {
certSigningRequestClient = kubeClient.CertificatesV1beta1().CertificateSigningRequests()
}
@ -78,8 +79,10 @@ func NewKubeletServerCertificateManager(kubeClient clientset.Interface, kubeCfg
}
m, err := certificate.NewManager(&certificate.Config{
CertificateSigningRequestClient: certSigningRequestClient,
GetTemplate: getTemplate,
ClientFn: func(current *tls.Certificate) (certificatesclient.CertificateSigningRequestInterface, error) {
return certSigningRequestClient, nil
},
GetTemplate: getTemplate,
Usages: []certificates.KeyUsage{
// https://tools.ietf.org/html/rfc5280#section-4.2.1.3
//
@ -142,10 +145,9 @@ func addressesToHostnamesAndIPs(addresses []v1.NodeAddress) (dnsNames []string,
}
// NewKubeletClientCertificateManager sets up a certificate manager without a
// client that can be used to sign new certificates (or rotate). It answers with
// whatever certificate it is initialized with. If a CSR client is set later, it
// may begin rotating/renewing the client cert
func NewKubeletClientCertificateManager(certDirectory string, nodeName types.NodeName, certData []byte, keyData []byte, certFile string, keyFile string) (certificate.Manager, error) {
// client that can be used to sign new certificates (or rotate). If a CSR
// client is set later, it may begin rotating/renewing the client cert.
func NewKubeletClientCertificateManager(certDirectory string, nodeName types.NodeName, certFile string, keyFile string, clientFn certificate.CSRClientFunc) (certificate.Manager, error) {
certificateStore, err := certificate.NewFileStore(
"kubelet-client",
certDirectory,
@ -166,6 +168,7 @@ func NewKubeletClientCertificateManager(certDirectory string, nodeName types.Nod
prometheus.MustRegister(certificateExpiration)
m, err := certificate.NewManager(&certificate.Config{
ClientFn: clientFn,
Template: &x509.CertificateRequest{
Subject: pkix.Name{
CommonName: fmt.Sprintf("system:node:%s", nodeName),
@ -187,10 +190,8 @@ func NewKubeletClientCertificateManager(certDirectory string, nodeName types.Nod
// authenticate itself to the TLS server.
certificates.UsageClientAuth,
},
CertificateStore: certificateStore,
BootstrapCertificatePEM: certData,
BootstrapKeyPEM: keyData,
CertificateExpiration: certificateExpiration,
CertificateStore: certificateStore,
CertificateExpiration: certificateExpiration,
})
if err != nil {
return nil, fmt.Errorf("failed to initialize client certificate manager: %v", err)

4
pkg/kubelet/certificate/transport_test.go
View File

@ -124,7 +124,9 @@ func (f *fakeManager) SetCertificateSigningRequestClient(certificatesclient.Cert
func (f *fakeManager) ServerHealthy() bool { return f.healthy }
func (f *fakeManager) Start() {}
func (f *fakeManager) Start() {}
func (f *fakeManager) Stop() {}
func (f *fakeManager) RotateCerts() (bool, error) { return false, nil }
func (f *fakeManager) Current() *tls.Certificate {
if val := f.cert.Load(); val != nil {

130
staging/src/k8s.io/client-go/util/certificate/certificate_manager.go
View File

@ -48,11 +48,10 @@ var certificateWaitBackoff = wait.Backoff{Duration: 30 * time.Second, Steps: 4,
// manager. In the background it communicates with the API server to get new
// certificates for certificates about to expire.
type Manager interface {
// CertificateSigningRequestClient sets the client interface that is used for
// signing new certificates generated as part of rotation.
SetCertificateSigningRequestClient(certificatesclient.CertificateSigningRequestInterface) error
// Start the API server status sync loop.
Start()
// Stop the cert manager loop.
Stop()
// Current returns the currently selected certificate from the
// certificate manager, as well as the associated certificate and key data
// in PEM format.
@ -67,11 +66,11 @@ type Manager interface {
// Config is the set of configuration parameters available for a new Manager.
type Config struct {
// CertificateSigningRequestClient will be used for signing new certificate
// requests generated when a key rotation occurs. It must be set either at
// initialization or by using CertificateSigningRequestClient before
// Manager.Start() is called.
CertificateSigningRequestClient certificatesclient.CertificateSigningRequestInterface
// ClientFn will be used to create a client for
// signing new certificate requests generated when a key rotation occurs.
// It must be set at initialization. The function will never be invoked
// in parallel. It is passed the current client certificate if one exists.
ClientFn CSRClientFunc
// Template is the CertificateRequest that will be used as a template for
// generating certificate signing requests for all new keys generated as
// part of rotation. It follows the same rules as the template parameter of
@ -141,21 +140,34 @@ type Gauge interface {
// NoCertKeyError indicates there is no cert/key currently available.
type NoCertKeyError string
// CSRClientFunc returns a new client for requesting CSRs. It passes the
// current certificate if one is available and valid.
type CSRClientFunc func(current *tls.Certificate) (certificatesclient.CertificateSigningRequestInterface, error)
func (e *NoCertKeyError) Error() string { return string(*e) }
type manager struct {
certSigningRequestClient certificatesclient.CertificateSigningRequestInterface
getTemplate func() *x509.CertificateRequest
lastRequestLock sync.Mutex
lastRequest *x509.CertificateRequest
dynamicTemplate bool
usages []certificates.KeyUsage
certStore Store
certAccessLock sync.RWMutex
cert *tls.Certificate
forceRotation bool
certificateExpiration Gauge
serverHealth bool
getTemplate func() *x509.CertificateRequest
lastRequestLock sync.Mutex
lastRequest *x509.CertificateRequest
dynamicTemplate bool
usages []certificates.KeyUsage
forceRotation bool
certStore Store
certificateExpiration Gauge
// the following variables must only be accessed under certAccessLock
certAccessLock sync.RWMutex
cert *tls.Certificate
serverHealth bool
// the clientFn must only be accessed under the clientAccessLock
clientAccessLock sync.Mutex
clientFn CSRClientFunc
stopCh chan struct{}
stopped bool
}
// NewManager returns a new certificate manager. A certificate manager is
@ -176,14 +188,15 @@ func NewManager(config *Config) (Manager, error) {
}
m := manager{
certSigningRequestClient: config.CertificateSigningRequestClient,
getTemplate: getTemplate,
dynamicTemplate: config.GetTemplate != nil,
usages: config.Usages,
certStore: config.CertificateStore,
cert: cert,
forceRotation: forceRotation,
certificateExpiration: config.CertificateExpiration,
stopCh: make(chan struct{}),
clientFn: config.ClientFn,
getTemplate: getTemplate,
dynamicTemplate: config.GetTemplate != nil,
usages: config.Usages,
certStore: config.CertificateStore,
cert: cert,
forceRotation: forceRotation,
certificateExpiration: config.CertificateExpiration,
}
return &m, nil
@ -192,10 +205,14 @@ func NewManager(config *Config) (Manager, error) {
// Current returns the currently selected certificate from the certificate
// manager. This can be nil if the manager was initialized without a
// certificate and has not yet received one from the
// CertificateSigningRequestClient.
// CertificateSigningRequestClient, or if the current cert has expired.
func (m *manager) Current() *tls.Certificate {
m.certAccessLock.RLock()
defer m.certAccessLock.RUnlock()
if m.cert != nil && m.cert.Leaf != nil && time.Now().After(m.cert.Leaf.NotAfter) {
klog.V(2).Infof("Current certificate is expired.")
return nil
}
return m.cert
}
@ -207,18 +224,15 @@ func (m *manager) ServerHealthy() bool {
return m.serverHealth
}
// SetCertificateSigningRequestClient sets the client interface that is used
// for signing new certificates generated as part of rotation. It must be
// called before Start() and can not be used to change the
// CertificateSigningRequestClient that has already been set. This method is to
// support the one specific scenario where the CertificateSigningRequestClient
// uses the CertificateManager.
func (m *manager) SetCertificateSigningRequestClient(certSigningRequestClient certificatesclient.CertificateSigningRequestInterface) error {
if m.certSigningRequestClient == nil {
m.certSigningRequestClient = certSigningRequestClient
return nil
// Stop terminates the manager.
func (m *manager) Stop() {
m.clientAccessLock.Lock()
defer m.clientAccessLock.Unlock()
if m.stopped {
return
}
return fmt.Errorf("property CertificateSigningRequestClient is already set")
close(m.stopCh)
m.stopped = true
}
// Start will start the background work of rotating the certificates.
@ -226,7 +240,7 @@ func (m *manager) Start() {
// Certificate rotation depends on access to the API server certificate
// signing API, so don't start the certificate manager if we don't have a
// client.
if m.certSigningRequestClient == nil {
if m.clientFn == nil {
klog.V(2).Infof("Certificate rotation is not enabled, no connection to the apiserver.")
return
}
@ -234,7 +248,7 @@ func (m *manager) Start() {
klog.V(2).Infof("Certificate rotation is enabled.")
templateChanged := make(chan struct{})
go wait.Forever(func() {
go wait.Until(func() {
deadline := m.nextRotationDeadline()
if sleepInterval := deadline.Sub(time.Now()); sleepInterval > 0 {
klog.V(2).Infof("Waiting %v for next certificate rotation", sleepInterval)
@ -269,17 +283,17 @@ func (m *manager) Start() {
utilruntime.HandleError(fmt.Errorf("Reached backoff limit, still unable to rotate certs: %v", err))
wait.PollInfinite(32*time.Second, m.rotateCerts)
}
}, time.Second)
}, time.Second, m.stopCh)
if m.dynamicTemplate {
go wait.Forever(func() {
go wait.Until(func() {
// check if the current template matches what we last requested
if !m.certSatisfiesTemplate() && !reflect.DeepEqual(m.getLastRequest(), m.getTemplate()) {
// if the template is different, queue up an interrupt of the rotation deadline loop.
// if we've requested a CSR that matches the new template by the time the interrupt is handled, the interrupt is disregarded.
templateChanged <- struct{}{}
}
}, time.Second)
}, time.Second, m.stopCh)
}
}
@ -327,11 +341,26 @@ func getCurrentCertificateOrBootstrap(
return &bootstrapCert, true, nil
}
func (m *manager) getClient() (certificatesclient.CertificateSigningRequestInterface, error) {
current := m.Current()
m.clientAccessLock.Lock()
defer m.clientAccessLock.Unlock()
return m.clientFn(current)
}
// RotateCerts is exposed for testing only and is not a part of the public interface.
// Returns true if it changed the cert, false otherwise. Error is only returned in
// exceptional cases.
func (m *manager) RotateCerts() (bool, error) {
return m.rotateCerts()
}
// rotateCerts attempts to request a client cert from the server, wait a reasonable
// period of time for it to be signed, and then update the cert on disk. If it cannot
// retrieve a cert, it will return false. It will only return error in exceptional cases.
// This method also keeps track of "server health" by interpreting the responses it gets
// from the server on the various calls it makes.
// TODO: return errors, have callers handle and log them correctly
func (m *manager) rotateCerts() (bool, error) {
klog.V(2).Infof("Rotating certificates")
@ -341,9 +370,16 @@ func (m *manager) rotateCerts() (bool, error) {
return false, nil
}
// request the client each time
client, err := m.getClient()
if err != nil {
utilruntime.HandleError(fmt.Errorf("Unable to load a client to request certificates: %v", err))
return false, nil
}
// Call the Certificate Signing Request API to get a certificate for the
// new private key.
req, err := csr.RequestCertificate(m.certSigningRequestClient, csrPEM, "", m.usages, privateKey)
req, err := csr.RequestCertificate(client, csrPEM, "", m.usages, privateKey)
if err != nil {
utilruntime.HandleError(fmt.Errorf("Failed while requesting a signed certificate from the master: %v", err))
return false, m.updateServerError(err)
@ -359,7 +395,7 @@ func (m *manager) rotateCerts() (bool, error) {
var crtPEM []byte
watchDuration := time.Minute
if err := wait.ExponentialBackoff(certificateWaitBackoff, func() (bool, error) {
data, err := csr.WaitForCertificate(m.certSigningRequestClient, req, watchDuration)
data, err := csr.WaitForCertificate(client, req, watchDuration)
switch {
case err == nil:
crtPEM = data

75
staging/src/k8s.io/client-go/util/certificate/certificate_manager_test.go
View File

@ -60,6 +60,23 @@ iQIgZX08DA8VfvcA5/Xj1Zjdey9FVY6POLXen6RPiabE97UCICp6eUW7ht+2jjar
e35EltCRCjoejRHTuN9TC0uCoVipAiAXaJIx/Q47vGwiw6Y8KXsNU6y54gTbOSxX
54LzHNk/+Q==
-----END RSA PRIVATE KEY-----`)
var expiredStoreCertData = newCertificateData(`-----BEGIN CERTIFICATE-----
MIIBFzCBwgIJALhygXnxXmN1MA0GCSqGSIb3DQEBCwUAMBMxETAPBgNVBAMMCGhv
c3QtMTIzMB4XDTE4MTEwNDIzNTc1NFoXDTE4MTEwNTIzNTc1NFowEzERMA8GA1UE
AwwIaG9zdC0xMjMwXDANBgkqhkiG9w0BAQEFAANLADBIAkEAtBMa7NWpv3BVlKTC
PGO/LEsguKqWHBtKzweMY2CVtAL1rQm913huhxF9w+ai76KQ3MHK5IVnLJjYYA5M
zP2H5QIDAQABMA0GCSqGSIb3DQEBCwUAA0EAN2DPFUtCzqnidL+5nh+46Sk6dkMI
T5DD11UuuIjZusKvThsHKVCIsyJ2bDo7cTbI+/nklLRP+FcC2wESFUgXbA==
-----END CERTIFICATE-----`, `-----BEGIN RSA PRIVATE KEY-----
MIIBUwIBADANBgkqhkiG9w0BAQEFAASCAT0wggE5AgEAAkEAtBMa7NWpv3BVlKTC
PGO/LEsguKqWHBtKzweMY2CVtAL1rQm913huhxF9w+ai76KQ3MHK5IVnLJjYYA5M
zP2H5QIDAQABAkAS9BfXab3OKpK3bIgNNyp+DQJKrZnTJ4Q+OjsqkpXvNltPJosf
G8GsiKu/vAt4HGqI3eU77NvRI+mL4MnHRmXBAiEA3qM4FAtKSRBbcJzPxxLEUSwg
XSCcosCktbkXvpYrS30CIQDPDxgqlwDEJQ0uKuHkZI38/SPWWqfUmkecwlbpXABK
iQIgZX08DA8VfvcA5/Xj1Zjdey9FVY6POLXen6RPiabE97UCICp6eUW7ht+2jjar
e35EltCRCjoejRHTuN9TC0uCoVipAiAXaJIx/Q47vGwiw6Y8KXsNU6y54gTbOSxX
54LzHNk/+Q==
-----END RSA PRIVATE KEY-----`)
var bootstrapCertData = newCertificateData(
`-----BEGIN CERTIFICATE-----
MIICRzCCAfGgAwIBAgIJANXr+UzRFq4TMA0GCSqGSIb3DQEBCwUAMH4xCzAJBgNV
@ -388,8 +405,8 @@ func TestRotateCertCreateCSRError(t *testing.T) {
},
getTemplate: func() *x509.CertificateRequest { return &x509.CertificateRequest{} },
usages: []certificates.KeyUsage{},
certSigningRequestClient: fakeClient{
failureType: createError,
clientFn: func(_ *tls.Certificate) (certificatesclient.CertificateSigningRequestInterface, error) {
return fakeClient{failureType: createError}, nil
},
}
@ -411,8 +428,8 @@ func TestRotateCertWaitingForResultError(t *testing.T) {
},
getTemplate: func() *x509.CertificateRequest { return &x509.CertificateRequest{} },
usages: []certificates.KeyUsage{},
certSigningRequestClient: fakeClient{
failureType: watchError,
clientFn: func(_ *tls.Certificate) (certificatesclient.CertificateSigningRequestInterface, error) {
return fakeClient{failureType: watchError}, nil
},
}
@ -598,6 +615,14 @@ func TestInitializeCertificateSigningRequestClient(t *testing.T) {
expectedCertBeforeStart: storeCertData,
expectedCertAfterStart: storeCertData,
},
{
description: "Current certificate expired, no bootstrap certificate",
storeCert: expiredStoreCertData,
bootstrapCert: nilCertificate,
apiCert: apiServerCertData,
expectedCertBeforeStart: nil,
expectedCertAfterStart: apiServerCertData,
},
}
for _, tc := range testCases {
@ -621,19 +646,25 @@ func TestInitializeCertificateSigningRequestClient(t *testing.T) {
CertificateStore: certificateStore,
BootstrapCertificatePEM: tc.bootstrapCert.certificatePEM,
BootstrapKeyPEM: tc.bootstrapCert.keyPEM,
ClientFn: func(_ *tls.Certificate) (certificatesclient.CertificateSigningRequestInterface, error) {
return &fakeClient{
certificatePEM: tc.apiCert.certificatePEM,
}, nil
},
})
if err != nil {
t.Errorf("Got %v, wanted no error.", err)
}
certificate := certificateManager.Current()
if !certificatesEqual(certificate, tc.expectedCertBeforeStart.certificate) {
t.Errorf("Got %v, wanted %v", certificateString(certificate), certificateString(tc.expectedCertBeforeStart.certificate))
}
if err := certificateManager.SetCertificateSigningRequestClient(&fakeClient{
certificatePEM: tc.apiCert.certificatePEM,
}); err != nil {
t.Errorf("Got error %v, expected none.", err)
if tc.expectedCertBeforeStart == nil {
if certificate != nil {
t.Errorf("Expected certificate to be nil, was %s", certificate.Leaf.NotAfter)
}
} else {
if !certificatesEqual(certificate, tc.expectedCertBeforeStart.certificate) {
t.Errorf("Got %v, wanted %v", certificateString(certificate), certificateString(tc.expectedCertBeforeStart.certificate))
}
}
if m, ok := certificateManager.(*manager); !ok {
@ -649,6 +680,12 @@ func TestInitializeCertificateSigningRequestClient(t *testing.T) {
}
certificate = certificateManager.Current()
if tc.expectedCertAfterStart == nil {
if certificate != nil {
t.Errorf("Expected certificate to be nil, was %s", certificate.Leaf.NotAfter)
}
return
}
if !certificatesEqual(certificate, tc.expectedCertAfterStart.certificate) {
t.Errorf("Got %v, wanted %v", certificateString(certificate), certificateString(tc.expectedCertAfterStart.certificate))
}
@ -721,8 +758,10 @@ func TestInitializeOtherRESTClients(t *testing.T) {
CertificateStore: certificateStore,
BootstrapCertificatePEM: tc.bootstrapCert.certificatePEM,
BootstrapKeyPEM: tc.bootstrapCert.keyPEM,
CertificateSigningRequestClient: &fakeClient{
certificatePEM: tc.apiCert.certificatePEM,
ClientFn: func(_ *tls.Certificate) (certificatesclient.CertificateSigningRequestInterface, error) {
return &fakeClient{
certificatePEM: tc.apiCert.certificatePEM,
}, nil
},
})
if err != nil {
@ -873,10 +912,12 @@ func TestServerHealth(t *testing.T) {
CertificateStore: certificateStore,
BootstrapCertificatePEM: tc.bootstrapCert.certificatePEM,
BootstrapKeyPEM: tc.bootstrapCert.keyPEM,
CertificateSigningRequestClient: &fakeClient{
certificatePEM: tc.apiCert.certificatePEM,
failureType: tc.failureType,
err: tc.clientErr,
ClientFn: func(_ *tls.Certificate) (certificatesclient.CertificateSigningRequestInterface, error) {
return &fakeClient{
certificatePEM: tc.apiCert.certificatePEM,
failureType: tc.failureType,
err: tc.clientErr,
}, nil
},
})
if err != nil {