fix(kuma-cp) signing token in multizone (#1007)

Signed-off-by: Jakub Dyszkiewicz <jakub.dyszkiewicz@gmail.com>
# Conflicts:
#	pkg/sds/auth/universal/auth_test.go
#	pkg/tokens/builtin/issuer/issuer.go
#	test/framework/universal_cluster.go

pkg/admin-server/server.go

@@ -189,16 +189,22 @@ func dataplaneTokenWs(rt runtime.Runtime) (*restful.WebService, error) {
 		return nil, nil
 	}
 
-	switch env := rt.Config().Environment; env {
-	case config_core.KubernetesEnvironment:
-		return nil, nil
-	case config_core.UniversalEnvironment:
+	start := true
+	switch rt.Config().Mode {
+	case config_core.Standalone, config_core.Remote:
+		// we still want to generate tokens on Universal even when Global CP is down, so we can scale up and down DPs
+		start = rt.Config().Environment == config_core.UniversalEnvironment
+	case config_core.Global:
+		// the flow may require to generate tokens for Universal's Remote on K8S Global
+		start = true
+	}
+
+	if start {
 		generator, err := builtin.NewDataplaneTokenIssuer(rt)
 		if err != nil {
 			return nil, err
 		}
 		return tokens_server.NewWebservice(generator), nil
-	default:
-		return nil, errors.Errorf("unknown environment type %s", env)
 	}
+	return nil, nil
 }

pkg/core/bootstrap/bootstrap.go

@@ -127,15 +127,19 @@ func onStartup(runtime core_runtime.Runtime) error {
 }
 
 func createDefaultSigningKey(runtime core_runtime.Runtime) error {
-	switch env := runtime.Config().Environment; env {
-	case config_core.KubernetesEnvironment:
-		// we use service account token on K8S, so there is no need for dataplane token server
-		return nil
-	case config_core.UniversalEnvironment:
+	create := false
+	switch runtime.Config().Mode {
+	case config_core.Standalone:
+		create = runtime.Config().Environment == config_core.UniversalEnvironment // Signing Key should be created only on Universal since it is not used on K8S
+	case config_core.Global:
+		create = true // Signing Key with multi-zone should be created on Global even if the Environment is K8S, because we may connect Universal Remote
+	case config_core.Remote:
+		create = false // Signing Key should be synced from Global
+	}
+	if create {
 		return builtin_issuer.CreateDefaultSigningKey(runtime.ResourceManager())
-	default:
-		return errors.Errorf("unknown environment type %s", env)
 	}
+	return nil
 }
 
 func createDefaultMesh(runtime core_runtime.Runtime) error {

pkg/sds/auth/universal/auth_test.go

@@ -21,7 +21,9 @@ import (
 var _ = Describe("Authentication flow", func() {
 	var privateKey = []byte("testPrivateKey")
 
-	issuer := builtin_issuer.NewDataplaneTokenIssuer(privateKey)
+	issuer := builtin_issuer.NewDataplaneTokenIssuer(func() ([]byte, error) {
+		return privateKey, nil
+	})
 	var authenticator auth.Authenticator
 	var resStore store.ResourceStore
 
@@ -149,4 +151,17 @@ var _ = Describe("Authentication flow", func() {
 		// then
 		Expect(err).To(MatchError(`unable to find Dataplane for proxy "default.non-existent-dp": Resource not found: type="Dataplane" name="non-existent-dp" mesh="default"`))
 	})
+
+	It("should throw an error when signing key is not found", func() {
+		// given
+		issuer := builtin_issuer.NewDataplaneTokenIssuer(func() ([]byte, error) {
+			return nil, nil
+		})
+
+		// when
+		_, err := issuer.Generate(xds.ProxyId{})
+
+		// then
+		Expect(err).To(MatchError("there is no Signing Key in the Control Plane. If you run multi-zone setup, make sure Remote is connected to the Global before generating tokens."))
+	})
 })

pkg/tokens/builtin/components.go

@@ -6,9 +6,7 @@ import (
 )
 
 func NewDataplaneTokenIssuer(rt runtime.Runtime) (issuer.DataplaneTokenIssuer, error) {
-	key, err := issuer.GetSigningKey(rt.ResourceManager())
-	if err != nil {
-		return nil, err
-	}
-	return issuer.NewDataplaneTokenIssuer(key), nil
+	return issuer.NewDataplaneTokenIssuer(func() ([]byte, error) {
+		return issuer.GetSigningKey(rt.ReadOnlyResourceManager())
+	}), nil
 }

pkg/tokens/builtin/issuer/issuer.go

@@ -19,36 +19,59 @@ type claims struct {
 	jwt.StandardClaims
 }
 
-func NewDataplaneTokenIssuer(privateKey []byte) DataplaneTokenIssuer {
-	return &jwtTokenIssuer{privateKey}
+type SigningKeyAccessor func() ([]byte, error)
+
+func NewDataplaneTokenIssuer(signingKeyAccessor SigningKeyAccessor) DataplaneTokenIssuer {
+	return &jwtTokenIssuer{signingKeyAccessor}
 }
 
 var _ DataplaneTokenIssuer = &jwtTokenIssuer{}
 
 type jwtTokenIssuer struct {
-	privateKey []byte
+	signingKeyAccessor SigningKeyAccessor
+}
+
+func (i *jwtTokenIssuer) signingKey() ([]byte, error) {
+	signingKey, err := i.signingKeyAccessor()
+	if err != nil {
+		return nil, err
+	}
+	if len(signingKey) == 0 {
+		return nil, SigningKeyNotFound
+	}
+	return signingKey, nil
 }
 
 func (i *jwtTokenIssuer) Generate(proxyId xds.ProxyId) (auth.Credential, error) {
+	signingKey, err := i.signingKey()
+	if err != nil {
+		return "", err
+	}
+
 	c := claims{
 		Name:           proxyId.Name,
 		Mesh:           proxyId.Mesh,
 		StandardClaims: jwt.StandardClaims{},
 	}
 
 	token := jwt.NewWithClaims(jwt.SigningMethodHS256, c)
-	tokenString, err := token.SignedString(i.privateKey)
+	tokenString, err := token.SignedString(signingKey)
 	if err != nil {
 		return "", errors.Wrap(err, "could not sign a token")
 	}
 	return auth.Credential(tokenString), nil
 }
 
 func (i *jwtTokenIssuer) Validate(credential auth.Credential) (xds.ProxyId, error) {
+	signingKey, err := i.signingKey()
+	if err != nil {
+		return xds.ProxyId{}, err
+	}
+
 	c := &claims{}
 
 	token, err := jwt.ParseWithClaims(string(credential), c, func(*jwt.Token) (interface{}, error) {
-		return i.privateKey, nil
+		return signingKey, nil
 	})
 	if err != nil {
 		return xds.ProxyId{}, errors.Wrap(err, "could not parse token")

pkg/tokens/builtin/issuer/signing_key.go

@@ -6,6 +6,7 @@ import (
 	"crypto/rsa"
 	"crypto/x509"
 
+	"github.com/kumahq/kuma/pkg/core"
 	"github.com/kumahq/kuma/pkg/core/resources/manager"
 
 	"github.com/golang/protobuf/ptypes/wrappers"
@@ -17,8 +18,12 @@ import (
 	"github.com/kumahq/kuma/pkg/core/resources/store"
 )
 
+var log = core.Log.WithName("tokens")
+
 const defaultRsaBits = 2048
 
+var SigningKeyNotFound = errors.New("there is no Signing Key in the Control Plane. If you run multi-zone setup, make sure Remote is connected to the Global before generating tokens.")
+
 var signingKeyResourceKey = model.ResourceKey{
 	Mesh: "default",
 	Name: "dataplane-token-signing-key",
@@ -37,6 +42,7 @@ func storeKeyIfNotExist(manager manager.ResourceManager, keyResource system.Secr
 	resource := system.SecretResource{}
 	if err := manager.Get(ctx, &resource, store.GetBy(signingKeyResourceKey)); err != nil {
 		if store.IsResourceNotFound(err) {
+			log.Info("generating signing key for generating dataplane tokens")
 			if err := manager.Create(ctx, &keyResource, store.CreateBy(signingKeyResourceKey)); err != nil {
 				return errors.Wrap(err, "could not store a private key")
 			}
@@ -60,10 +66,12 @@ func createSigningKey() (system.SecretResource, error) {
 	}
 	return res, nil
 }
-
-func GetSigningKey(manager manager.ResourceManager) ([]byte, error) {
+func GetSigningKey(manager manager.ReadOnlyResourceManager) ([]byte, error) {
 	resource := system.SecretResource{}
 	if err := manager.Get(context.Background(), &resource, store.GetBy(signingKeyResourceKey)); err != nil {
+		if store.IsResourceNotFound(err) {
+			return nil, SigningKeyNotFound
+		}
 		return nil, errors.Wrap(err, "could not retrieve signing key from secret manager")
 	}
 	return resource.Spec.GetData().GetValue(), nil

test/e2e/kuma_deploy_hybrid_kube_global_test.go

@@ -0,0 +1,98 @@
+package e2e
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/gruntwork-io/terratest/modules/k8s"
+	"github.com/gruntwork-io/terratest/modules/retry"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"github.com/pkg/errors"
+
+	"github.com/kumahq/kuma/pkg/config/core"
+	. "github.com/kumahq/kuma/test/framework"
+)
+
+var _ = Describe("Test Kubernetes/Universal deployment when Global is on K8S", func() {
+
+	var globalCluster, remoteCluster Cluster
+
+	BeforeEach(func() {
+		k8sClusters, err := NewK8sClusters(
+			[]string{Kuma1},
+			Verbose)
+		Expect(err).ToNot(HaveOccurred())
+
+		universalClusters, err := NewUniversalClusters(
+			[]string{Kuma3},
+			Verbose)
+		Expect(err).ToNot(HaveOccurred())
+
+		// Global
+		globalCluster = k8sClusters.GetCluster(Kuma1)
+		err = NewClusterSetup().
+			Install(Kuma(core.Global)).
+			Setup(globalCluster)
+		Expect(err).ToNot(HaveOccurred())
+		err = globalCluster.VerifyKuma()
+		Expect(err).ToNot(HaveOccurred())
+		globalCP := globalCluster.GetKuma()
+
+		echoServerToken, err := globalCP.GenerateDpToken("echo-server_kuma-test_svc_8080")
+		Expect(err).ToNot(HaveOccurred())
+		demoClientToken, err := globalCP.GenerateDpToken("demo-client")
+		Expect(err).ToNot(HaveOccurred())
+
+		// Remote
+		remoteCluster = universalClusters.GetCluster(Kuma3)
+		err = NewClusterSetup().
+			Install(Kuma(core.Remote, WithGlobalAddress(globalCP.GetKDSServerAddress()))).
+			Install(EchoServerUniversal(echoServerToken)).
+			Install(DemoClientUniversal(demoClientToken)).
+			Setup(remoteCluster)
+		Expect(err).ToNot(HaveOccurred())
+		err = remoteCluster.VerifyKuma()
+		Expect(err).ToNot(HaveOccurred())
+
+		// connect Remote with Global
+		err = k8s.KubectlApplyFromStringE(globalCluster.GetTesting(), globalCluster.GetKubectlOptions(),
+			fmt.Sprintf(ZoneTemplateK8s,
+				Kuma3,
+				remoteCluster.GetKuma().GetIngressAddress()))
+		Expect(err).ToNot(HaveOccurred())
+	})
+
+	AfterEach(func() {
+		err := globalCluster.DeleteKuma()
+		Expect(err).ToNot(HaveOccurred())
+		err = globalCluster.DismissCluster()
+		Expect(err).ToNot(HaveOccurred())
+
+		err = remoteCluster.DeleteKuma()
+		Expect(err).ToNot(HaveOccurred())
+		err = remoteCluster.DismissCluster()
+		Expect(err).ToNot(HaveOccurred())
+	})
+
+	It("communication in between apps in remote zone works", func() {
+		stdout, _, err := remoteCluster.ExecWithRetries("", "", "demo-client",
+			"curl", "-v", "-m", "3", "localhost:4001")
+		Expect(err).ToNot(HaveOccurred())
+		Expect(stdout).To(ContainSubstring("HTTP/1.1 200 OK"))
+
+		retry.DoWithRetry(remoteCluster.GetTesting(), "curl remote service",
+			DefaultRetries, DefaultTimeout,
+			func() (string, error) {
+				stdout, _, err = remoteCluster.ExecWithRetries("", "", "demo-client",
+					"curl", "-v", "-m", "3", "localhost:4001")
+				if err != nil {
+					return "should retry", err
+				}
+				if strings.Contains(stdout, "HTTP/1.1 200 OK") {
+					return "Accessing service successful", nil
+				}
+				return "should retry", errors.Errorf("should retry")
+			})
+	})
+})

test/e2e/kuma_deploy_hybrid_test.go

@@ -85,6 +85,11 @@ metadata:
 
 		globalCP := global.GetKuma()
 
+		echoServerToken, err := globalCP.GenerateDpToken("echo-server_kuma-test_svc_8080")
+		Expect(err).ToNot(HaveOccurred())
+		demoClientToken, err := globalCP.GenerateDpToken("demo-client")
+		Expect(err).ToNot(HaveOccurred())
+
 		// K8s Cluster 1
 		remote_1 = k8sClusters.GetCluster(Kuma1)
 
@@ -119,8 +124,8 @@ metadata:
 
 		err = NewClusterSetup().
 			Install(Kuma(core.Remote, WithGlobalAddress(globalCP.GetKDSServerAddress()))).
-			Install(EchoServerUniversal()).
-			Install(DemoClientUniversal()).
+			Install(EchoServerUniversal(echoServerToken)).
+			Install(DemoClientUniversal(demoClientToken)).
 			Setup(remote_3)
 		Expect(err).ToNot(HaveOccurred())
 		err = remote_3.VerifyKuma()
@@ -131,7 +136,7 @@ metadata:
 
 		err = NewClusterSetup().
 			Install(Kuma(core.Remote, WithGlobalAddress(globalCP.GetKDSServerAddress()))).
-			Install(DemoClientUniversal()).
+			Install(DemoClientUniversal(demoClientToken)).
 			Setup(remote_4)
 		Expect(err).ToNot(HaveOccurred())
 		err = remote_4.VerifyKuma()

test/e2e/kuma_deploy_universal_test.go

@@ -56,13 +56,18 @@ destinations:
 
 		globalCP := global.GetKuma()
 
+		echoServerToken, err := globalCP.GenerateDpToken("echo-server_kuma-test_svc_8080")
+		Expect(err).ToNot(HaveOccurred())
+		demoClientToken, err := globalCP.GenerateDpToken("demo-client")
+		Expect(err).ToNot(HaveOccurred())
+
 		// Cluster 1
 		remote_1 = clusters.GetCluster(Kuma2)
 
 		err = NewClusterSetup().
 			Install(Kuma(core.Remote, WithGlobalAddress(globalCP.GetKDSServerAddress()))).
-			Install(EchoServerUniversal()).
-			Install(DemoClientUniversal()).
+			Install(EchoServerUniversal(echoServerToken)).
+			Install(DemoClientUniversal(demoClientToken)).
 			Setup(remote_1)
 		Expect(err).ToNot(HaveOccurred())
 		err = remote_1.VerifyKuma()
@@ -73,7 +78,7 @@ destinations:
 
 		err = NewClusterSetup().
 			Install(Kuma(core.Remote, WithGlobalAddress(globalCP.GetKDSServerAddress()))).
-			Install(DemoClientUniversal()).
+			Install(DemoClientUniversal(demoClientToken)).
 			Setup(remote_2)
 		Expect(err).ToNot(HaveOccurred())
 		err = remote_2.VerifyKuma()

test/e2e/tracing_universal_test.go

@@ -46,13 +46,21 @@ selectors:
 
 		err := NewClusterSetup().
 			Install(Kuma(core.Standalone)).
-			Install(EchoServerUniversal()).
-			Install(DemoClientUniversal()).
 			Install(tracing.Install()).
 			Setup(cluster)
 		Expect(err).ToNot(HaveOccurred())
 		err = cluster.VerifyKuma()
 		Expect(err).ToNot(HaveOccurred())
+
+		echoServerToken, err := cluster.GetKuma().GenerateDpToken("echo-server_kuma-test_svc_8080")
+		Expect(err).ToNot(HaveOccurred())
+		demoClientToken, err := cluster.GetKuma().GenerateDpToken("demo-client")
+		Expect(err).ToNot(HaveOccurred())
+
+		err = EchoServerUniversal(echoServerToken)(cluster)
+		Expect(err).ToNot(HaveOccurred())
+		err = DemoClientUniversal(demoClientToken)(cluster)
+		Expect(err).ToNot(HaveOccurred())
 	})
 
 	AfterEach(func() {

test/framework/constants.go

@@ -63,5 +63,6 @@ const (
 	helmChartPath = "../../deployments/charts/kuma"
 
 	kumaCPAPIPort        = 5681
+	kumaCPAdminPort      = 5679
 	kumaCPAPIPortFwdBase = 32000 + kumaCPAPIPort
 )