Merge pull request #80 from carlbraganza/default-sa-in-lister

The snapshot iterator now determines the ServiceAccount if unspecified.

examples/snapshot-metadata-lister/main.go

@@ -50,6 +50,9 @@ If a previous VolumeSnapshot object is also specified then the metadata
 describes the content changes between the two snapshots, which must both
 be from the same PersistentVolume.
 
+The command is usually invoked in a Pod in the cluster, as the gRPC client
+needs to resolve the DNS address in the SnapshotMetadataService CR.
+
 ` + shortUsageFmt + `
 
 Flags:
@@ -80,7 +83,7 @@ func parseFlags() {
 		flag.StringVar(&kubeConfig, "kubeconfig", "", "Path to the kubeconfig file.")
 	}
 
-	flag.StringVar(&args.ServiceAccount, "service-account", "default", "ServiceAccount used to create a security token.")
+	flag.StringVar(&args.ServiceAccount, "service-account", "", "ServiceAccount used to create a security token. If unspecified the ServiceAccount of the Pod in which the command is invoked will be used.")
 
 	flag.Int64Var(&args.TokenExpirySecs, "token-expiry", 600, "Expiry time in seconds for the security token.")
 	flag.Int64Var(&args.StartingOffset, "starting-offset", 0, "The starting byte offset.")

pkg/iterator/common_test.go

