AGENT-903: monitor-add-nodes should only show CSRs matching node

The first and second CSRs pending approval have the node name
(hostname) embedded in their specs. monitor-add-nodes should only
show CSRs pending approval for a specific node. Currently it shows
all CSRs pending approval for all nodes.

If the IP address of the node cannot be resolved to a hostname,
we will not be able to determine if there are any CSRs pending
approval for that node. The monitoring command will skip showing
CSRs pending approval. In this case, users can still approve the
CSRs, and the monitoring command will continue to check if the node
has joined the cluster and has become Ready.

Author: rwsu

cmd/node-joiner/main.go

@@ -34,9 +34,9 @@ func main() {
 		Use:   "monitor-add-nodes",
 		Short: "Monitors the configured nodes while they are joining an existing cluster",
 		RunE: func(cmd *cobra.Command, args []string) error {
-			wd, err := os.Getwd()
+			dir, err := cmd.Flags().GetString("dir")
 			if err != nil {
-				logrus.Fatal(err)
+				return err
 			}
 
 			kubeConfig, err := cmd.Flags().GetString("kubeconfig")
@@ -49,7 +49,7 @@ func main() {
 			if len(ips) == 0 {
 				logrus.Fatal("At least one IP address must be specified")
 			}
-			return nodejoiner.NewMonitorAddNodesCommand(wd, kubeConfig, ips)
+			return nodejoiner.NewMonitorAddNodesCommand(dir, kubeConfig, ips)
 		},
 	}
 

pkg/agent/kube.go

@@ -104,21 +104,3 @@ func (kube *ClusterKubeAPIClient) ListCSRs() (*certificatesv1.CertificateSigning
 	}
 	return csrs, nil
 }
-
-func (kube *ClusterKubeAPIClient) getCSRsPendingApproval(signerName string) []certificatesv1.CertificateSigningRequest {
-	csrs, err := kube.ListCSRs()
-	if err != nil {
-		logrus.Debugf("error calling listCSRs(): %v ", err)
-	}
-
-	var matchedCSRs []certificatesv1.CertificateSigningRequest
-	for _, csr := range csrs.Items {
-		if len(csr.Status.Conditions) == 0 {
-			// CSR is Pending and awaiting approval
-			if csr.Spec.SignerName == signerName {
-				matchedCSRs = append(matchedCSRs, csr)
-			}
-		}
-	}
-	return matchedCSRs
-}

pkg/agent/monitoraddnodes.go

