tests(cmd/influxd/launcher): add a launcher test for the secret service (#14813)

The secret service is tested by creating a secret and then attempting to
use it in a flux query. There is one test where accessing the secret
should work and one where it should return that the action is forbidden.

cmd/influxd/launcher/launcher.go

@@ -721,7 +721,7 @@ func (m *Launcher) BucketService() platform.BucketService {
 	return m.apibackend.BucketService
 }
 
-// UserService returns the internal suser service.
+// UserService returns the internal user service.
 func (m *Launcher) UserService() platform.UserService {
 	return m.apibackend.UserService
 }
@@ -731,6 +731,11 @@ func (m *Launcher) AuthorizationService() platform.AuthorizationService {
 	return m.apibackend.AuthorizationService
 }
 
+// SecretService returns the internal secret service.
+func (m *Launcher) SecretService() platform.SecretService {
+	return m.apibackend.SecretService
+}
+
 // TaskService returns the internal task service.
 func (m *Launcher) TaskService() platform.TaskService {
 	return m.apibackend.TaskService

cmd/influxd/launcher/query_test.go

@@ -3,6 +3,7 @@ package launcher_test
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	nethttp "net/http"
@@ -11,7 +12,10 @@ import (
 	"time"
 
 	"github.com/influxdata/flux"
+	"github.com/influxdata/flux/execute"
 	"github.com/influxdata/flux/lang"
+	"github.com/influxdata/flux/values"
+	"github.com/influxdata/influxdb"
 	"github.com/influxdata/influxdb/cmd/influxd/launcher"
 	phttp "github.com/influxdata/influxdb/http"
 	"github.com/influxdata/influxdb/query"
@@ -147,3 +151,107 @@ func TestPipeline_QueryMemoryLimits(t *testing.T) {
 		t.Fatal("expected error, got successful query execution")
 	}
 }
+
+func TestPipeline_Query_LoadSecret_Success(t *testing.T) {
+	l := launcher.RunTestLauncherOrFail(t, ctx)
+	l.SetupOrFail(t)
+	defer l.ShutdownOrFail(t, ctx)
+
+	const key, value = "mytoken", "secrettoken"
+	if err := l.SecretService().PutSecret(ctx, l.Org.ID, key, value); err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+
+	// write one point so we can use it
+	l.WritePointsOrFail(t, fmt.Sprintf(`m,k=v1 f=%di %d`, 0, time.Now().UnixNano()))
+
+	// we expect this request to succeed
+	req := &query.Request{
+		Authorization:  l.Auth,
+		OrganizationID: l.Org.ID,
+		Compiler: lang.FluxCompiler{
+			Query: fmt.Sprintf(`
+import "influxdata/influxdb/secrets"
+
+token = secrets.get(key: "mytoken")
+from(bucket: "%s")
+	|> range(start: -5m)
+	|> set(key: "token", value: token)
+`, l.Bucket.Name),
+		},
+	}
+	if err := l.QueryAndConsume(ctx, req, func(r flux.Result) error {
+		return r.Tables().Do(func(tbl flux.Table) error {
+			return tbl.Do(func(cr flux.ColReader) error {
+				j := execute.ColIdx("token", cr.Cols())
+				if j == -1 {
+					return errors.New("cannot find table column \"token\"")
+				}
+
+				for i := 0; i < cr.Len(); i++ {
+					v := execute.ValueForRow(cr, i, j)
+					if got, want := v, values.NewString("secrettoken"); !got.Equal(want) {
+						t.Errorf("unexpected value at row %d -want/+got:\n\t- %v\n\t+ %v", i, got, want)
+					}
+				}
+				return nil
+			})
+		})
+	}); err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+}
+
+func TestPipeline_Query_LoadSecret_Forbidden(t *testing.T) {
+	l := launcher.RunTestLauncherOrFail(t, ctx)
+	l.SetupOrFail(t)
+	defer l.ShutdownOrFail(t, ctx)
+
+	const key, value = "mytoken", "secrettoken"
+	if err := l.SecretService().PutSecret(ctx, l.Org.ID, key, value); err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+
+	// write one point so we can use it
+	l.WritePointsOrFail(t, fmt.Sprintf(`m,k=v1 f=%di %d`, 0, time.Now().UnixNano()))
+
+	auth := &influxdb.Authorization{
+		OrgID:  l.Org.ID,
+		UserID: l.User.ID,
+		Permissions: []influxdb.Permission{
+			{
+				Action: influxdb.ReadAction,
+				Resource: influxdb.Resource{
+					Type:  influxdb.BucketsResourceType,
+					ID:    &l.Bucket.ID,
+					OrgID: &l.Org.ID,
+				},
+			},
+		},
+	}
+	if err := l.AuthorizationService().CreateAuthorization(ctx, auth); err != nil {
+		t.Fatalf("unexpected error creating authorization: %s", err)
+	}
+	l.Auth = auth
+
+	// we expect this request to succeed
+	req := &query.Request{
+		Authorization:  l.Auth,
+		OrganizationID: l.Org.ID,
+		Compiler: lang.FluxCompiler{
+			Query: fmt.Sprintf(`
+import "influxdata/influxdb/secrets"
+
+token = secrets.get(key: "mytoken")
+from(bucket: "%s")
+	|> range(start: -5m)
+	|> set(key: "token", value: token)
+`, l.Bucket.Name),
+		},
+	}
+	if err := l.QueryAndNopConsume(ctx, req); err == nil {
+		t.Error("expected error")
+	} else if got, want := influxdb.ErrorCode(err), influxdb.EUnauthorized; got != want {
+		t.Errorf("unexpected error code -want/+got:\n\t- %v\n\t+ %v", got, want)
+	}
+}

query/control/controller.go

@@ -921,9 +921,18 @@ func handleFluxError(err error) error {
 	if !ok {
 		return err
 	}
+	werr := handleFluxError(ferr.Err)
 
 	code := influxdb.EInternal
-	switch flux.ErrorCode(err) {
+	switch ferr.Code {
+	case codes.Inherit:
+		// If we are inheriting the error code, influxdb doesn't
+		// have an equivalent of this so we need to retrieve
+		// the error code from the wrapped error which has already
+		// been translated to an influxdb error (if possible).
+		if werr != nil {
+			code = influxdb.ErrorCode(werr)
+		}
 	case codes.NotFound:
 		code = influxdb.ENotFound
 	case codes.Invalid:
@@ -950,6 +959,6 @@ func handleFluxError(err error) error {
 	return &influxdb.Error{
 		Code: code,
 		Msg:  ferr.Msg,
-		Err:  handleFluxError(ferr.Err),
+		Err:  werr,
 	}
 }