secrets: Fix vault CSI support

In Vault CSI the secret files are actually written under `..data`
subdirectory which is also a symlink, and fs.WalkDir will skip symlinks
unless they are the root.

So walk under `..data` instead, and also strip the prefix from the path
key. Also add a test to test the rotation case and some improvements on
existing tests.

go.mod

@@ -21,10 +21,11 @@ require (
 	github.com/prometheus/client_model v0.2.0
 	github.com/sony/gobreaker v0.4.1
 	go.uber.org/automaxprocs v1.5.1
-	go.uber.org/zap v1.15.0
-	golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a
-	google.golang.org/grpc v1.41.0
+	go.uber.org/zap v1.21.0
+	golang.org/x/sys v0.3.0
+	google.golang.org/grpc v1.47.0
 	gopkg.in/yaml.v2 v2.4.0
+	sigs.k8s.io/secrets-store-csi-driver v1.3.0
 )
 
 require (
@@ -40,6 +41,7 @@ require (
 	github.com/eapache/queue v1.1.0 // indirect
 	github.com/garyburd/redigo v1.6.2 // indirect
 	github.com/go-logfmt/logfmt v0.5.1 // indirect
+	github.com/go-logr/logr v1.2.3 // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/golang/snappy v0.0.3 // indirect
 	github.com/hashicorp/go-uuid v1.0.2 // indirect
@@ -49,7 +51,7 @@ require (
 	github.com/jcmturner/gokrb5/v8 v8.4.2 // indirect
 	github.com/jcmturner/rpc/v2 v2.0.3 // indirect
 	github.com/klauspost/compress v1.12.2 // indirect
-	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 // indirect
 	github.com/mediocregopher/radix.v2 v0.0.0-20181115013041-b67df6e626f9 // indirect
 	github.com/pierrec/lz4 v2.6.0+incompatible // indirect
 	github.com/prometheus/common v0.37.0 // indirect
@@ -59,15 +61,15 @@ require (
 	go.opentelemetry.io/otel v0.20.0 // indirect
 	go.opentelemetry.io/otel/metric v0.20.0 // indirect
 	go.opentelemetry.io/otel/trace v0.20.0 // indirect
-	go.uber.org/atomic v1.6.0 // indirect
-	go.uber.org/multierr v1.5.0 // indirect
-	golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e // indirect
-	golang.org/x/net v0.0.0-20220225172249-27dd8689420f // indirect
-	golang.org/x/text v0.3.7 // indirect
-	golang.org/x/tools v0.1.3-0.20210608163600-9ed039809d4c // indirect
-	google.golang.org/genproto v0.0.0-20200825200019-8632dd797987 // indirect
+	go.uber.org/atomic v1.7.0 // indirect
+	go.uber.org/multierr v1.6.0 // indirect
+	golang.org/x/crypto v0.0.0-20221005025214-4161e89ecf1b // indirect
+	golang.org/x/net v0.4.0 // indirect
+	golang.org/x/text v0.5.0 // indirect
+	google.golang.org/genproto v0.0.0-20220502173005-c8bf987b8c21 // indirect
 	google.golang.org/protobuf v1.28.1 // indirect
-	honnef.co/go/tools v0.2.0 // indirect
+	k8s.io/apimachinery v0.25.0 // indirect
+	k8s.io/klog/v2 v2.80.0 // indirect
 )
 
 // Please use v0.9.3 or later versions instead.

grpcbp/client_middlewares_test.go

@@ -10,6 +10,7 @@ import (
 	pb "github.com/grpc-ecosystem/go-grpc-middleware/testing/testproto"
 	"github.com/opentracing/opentracing-go"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/test/bufconn"
 
 	"github.com/reddit/baseplate.go/mqsend"
@@ -116,7 +117,7 @@ func setupClient(t *testing.T, l *bufconn.Listener, opts ...grpc.DialOption) *gr
 	}
 	opts = append([]grpc.DialOption{
 		grpc.WithContextDialer(bufDialer),
-		grpc.WithInsecure(),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
 	}, opts...)
 
 	// create connection to be used by gRPC client

kafkabp/consumer_test.go

@@ -164,7 +164,7 @@ func setupPartitionConsumers(t *testing.T, kc *consumer) (*mocks.PartitionConsum
 	c := *kc.consumer.Load()
 	mc, ok := c.(*mocks.Consumer)
 	if !ok {
-		t.Fatalf("kc.consumer is not *mocks.Consumer. %#v", kc.consumer)
+		t.Fatalf("kc.consumer is not *mocks.Consumer. %#v", kc.consumer.Load())
 	}
 	pc := mc.ExpectConsumePartition(kc.cfg.Topic, (*kc.partitions.Load())[0], kc.offset)
 	pc1 := mc.ExpectConsumePartition(kc.cfg.Topic, (*kc.partitions.Load())[1], kc.offset)

secrets/secrets.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io"
 	"io/fs"
+	"path/filepath"
 
 	"github.com/reddit/baseplate.go/errorsbp"
 )
@@ -323,19 +324,30 @@ func secretsValidate(secretsDocument Document) (*Secrets, error) {
 
 // walkCSIDirectory parses a directory for vault secrets and merges them into one object
 func walkCSIDirectory(dir fs.FS) (Document, error) {
+	const (
+		// This is where k8s actually writes the content,
+		// ref: https://pkg.go.dev/sigs.k8s.io/secrets-store-csi-driver/pkg/util/fileutil#AtomicWriter.Write
+		k8sSubdirectory = "..data"
+	)
 	secretsDocument := Document{
 		Secrets: make(map[string]GenericSecret),
 	}
 	err := fs.WalkDir(
 		dir,
-		".",
+		k8sSubdirectory,
 		func(path string, d fs.DirEntry, err error) error {
 			if err != nil {
 				return err
 			}
 			if d.IsDir() {
 				return nil
 			}
+			relPath, err := filepath.Rel(k8sSubdirectory, path)
+			if err != nil {
+				// Should not happen as this means path is not under k8sSubdirectory,
+				// but just in case
+				return nil
+			}
 			file, err := dir.Open(path)
 			if err != nil {
 				return err
@@ -347,7 +359,7 @@ func walkCSIDirectory(dir fs.FS) (Document, error) {
 			if err != nil {
 				return fmt.Errorf("decoding %q: %w", path, err)
 			}
-			secretsDocument.Secrets[path] = secretFile.Secret
+			secretsDocument.Secrets[relPath] = secretFile.Secret
 			return nil
 		},
 	)

secrets/store.go

@@ -6,6 +6,7 @@ import (
 	"io"
 	"io/fs"
 	"os"
+	"time"
 
 	"github.com/reddit/baseplate.go/filewatcher"
 	"github.com/reddit/baseplate.go/log"
@@ -41,6 +42,11 @@ type Store struct {
 // Context should come with a timeout otherwise this might block forever, i.e.
 // if the path never becomes available.
 func NewStore(ctx context.Context, path string, logger log.Wrapper, middlewares ...SecretMiddleware) (*Store, error) {
+	return newStoreOverrideFSEventsDelay(ctx, filewatcher.DefaultFSEventsDelay, path, logger, middlewares...)
+}
+
+// Used in tests
+func newStoreOverrideFSEventsDelay(ctx context.Context, fsEventsDelay time.Duration, path string, logger log.Wrapper, middlewares ...SecretMiddleware) (*Store, error) {
 	store := &Store{
 		secretHandlerFunc: nopSecretHandlerFunc,
 	}
@@ -61,6 +67,8 @@ func NewStore(ctx context.Context, path string, logger log.Wrapper, middlewares
 			Path:   path,
 			Parser: parser,
 			Logger: logger,
+
+			FSEventsDelay: fsEventsDelay,
 		},
 	)
 	if err != nil {

secrets/store_internal_test.go

@@ -3,7 +3,11 @@ package secrets
 import (
 	"context"
 	"os"
+	"reflect"
 	"testing"
+	"time"
+
+	"sigs.k8s.io/secrets-store-csi-driver/pkg/util/fileutil"
 
 	"github.com/reddit/baseplate.go/log"
 )
@@ -44,11 +48,7 @@ func TestNewStore(t *testing.T) {
 		},
 	}
 
-	dir, err := os.MkdirTemp("", "secret_test_")
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer os.RemoveAll(dir)
+	dir := t.TempDir()
 
 	for _, tt := range tests {
 		t.Run(
@@ -77,3 +77,88 @@ func TestNewStore(t *testing.T) {
 		)
 	}
 }
+
+func TestDirRotation(t *testing.T) {
+	const (
+		delay = 50 * time.Millisecond
+		sleep = delay * 5
+	)
+	const (
+		key1 = "secrets/foo"
+		key2 = "secrets/bar"
+
+		apiKey = `
+{
+  "request_id": "1afc3036-2282-d483-c2d4-6d483efdf16c",
+  "lease_id": "",
+  "lease_duration": 2764800,
+  "renewable": false,
+  "data": {
+    "type": "simple",
+    "value": "Y2RvVXhNMVdsTXJma3BDaHRGZ0dPYkVGSg==",
+    "encoding": "base64"
+  },
+  "warnings": null
+}
+`
+	)
+	var wantSecret = SimpleSecret{Value: Secret("cdoUxM1WlMrfkpChtFgGObEFJ")}
+
+	dir := t.TempDir()
+	writer, err := fileutil.NewAtomicWriter(dir, "")
+	if err != nil {
+		t.Fatalf("Failed to create k8s atomic writer: %v", err)
+	}
+	content := fileutil.FileProjection{
+		Data: []byte(apiKey),
+		Mode: 0777,
+	}
+	if err := writer.Write(map[string]fileutil.FileProjection{
+		key1: content,
+	}); err != nil {
+		t.Fatalf("Failed to write initial payload: %v", err)
+	}
+
+	store, err := newStoreOverrideFSEventsDelay(context.Background(), delay, dir, log.TestWrapper(t))
+	if err != nil {
+		t.Fatalf("Failed to create secrets store: %v", err)
+	}
+	t.Cleanup(func() { store.Close() })
+
+	t.Run("initial-payload", func(t *testing.T) {
+		_, err := store.GetSimpleSecret(key2)
+		if err == nil {
+			t.Errorf("Expected error when getting %q, got nil", key2)
+		}
+
+		secret, err := store.GetSimpleSecret(key1)
+		if err != nil {
+			t.Fatalf("Expected no error when getting %q, got %v", key1, err)
+		}
+		if !reflect.DeepEqual(secret, wantSecret) {
+			t.Errorf("Got secret %+v, want %+v", secret, wantSecret)
+		}
+	})
+
+	t.Run("rotated-payload", func(t *testing.T) {
+		if err := writer.Write(map[string]fileutil.FileProjection{
+			key2: content,
+		}); err != nil {
+			t.Fatalf("Failed to write rotated payload: %v", err)
+		}
+		time.Sleep(sleep)
+
+		_, err := store.GetSimpleSecret(key1)
+		if err == nil {
+			t.Errorf("Expected error when getting %q, got nil", key1)
+		}
+
+		secret, err := store.GetSimpleSecret(key2)
+		if err != nil {
+			t.Fatalf("Expected no error when getting %q, got %v", key2, err)
+		}
+		if !reflect.DeepEqual(secret, wantSecret) {
+			t.Errorf("Got secret %+v, want %+v", secret, wantSecret)
+		}
+	})
+}