@@ -2,6 +2,8 @@ package agent
 
 import (
 	"context"
+	"crypto/x509"
+	"encoding/pem"
 	"fmt"
 	"net"
 	"net/http"
@@ -10,6 +12,7 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
+	certificatesv1 "k8s.io/api/certificates/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 )
@@ -34,9 +37,13 @@ type addNodeMonitor struct {
 	status        addNodeStatusHistory
 }
 
-func newAddNodeMonitor(nodeIP string, cluster *Cluster) *addNodeMonitor {
+func newAddNodeMonitor(nodeIP string, cluster *Cluster) (*addNodeMonitor, error) {
+	parsedIPAddress := net.ParseIP(nodeIP)
+	if parsedIPAddress == nil {
+		return nil, fmt.Errorf("%s is not valid IP Address", nodeIP)
+	}
 	mon := addNodeMonitor{
-		nodeIPAddress: nodeIP,
+		nodeIPAddress: parsedIPAddress.String(),
 		cluster:       cluster,
 		status: addNodeStatusHistory{
 			RestAPISeen:            false,
@@ -47,7 +54,7 @@ func newAddNodeMonitor(nodeIP string, cluster *Cluster) *addNodeMonitor {
 			NodeIsReady:            false,
 		},
 	}
-	return &mon
+	return &mon, nil
 }
 
 func (mon *addNodeMonitor) logStatus(status string) {
@@ -57,16 +64,14 @@ func (mon *addNodeMonitor) logStatus(status string) {
 // MonitorAddNodes waits for the a node to be added to the cluster
 // and reports its status until it becomes Ready.
 func MonitorAddNodes(cluster *Cluster, nodeIPAddress string) error {
-	parsedIPAddress := net.ParseIP(nodeIPAddress)
-	if parsedIPAddress == nil {
-		return fmt.Errorf("%s is not valid IP Address", nodeIPAddress)
-	}
-
 	timeout := 90 * time.Minute
 	waitContext, cancel := context.WithTimeout(cluster.Ctx, timeout)
 	defer cancel()
 
-	mon := newAddNodeMonitor(nodeIPAddress, cluster)
+	mon, err := newAddNodeMonitor(nodeIPAddress, cluster)
+	if err != nil {
+		return err
+	}
 
 	wait.Until(func() {
 		if !mon.status.RestAPISeen &&
@@ -180,36 +185,65 @@ func (mon *addNodeMonitor) nodeHasJoinedClusterAndIsReady() (bool, bool, error)
 }
 
 func (mon *addNodeMonitor) logCSRsPendingApproval(signerName string) {
-	// TODO? The CSRs have no IP address to identify for which
-	// node it is for, so it is possible to log CSRs pending for
-	// other nodes.
+	csrsPendingApproval := mon.getCSRsPendingApproval(signerName)
+
+	for _, csr := range csrsPendingApproval {
+		mon.logStatus(fmt.Sprintf("CSR %s with signerName %s and username %s is Pending and awaiting approval",
+			csr.Name, csr.Spec.SignerName, csr.Spec.Username))
+	}
+}
+
+func (mon *addNodeMonitor) clusterHasFirstCSRPending() bool {
+	return len(mon.getCSRsPendingApproval(firstCSRSignerName)) > 0
+}
+
+func (mon *addNodeMonitor) clusterHasSecondCSRPending() bool {
+	return len(mon.getCSRsPendingApproval(secondCSRSignerName)) > 0
+}
+
+func (mon *addNodeMonitor) getCSRsPendingApproval(signerName string) []certificatesv1.CertificateSigningRequest {
+	hostnames, err := net.LookupAddr(mon.nodeIPAddress)
+	if err != nil {
+		logrus.Debugf("error looking up hostnames from IP address: %v", err)
+		return []certificatesv1.CertificateSigningRequest{}
+	}
+
 	csrs, err := mon.cluster.API.Kube.ListCSRs()
 	if err != nil {
-		logrus.Debugf("error calling listCSRs(): %v ", err)
+		logrus.Debugf("error calling listCSRs(): %v", err)
+		return []certificatesv1.CertificateSigningRequest{}
 	}
 
-	for _, csr := range csrs.Items {
-		if len(csr.Status.Conditions) == 0 {
-			// CSR is Pending and awaiting approval
-			if signerName != "" && signerName != csr.Spec.SignerName {
-				continue
-			}
+	return filterCSRsMatchingHostname(signerName, csrs, hostnames)
+}
 
-			logrus.Infof("CSR %s with signerName %s and username %s is Pending and awaiting approval",
-				csr.Name, csr.Spec.SignerName, csr.Spec.Username)
+func filterCSRsMatchingHostname(signerName string, csrs *certificatesv1.CertificateSigningRequestList, hostnames []string) []certificatesv1.CertificateSigningRequest {
+	matchedCSRs := []certificatesv1.CertificateSigningRequest{}
+	for _, csr := range csrs.Items {
+		if len(csr.Status.Conditions) > 0 {
+			// CSR is not Pending and not awaiting approval
+			continue
+		}
+		if signerName == firstCSRSignerName && csr.Spec.SignerName == firstCSRSignerName &&
+			containsHostname(decodedFirstCSRSubject(csr.Spec.Request), hostnames) {
+			matchedCSRs = append(matchedCSRs, csr)
+		}
+		if signerName == secondCSRSignerName && csr.Spec.SignerName == secondCSRSignerName &&
+			containsHostname(csr.Spec.Username, hostnames) {
+			matchedCSRs = append(matchedCSRs, csr)
 		}
 	}
+	return matchedCSRs
 }
 
-func (mon *addNodeMonitor) clusterHasFirstCSRPending() bool {
-	return len(mon.cluster.API.Kube.getCSRsPendingApproval(firstCSRSignerName)) > 0
-}
-
-func (mon *addNodeMonitor) clusterHasSecondCSRPending() bool {
-	// TODO: the csr.Spec.Username contains the node name
-	// can we use it as an additional filter to only show
-	// those matching mon.nodeIPAddress?
-	return len(mon.cluster.API.Kube.getCSRsPendingApproval(secondCSRSignerName)) > 0
+func containsHostname(searchString string, hostnames []string) bool {
+	for _, hostname := range hostnames {
+		parts := strings.Split(hostname, ".")
+		if strings.Contains(searchString, parts[0]) {
+			return true
+		}
+	}
+	return false
 }
 
 // isKubeletRunningOnNode checks if kubelet responds
@@ -237,3 +271,41 @@ func (mon *addNodeMonitor) isKubeletRunningOnNode() bool {
 	}
 	return false
 }
+
+// decodedFirstCSRSubject decodes the CSR.Spec.Request PEM block
+// into readable output and returns the subject as string.
+//
+// Example of decoded request:
+// Certificate Request:
+// Data:
+// Version: 1 (0x0)
+// Subject: O = system:nodes, CN = system:node:extraworker-1
+// Subject Public Key Info:
+//
+//	Public Key Algorithm: id-ecPublicKey
+//		Public-Key: (256 bit)
+//		pub:
+//			*snip*
+//		ASN1 OID: prime256v1
+//		NIST CURVE: P-256
+//
+// Attributes:
+//
+//	a0:00
+//
+// Signature Algorithm: ecdsa-with-SHA256
+//
+//	*snip*
+func decodedFirstCSRSubject(request []byte) string {
+	block, _ := pem.Decode(request)
+	if block == nil {
+		return ""
+	}
+	csrDER := block.Bytes
+	decodedRequest, err := x509.ParseCertificateRequest(csrDER)
+	if err != nil {
+		logrus.Warn("error in x509.ParseCertificateRequest(csrDER)")
+		return ""
+	}
+	return decodedRequest.Subject.String()
+}

pkg/agent/monitoraddnodes_test.go

@@ -0,0 +1,161 @@
+package agent
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	certificatesv1 "k8s.io/api/certificates/v1"
+)
+
+func TestDecodedFirstCSRSubjectContainsHostname(t *testing.T) {
+	firstCSRRequestForExtraworker0 := "-----BEGIN CERTIFICATE REQUEST-----\nMIH3MIGdAgEAMDsxFTATBgNVBAoTDHN5c3RlbTpub2RlczEiMCAGA1UEAxMZc3lz\ndGVtOm5vZGU6ZXh0cmF3b3JrZXItMDBZMBMGByqGSM49AgEGCCqGSM49AwEHA0IA\nBGaK3U+3X3lM6tdgjD2b/y7Kysws8xgFW1rNd/wvKEvXzP5+A1K1M38zJiAWqKXP\n5AL2IDklO4GaO7PcRDNPabigADAKBggqhkjOPQQDAgNJADBGAiEA7C33Nym0Go73\nCZY+XOmyqE/IhaBMSwign+fgbPX1ibkCIQDHIfF7QpZReF93IW0v864/yLoXKyXy\nTGygkuR4KtXTDw==\n-----END CERTIFICATE REQUEST-----\n"
+	tests := []struct {
+		name           string
+		hostnames      []string
+		request        string
+		expectedResult bool
+	}{
+		{
+			name:           "request contains hostname",
+			hostnames:      []string{"extraworker-0"},
+			request:        firstCSRRequestForExtraworker0,
+			expectedResult: true,
+		},
+		{
+			name:           "request contains hostname using FQDN",
+			hostnames:      []string{"extraworker-0.ostest.test.metalkube.org"},
+			request:        firstCSRRequestForExtraworker0,
+			expectedResult: true,
+		},
+		{
+			name:           "request contains hostname when multiple names are resolved",
+			hostnames:      []string{"somename", "extraworker-0.ostest.test.metalkube.org"},
+			request:        firstCSRRequestForExtraworker0,
+			expectedResult: true,
+		},
+		{
+			name:           "request does not contain hostname",
+			hostnames:      []string{"extraworker-1"},
+			request:        firstCSRRequestForExtraworker0,
+			expectedResult: false,
+		},
+		{
+			name:           "request is empty string",
+			hostnames:      []string{"hostname-not-specified"},
+			request:        "",
+			expectedResult: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			containsHostname := containsHostname(decodedFirstCSRSubject([]byte(tt.request)), tt.hostnames)
+			assert.Equal(t, tt.expectedResult, containsHostname)
+		})
+	}
+}
+
+func TestFilterCSRsMatchingHostnames(t *testing.T) {
+	firstCSRRequestForExtraworker0 := "-----BEGIN CERTIFICATE REQUEST-----\nMIH3MIGdAgEAMDsxFTATBgNVBAoTDHN5c3RlbTpub2RlczEiMCAGA1UEAxMZc3lz\ndGVtOm5vZGU6ZXh0cmF3b3JrZXItMDBZMBMGByqGSM49AgEGCCqGSM49AwEHA0IA\nBGaK3U+3X3lM6tdgjD2b/y7Kysws8xgFW1rNd/wvKEvXzP5+A1K1M38zJiAWqKXP\n5AL2IDklO4GaO7PcRDNPabigADAKBggqhkjOPQQDAgNJADBGAiEA7C33Nym0Go73\nCZY+XOmyqE/IhaBMSwign+fgbPX1ibkCIQDHIfF7QpZReF93IW0v864/yLoXKyXy\nTGygkuR4KtXTDw==\n-----END CERTIFICATE REQUEST-----\n"
+
+	tests := []struct {
+		name           string
+		csrs           *certificatesv1.CertificateSigningRequestList
+		hostnames      []string
+		signerName     string
+		expectedResult []certificatesv1.CertificateSigningRequest
+	}{
+		{
+			name: "first CSR filtering",
+			csrs: &certificatesv1.CertificateSigningRequestList{
+				Items: []certificatesv1.CertificateSigningRequest{
+					{
+						// should match only this one
+						Spec: certificatesv1.CertificateSigningRequestSpec{
+							SignerName: firstCSRSignerName,
+							Request:    []byte(firstCSRRequestForExtraworker0),
+						},
+					},
+					{
+						Spec: certificatesv1.CertificateSigningRequestSpec{
+							SignerName: "other-request",
+							Request:    []byte("other-request"),
+						},
+					},
+				},
+			},
+			hostnames:  []string{"extraworker-0.ostest.test.metalkube.org"},
+			signerName: "kubernetes.io/kube-apiserver-client-kubelet",
+			expectedResult: []certificatesv1.CertificateSigningRequest{
+				{
+					Spec: certificatesv1.CertificateSigningRequestSpec{
+						SignerName: "kubernetes.io/kube-apiserver-client-kubelet",
+						Request:    []byte(firstCSRRequestForExtraworker0),
+					},
+				},
+			},
+		},
+		{
+			name: "second CSR filtering",
+			csrs: &certificatesv1.CertificateSigningRequestList{
+				Items: []certificatesv1.CertificateSigningRequest{
+					{
+						// should match only this one
+						Spec: certificatesv1.CertificateSigningRequestSpec{
+							SignerName: secondCSRSignerName,
+							Username:   "system:node:extraworker-0",
+							Request:    []byte("something"),
+						},
+					},
+					{
+						Spec: certificatesv1.CertificateSigningRequestSpec{
+							SignerName: secondCSRSignerName,
+							Username:   "system:node:extraworker-1",
+							Request:    []byte("something"),
+						},
+					},
+					{
+						Spec: certificatesv1.CertificateSigningRequestSpec{
+							SignerName: "other-request",
+							Request:    []byte("other-request"),
+						},
+					},
+				},
+			},
+			hostnames:  []string{"extraworker-0.ostest.test.metalkube.org"},
+			signerName: secondCSRSignerName,
+			expectedResult: []certificatesv1.CertificateSigningRequest{
+				{
+					Spec: certificatesv1.CertificateSigningRequestSpec{
+						SignerName: "kubernetes.io/kubelet-serving",
+						Username:   "system:node:extraworker-0",
+						Request:    []byte("something"),
+					},
+				},
+			},
+		},
+		{
+			name: "no CSRs should not result in error",
+			csrs: &certificatesv1.CertificateSigningRequestList{
+				Items: []certificatesv1.CertificateSigningRequest{},
+			},
+			hostnames:      []string{"extraworker-0.ostest.test.metalkube.org"},
+			signerName:     secondCSRSignerName,
+			expectedResult: []certificatesv1.CertificateSigningRequest{},
+		},
+		{
+			name: "no hostnames should not result in error",
+			csrs: &certificatesv1.CertificateSigningRequestList{
+				Items: []certificatesv1.CertificateSigningRequest{},
+			},
+			hostnames:      []string{},
+			signerName:     secondCSRSignerName,
+			expectedResult: []certificatesv1.CertificateSigningRequest{},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			filteredCSRs := filterCSRsMatchingHostname(tt.signerName, tt.csrs, tt.hostnames)
+			assert.Equal(t, tt.expectedResult, filteredCSRs)
+		})
+	}
+}