@@ -61,6 +61,10 @@ type testHarness struct {
 	InSnapshotMetadataIteratorDoneNR int
 
 	// fake helpers
+	CalledGetDefaultServiceAccount bool
+	RetGetDefaultServiceAccount    string
+	RetGetDefaultServiceAccountErr error
+
 	CalledGetCSIDriverFromPrimarySnapshot bool
 	RetGetCSIDriverFromPrimarySnapshot    string
 	RetGetCSIDriverFromPrimarySnapshotErr error
@@ -69,6 +73,7 @@ type testHarness struct {
 	RetGetSnapshotMetadataServiceCRService  *smsCRv1alpha1.SnapshotMetadataService
 	RetGetSnapshotMetadataServiceCRErr      error
 
+	InCreateSecurityTokenSA       string
 	InCreateSecurityTokenAudience string
 	RetCreateSecurityToken        string
 	RetCreateSecurityTokenErr     error
@@ -169,6 +174,16 @@ func (th *testHarness) FakeVS() (*snapshotv1.VolumeSnapshot, *snapshotv1.VolumeS
 	return vs, vsc
 }
 
+func (th *testHarness) FakeAuthSelfSubjectReview() *authv1.SelfSubjectReview {
+	return &authv1.SelfSubjectReview{
+		Status: authv1.SelfSubjectReviewStatus{
+			UserInfo: authv1.UserInfo{
+				Username: K8sServiceAccountUserNamePrefix + th.Namespace + ":" + th.ServiceAccount,
+			},
+		},
+	}
+}
+
 func (th *testHarness) FakeTokenRequest() *authv1.TokenRequest {
 	expirySecs := th.Args().TokenExpirySecs
 	return &authv1.TokenRequest{
@@ -236,6 +251,11 @@ func (th *testHarness) SnapshotMetadataIteratorDone(numberRecords int) {
 }
 
 // fake helpers
+func (th *testHarness) getDefaultServiceAccount(ctx context.Context) (string, error) {
+	th.CalledGetDefaultServiceAccount = true
+	return th.RetGetDefaultServiceAccount, th.RetGetDefaultServiceAccountErr
+}
+
 func (th *testHarness) getCSIDriverFromPrimarySnapshot(ctx context.Context) (string, error) {
 	th.CalledGetCSIDriverFromPrimarySnapshot = true
 	return th.RetGetCSIDriverFromPrimarySnapshot, th.RetGetCSIDriverFromPrimarySnapshotErr
@@ -246,7 +266,8 @@ func (th *testHarness) getSnapshotMetadataServiceCR(ctx context.Context, csiDriv
 	return th.RetGetSnapshotMetadataServiceCRService, th.RetGetSnapshotMetadataServiceCRErr
 }
 
-func (th *testHarness) createSecurityToken(ctx context.Context, audience string) (string, error) {
+func (th *testHarness) createSecurityToken(ctx context.Context, serviceAccount, audience string) (string, error) {
+	th.InCreateSecurityTokenSA = serviceAccount
 	th.InCreateSecurityTokenAudience = audience
 	return th.RetCreateSecurityToken, th.RetCreateSecurityTokenErr
 }

pkg/iterator/iter.go

@@ -23,6 +23,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"strings"
 
 	"google.golang.org/grpc"
 	grpcCreds "google.golang.org/grpc/credentials"
@@ -39,7 +40,15 @@ var (
 	ErrCancelled   = errors.New("enumeration cancelled")
 )
 
-const DefaultTokenExpirySeconds = int64(600)
+const (
+	DefaultTokenExpirySeconds = int64(600)
+
+	// See "Service account tokens" in
+	// https://kubernetes.io/docs/reference/access-authn-authz/authentication/.
+	// It turns out that a service account name starts with a well defined prefix
+	// that is guaranteed not to match any other user name.
+	K8sServiceAccountUserNamePrefix = "system:serviceaccount:"
+)
 
 // GetSnapshotMetadata enumerates either the allocated blocks of a
 // VolumeSnapshot object, or the blocks changed between a pair of
@@ -92,6 +101,7 @@ type Args struct {
 
 	// ServiceAccount is used to construct a security token
 	// with the audience string from the SnapshotMetadataService CR.
+	// If unspecified the default for the given client will be used.
 	ServiceAccount string
 
 	// TokenExpirySecs specifies the time in seconds after which the
@@ -108,8 +118,6 @@ func (a Args) Validate() error {
 		return fmt.Errorf("%w: missing Namespace", ErrInvalidArgs)
 	case a.SnapshotName == "":
 		return fmt.Errorf("%w: missing SnapshotName", ErrInvalidArgs)
-	case a.ServiceAccount == "":
-		return fmt.Errorf("%w: missing ServiceAccount", ErrInvalidArgs)
 	case a.TokenExpirySecs < 0:
 		return fmt.Errorf("%w: invalid TokenExpirySecs", ErrInvalidArgs)
 	case a.MaxResults < 0:
@@ -154,8 +162,9 @@ type iterator struct {
 
 type iteratorHelpers interface {
 	getCSIDriverFromPrimarySnapshot(ctx context.Context) (string, error)
+	getDefaultServiceAccount(ctx context.Context) (string, error)
 	getSnapshotMetadataServiceCR(ctx context.Context, csiDriver string) (*smsCRv1alpha1.SnapshotMetadataService, error)
-	createSecurityToken(ctx context.Context, audience string) (string, error)
+	createSecurityToken(ctx context.Context, serviceAccount, audience string) (string, error)
 	getGRPCClient(caCert []byte, URL string) (api.SnapshotMetadataClient, error)
 	getAllocatedBlocks(ctx context.Context, grpcClient api.SnapshotMetadataClient, securityToken string) error
 	getChangedBlocks(ctx context.Context, grpcClient api.SnapshotMetadataClient, securityToken string) error
@@ -182,6 +191,14 @@ func newIterator(args Args) *iterator {
 func (iter *iterator) run(ctx context.Context) error {
 	var err error
 
+	serviceAccount := iter.ServiceAccount // optional field
+	if serviceAccount == "" {
+		serviceAccount, err = iter.h.getDefaultServiceAccount(ctx)
+		if err != nil {
+			return err
+		}
+	}
+
 	csiDriver := iter.CSIDriver // optional field
 	if csiDriver == "" {
 		if csiDriver, err = iter.h.getCSIDriverFromPrimarySnapshot(ctx); err != nil {
@@ -196,7 +213,7 @@ func (iter *iterator) run(ctx context.Context) error {
 	}
 
 	// get the security token to use in the API
-	securityToken, err := iter.h.createSecurityToken(ctx, smsCR.Spec.Audience)
+	securityToken, err := iter.h.createSecurityToken(ctx, serviceAccount, smsCR.Spec.Audience)
 	if err != nil {
 		return err
 	}
@@ -225,6 +242,22 @@ func (iter *iterator) run(ctx context.Context) error {
 	return err
 }
 
+func (iter *iterator) getDefaultServiceAccount(ctx context.Context) (string, error) {
+	ssr, err := iter.KubeClient.AuthenticationV1().SelfSubjectReviews().Create(ctx, &authv1.SelfSubjectReview{}, apimetav1.CreateOptions{})
+	if err != nil {
+		return "", fmt.Errorf("SelfSubjectReviews.Create(): %w", err)
+	}
+
+	if strings.HasPrefix(ssr.Status.UserInfo.Username, K8sServiceAccountUserNamePrefix) {
+		fields := strings.Split(ssr.Status.UserInfo.Username, ":")
+		if len(fields) == 4 {
+			return fields[3], nil
+		}
+	}
+
+	return "", fmt.Errorf("%w: ServiceAccount unspecified and default cannot be determined", ErrInvalidArgs)
+}
+
 // getCSIDriverFromPrimarySnapshot loads the bound VolumeSnapshotContent
 // of the VolumeSnapshot identified by SnapshotName to fetch the CSI driver.
 func (iter *iterator) getCSIDriverFromPrimarySnapshot(ctx context.Context) (string, error) {
@@ -258,33 +291,33 @@ func (iter *iterator) getSnapshotMetadataServiceCR(ctx context.Context, csiDrive
 
 // createSecurityToken will create a security token for the specified storage
 // account using the audience string from the SnapshotMetadataService CR.
-func (iter *iterator) createSecurityToken(ctx context.Context, audience string) (string, error) {
+func (iter *iterator) createSecurityToken(ctx context.Context, serviceAccount, audience string) (string, error) {
 	tokenRequest := authv1.TokenRequest{
 		Spec: authv1.TokenRequestSpec{
 			Audiences:         []string{audience},
 			ExpirationSeconds: &iter.TokenExpirySecs,
 		},
 	}
 
-	tokenResp, err := iter.KubeClient.CoreV1().ServiceAccounts(iter.Namespace).CreateToken(ctx, iter.ServiceAccount, &tokenRequest, apimetav1.CreateOptions{})
+	tokenResp, err := iter.KubeClient.CoreV1().ServiceAccounts(iter.Namespace).CreateToken(ctx, serviceAccount, &tokenRequest, apimetav1.CreateOptions{})
 	if err != nil {
-		return "", fmt.Errorf("ServiceAccounts.CreateToken(%s): %v", iter.ServiceAccount, err)
+		return "", fmt.Errorf("ServiceAccounts.CreateToken(%s): %v", serviceAccount, err)
 	}
 
 	return tokenResp.Status.Token, nil
 }
 
-func (iter *iterator) getGRPCClient(caCert []byte, URL string) (api.SnapshotMetadataClient, error) {
+func (iter *iterator) getGRPCClient(caCert []byte, url string) (api.SnapshotMetadataClient, error) {
 	// Add the CA to the cert pool
 	certPool := x509.NewCertPool()
 	if !certPool.AppendCertsFromPEM(caCert) {
 		return nil, ErrCACert
 	}
 
 	tlsCredentials := grpcCreds.NewTLS(&tls.Config{RootCAs: certPool})
-	conn, err := grpc.NewClient(URL, grpc.WithTransportCredentials(tlsCredentials))
+	conn, err := grpc.NewClient(url, grpc.WithTransportCredentials(tlsCredentials))
 	if err != nil {
-		return nil, fmt.Errorf("grpc.NewClient(%s): %w", URL, err)
+		return nil, fmt.Errorf("grpc.NewClient(%s): %w", url, err)
 	}
 
 	return api.NewSnapshotMetadataClient(conn), nil

pkg/iterator/iter_test.go

@@ -61,12 +61,7 @@ func TestValidateArgs(t *testing.T) {
 	err = args.Validate()
 	assert.Error(t, err)
 	assert.ErrorIs(t, err, ErrInvalidArgs)
-	assert.ErrorContains(t, err, "ServiceAccount")
-
-	args.ServiceAccount = "service-account"
-	err = args.Validate()
-	assert.Error(t, err)
-	assert.ErrorIs(t, err, ErrInvalidArgs)
+	// assert.ErrorContains(t, err, "ServiceAccount")
 	assert.ErrorContains(t, err, "KubeClient")
 
 	args.Clients.KubeClient = fake.NewSimpleClientset()
@@ -123,23 +118,27 @@ func TestNewIterator(t *testing.T) {
 func TestRun(t *testing.T) {
 	testErr := errors.New("test-error")
 
-	t.Run("get-changed-blocks-no-csi-driver", func(t *testing.T) {
+	t.Run("get-changed-blocks-no-csi-driver-no-sa", func(t *testing.T) {
 		th := newTestHarness()
 		th.RetGetCSIDriverFromPrimarySnapshot = th.CSIDriver
 		th.RetGetSnapshotMetadataServiceCRService = th.FakeCR()
 		th.RetGetGRPCClient = th.GRPCSnapshotMetadataClient(t)
 		th.RetCreateSecurityToken = "security-token"
+		th.RetGetDefaultServiceAccount = th.ServiceAccount
 
 		iter := th.NewTestIterator()
 		iter.recordNum = 100
+		iter.ServiceAccount = ""
 		assert.NotEmpty(t, iter.PrevSnapshotName) // changed block flow
 
 		err := iter.run(context.Background())
 		assert.NoError(t, err)
 
 		// check data passed through the helpers
+		assert.True(t, th.CalledGetDefaultServiceAccount)
 		assert.True(t, th.CalledGetCSIDriverFromPrimarySnapshot)
 		assert.Equal(t, th.CSIDriver, th.InGetSnapshotMetadataServiceCRCSIDriver)
+		assert.Equal(t, th.ServiceAccount, th.InCreateSecurityTokenSA)
 		assert.Equal(t, th.Audience, th.InCreateSecurityTokenAudience)
 		assert.Equal(t, th.CACert, th.InGetGRPCClientCA)
 		assert.Equal(t, th.Address, th.InGetGRPCClientURL)
@@ -298,6 +297,49 @@ func TestRun(t *testing.T) {
 	})
 }
 
+func TestGetDefaultServiceAccount(t *testing.T) {
+	t.Run("self-subject-review-err", func(t *testing.T) {
+		th := newTestHarness()
+		args := th.Args()
+		args.ServiceAccount = ""
+
+		// invoke via GetSnapshotMetadata directly to cover that code path
+		err := GetSnapshotMetadata(context.Background(), args)
+		assert.Error(t, err)
+		assert.ErrorContains(t, err, "SelfSubjectReviews.Create")
+	})
+
+	t.Run("not-a-service-account", func(t *testing.T) {
+		th := newTestHarness()
+		iter := th.NewTestIterator()
+
+		th.FakeKubeClient.PrependReactor("create", "selfsubjectreviews", func(action clientgotesting.Action) (handled bool, ret apiruntime.Object, err error) {
+			ssr := th.FakeAuthSelfSubjectReview()
+			ssr.Status.UserInfo.Username += ":additionalfield"
+			return true, ssr, nil
+		})
+
+		sa, err := iter.getDefaultServiceAccount(context.Background())
+		assert.Error(t, err)
+		assert.ErrorIs(t, err, ErrInvalidArgs)
+		assert.Empty(t, sa)
+	})
+
+	t.Run("success", func(t *testing.T) {
+		th := newTestHarness()
+		iter := th.NewTestIterator()
+
+		th.FakeKubeClient.PrependReactor("create", "selfsubjectreviews", func(action clientgotesting.Action) (handled bool, ret apiruntime.Object, err error) {
+			ssr := th.FakeAuthSelfSubjectReview()
+			return true, ssr, nil
+		})
+
+		sa, err := iter.getDefaultServiceAccount(context.Background())
+		assert.NoError(t, err)
+		assert.Equal(t, th.ServiceAccount, sa)
+	})
+}
+
 func TestGetCSIDriverFromPrimarySnapshot(t *testing.T) {
 	t.Run("snapshot-get-err", func(t *testing.T) {
 		th := newTestHarness()
@@ -425,7 +467,7 @@ func TestCreateSecurityToken(t *testing.T) {
 		th := newTestHarness()
 		iter := th.NewTestIterator()
 
-		securityToken, err := iter.createSecurityToken(context.Background(), th.Audience)
+		securityToken, err := iter.createSecurityToken(context.Background(), th.ServiceAccount, th.Audience)
 		assert.Error(t, err)
 		assert.ErrorContains(t, err, "ServiceAccounts.CreateToken")
 		assert.Empty(t, securityToken)
@@ -439,7 +481,7 @@ func TestCreateSecurityToken(t *testing.T) {
 			return true, th.FakeTokenRequest(), nil
 		})
 
-		securityToken, err := iter.createSecurityToken(context.Background(), th.Audience)
+		securityToken, err := iter.createSecurityToken(context.Background(), th.ServiceAccount, th.Audience)
 		assert.NoError(t, err)
 		assert.Equal(t, th.SecurityToken, securityToken)
 	})