explicit Dapr API authentication token support (dapr#73)

* explicit Dapr API authentication token

* avoiding race condition on api token

* allows for env var override

Readme.md

@@ -53,6 +53,7 @@ Check the [example folder](./example) for working Dapr go client examples.
 
 The Dapr go client supports following functionality: 
 
+
 ##### State 
 
 For simple use-cases, Dapr client provides easy to use methods for `Save`, `Get`, and `Delete`: 
@@ -211,6 +212,30 @@ opt := map[string]string{
 secret, err := client.GetSecret(ctx, "store-name", "secret-name", opt)
 ```
 
+
+#### Authentication 
+
+By default, Dapr relies on the network boundary to limit access to its API. If however the target Dapr API is configured with token-based authentication, users can configure the go Dapr client with that token in two ways:
+
+##### Environment Variable 
+
+If the `DAPR_API_TOKEN` environment variable is defined, Dapr will automatically use it to augment its Dapr API invocations to ensure authentication. 
+
+##### Explicit Method 
+
+In addition, users can also set the API token explicitly on any Dapr client instance. This approach is helpful in cases when the user code needs to create multiple clients for different Dapr API endpoints. 
+
+```go
+func main() {
+    client, err := dapr.NewClient()
+    if err != nil {
+        panic(err)
+    }
+    defer client.Close()
+    client.WithAuthToken("your-Dapr-API-token-here")
+}
+```
+
 ## Service (callback)
 
 In addition to this Dapr API client, Dapr go SDK also provides `service` package to bootstrap your Dapr callback services in either gRPC or HTTP. Instructions on how to use it are located [here](./service/Readme.md)

client/binding.go

@@ -47,7 +47,7 @@ func (c *GRPCClient) InvokeBinding(ctx context.Context, in *BindingInvocation) (
 		Metadata:  in.Metadata,
 	}
 
-	resp, err := c.protoClient.InvokeBinding(authContext(ctx), req)
+	resp, err := c.protoClient.InvokeBinding(c.withAuthToken(ctx), req)
 	if err != nil {
 		return nil, errors.Wrapf(err, "error invoking binding %s/%s", in.Name, in.Operation)
 	}

client/client.go

@@ -75,9 +75,12 @@ type Client interface {
 	// ExecuteStateTransaction provides way to execute multiple operations on a specified store.
 	ExecuteStateTransaction(ctx context.Context, store string, meta map[string]string, ops []*StateOperation) error
 
-	// WithTraceID adds existing trace ID to the outgoing context
+	// WithTraceID adds existing trace ID to the outgoing context.
 	WithTraceID(ctx context.Context, id string) context.Context
 
+	// WithAuthToken sets Dapr API token on the instantiated client.
+	WithAuthToken(token string)
+
 	// Close cleans up all resources created by the client.
 	Close()
 }
@@ -133,13 +136,16 @@ func NewClientWithConnection(conn *grpc.ClientConn) Client {
 	return &GRPCClient{
 		connection:  conn,
 		protoClient: pb.NewDaprClient(conn),
+		authToken:   os.Getenv(apiTokenEnvVarName),
 	}
 }
 
 // GRPCClient is the gRPC implementation of Dapr client.
 type GRPCClient struct {
 	connection  *grpc.ClientConn
 	protoClient pb.DaprClient
+	authToken   string
+	mux         sync.Mutex
 }
 
 // Close cleans up all resources created by the client.
@@ -149,6 +155,14 @@ func (c *GRPCClient) Close() {
 	}
 }
 
+// WithAuthToken sets Dapr API token on the instantiated client.
+// Allows empty string to reset token on existing client
+func (c *GRPCClient) WithAuthToken(token string) {
+	c.mux.Lock()
+	c.authToken = token
+	c.mux.Unlock()
+}
+
 // WithTraceID adds existing trace ID to the outgoing context
 func (c *GRPCClient) WithTraceID(ctx context.Context, id string) context.Context {
 	if id == "" {
@@ -159,11 +173,9 @@ func (c *GRPCClient) WithTraceID(ctx context.Context, id string) context.Context
 	return metadata.NewOutgoingContext(ctx, md)
 }
 
-func authContext(ctx context.Context) context.Context {
-	token := os.Getenv(apiTokenEnvVarName)
-	if token == "" {
+func (c *GRPCClient) withAuthToken(ctx context.Context) context.Context {
+	if c.authToken == "" {
 		return ctx
 	}
-	md := metadata.Pairs(apiTokenKey, token)
-	return metadata.NewOutgoingContext(ctx, md)
+	return metadata.NewOutgoingContext(ctx, metadata.Pairs(apiTokenKey, string(c.authToken)))
 }

client/client_test.go

@@ -35,11 +35,23 @@ func TestMain(m *testing.M) {
 	os.Exit(r)
 }
 
-func TestNewClientWithoutArgs(t *testing.T) {
-	_, err := NewClientWithPort("")
-	assert.NotNil(t, err)
-	_, err = NewClientWithAddress("")
-	assert.NotNil(t, err)
+func TestNewClient(t *testing.T) {
+	t.Run("no arg for with port", func(t *testing.T) {
+		_, err := NewClientWithPort("")
+		assert.Error(t, err)
+	})
+
+	t.Run("no arg for with address", func(t *testing.T) {
+		_, err := NewClientWithAddress("")
+		assert.Error(t, err)
+	})
+
+	t.Run("new client closed with empty token", func(t *testing.T) {
+		c, err := NewClient()
+		assert.NoError(t, err)
+		c.WithAuthToken("")
+		c.Close()
+	})
 }
 
 func getTestClient(ctx context.Context) (client Client, closer func()) {

client/invoke.go

@@ -22,7 +22,7 @@ func (c *GRPCClient) invokeServiceWithRequest(ctx context.Context, req *pb.Invok
 		return nil, errors.New("nil request")
 	}
 
-	resp, err := c.protoClient.InvokeService(authContext(ctx), req)
+	resp, err := c.protoClient.InvokeService(c.withAuthToken(ctx), req)
 	if err != nil {
 		return nil, errors.Wrap(err, "error invoking service")
 	}

client/pubsub.go

@@ -22,7 +22,7 @@ func (c *GRPCClient) PublishEvent(ctx context.Context, component, topic string,
 		Data:       in,
 	}
 
-	_, err := c.protoClient.PublishEvent(authContext(ctx), envelop)
+	_, err := c.protoClient.PublishEvent(c.withAuthToken(ctx), envelop)
 	if err != nil {
 		return errors.Wrapf(err, "error publishing event unto %s topic", topic)
 	}

client/secret.go

@@ -22,7 +22,7 @@ func (c *GRPCClient) GetSecret(ctx context.Context, store, key string, meta map[
 		Metadata:  meta,
 	}
 
-	resp, err := c.protoClient.GetSecret(authContext(ctx), req)
+	resp, err := c.protoClient.GetSecret(c.withAuthToken(ctx), req)
 	if err != nil {
 		return nil, errors.Wrap(err, "error invoking service")
 	}

client/state.go

@@ -172,7 +172,7 @@ func (c *GRPCClient) ExecuteStateTransaction(ctx context.Context, store string,
 		StoreName:  store,
 		Operations: items,
 	}
-	_, err := c.protoClient.ExecuteStateTransaction(authContext(ctx), req)
+	_, err := c.protoClient.ExecuteStateTransaction(c.withAuthToken(ctx), req)
 	if err != nil {
 		return errors.Wrap(err, "error executing state transaction")
 	}
@@ -204,7 +204,7 @@ func (c *GRPCClient) SaveStateItems(ctx context.Context, store string, items ...
 		req.States = append(req.States, item)
 	}
 
-	_, err := c.protoClient.SaveState(authContext(ctx), req)
+	_, err := c.protoClient.SaveState(c.withAuthToken(ctx), req)
 	if err != nil {
 		return errors.Wrap(err, "error saving state")
 	}
@@ -228,7 +228,7 @@ func (c *GRPCClient) GetBulkItems(ctx context.Context, store string, keys []stri
 		Parallelism: parallelism,
 	}
 
-	results, err := c.protoClient.GetBulkState(authContext(ctx), req)
+	results, err := c.protoClient.GetBulkState(c.withAuthToken(ctx), req)
 	if err != nil {
 		return nil, errors.Wrap(err, "error getting state")
 	}
@@ -269,7 +269,7 @@ func (c *GRPCClient) GetStateWithConsistency(ctx context.Context, store, key str
 		Consistency: (v1.StateOptions_StateConsistency(sc)),
 	}
 
-	result, err := c.protoClient.GetState(authContext(ctx), req)
+	result, err := c.protoClient.GetState(c.withAuthToken(ctx), req)
 	if err != nil {
 		return nil, errors.Wrap(err, "error getting state")
 	}
@@ -302,7 +302,7 @@ func (c *GRPCClient) DeleteStateWithETag(ctx context.Context, store, key, etag s
 		Options:   toProtoStateOptions(opts),
 	}
 
-	_, err := c.protoClient.DeleteState(authContext(ctx), req)
+	_, err := c.protoClient.DeleteState(c.withAuthToken(ctx), req)
 	if err != nil {
 		return errors.Wrap(err, "error deleting state")
 	}